/****************************************************************************** * Copyright (C) 2012 Heng Sin Low * * Copyright (C) 2012 Trek Global * * This program is free software; you can redistribute it and/or modify it * * under the terms version 2 of the GNU General Public License as published * * by the Free Software Foundation. This program is distributed in the hope * * that it will be useful, but WITHOUT ANY WARRANTY; without even the implied * * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. * * See the GNU General Public License for more details. * * You should have received a copy of the GNU General Public License along * * with this program; if not, write to the Free Software Foundation, Inc., * * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. * *****************************************************************************/ package org.idempiere.hazelcast.service; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; import org.idempiere.distributed.ITopic; import org.idempiere.distributed.ITopicSubscriber; import com.hazelcast.topic.Message; import com.hazelcast.topic.MessageListener; /** * @author hengsin * */ public class TopicImpl implements ITopic { private com.hazelcast.topic.ITopic topic; private List> adapters; private Map, String> registrationMap; /** * */ public TopicImpl(com.hazelcast.topic.ITopic topic) { this.topic = topic; adapters = new ArrayList>(); registrationMap = new HashMap<>(); } @Override public String getName() { return topic.getName(); } @Override public void subscribe(final ITopicSubscriber subscriber) { TopicSubscriberAdapter adapter = new TopicSubscriberAdapter(subscriber); String registrationId = topic.addMessageListener(adapter).toString(); adapters.add(adapter); registrationMap.put(adapter, registrationId); } @Override public void unsubscribe(ITopicSubscriber subscriber) { TopicSubscriberAdapter found = null; for(TopicSubscriberAdapter adapter : adapters) { if (adapter.subscriber == subscriber) { found = adapter; String registrationId = registrationMap.get(adapter); if (topic.removeMessageListener(UUID.fromString(registrationId))) registrationMap.remove(adapter); break; } } if (found != null) adapters.remove(found); } @Override public void publish(E message) { topic.publish(message); } class TopicSubscriberAdapter implements MessageListener { protected ITopicSubscriber subscriber; protected TopicSubscriberAdapter(ITopicSubscriber subscriber) { this.subscriber = subscriber; } @Override public void onMessage(Message message) { subscriber.onMessage(message.getMessageObject()); } } }