source: trunk/src/main/java/omq/common/event/EventDispatcher.java @ 54

package omq.common.event;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;

import omq.common.broker.Broker;
import omq.common.util.ParameterQueue;
import omq.common.util.Serializer;

import org.apache.log4j.Logger;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * This class dispatches the events received in the client side and stores them
 * into the different listeners that could exists among the different proxies
 * generated
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
@SuppressWarnings("rawtypes")
public class EventDispatcher extends Thread {
        private static final Logger logger = Logger.getLogger(EventDispatcher.class.getName());

        private Broker broker;
        private Serializer serializer;
        private Map<String, Vector<EventListener>> listeners;
        private Channel channel;
        private QueueingConsumer consumer;
        private Properties env;
        private boolean killed = false;

        public EventDispatcher(Broker broker) throws Exception {
                this.broker = broker;
                env = broker.getEnvironment();
                serializer = broker.getSerializer();

                // Declare the listeners map
                listeners = new HashMap<String, Vector<EventListener>>();

                startEventQueue();
        }

        private void startEventQueue() throws Exception {
                // Get a new connection and a new channel
                channel = broker.getNewChannel();

                String event_queue = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE);
                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
                channel.queueDeclare(event_queue, durable, false, false, null);
                logger.info("EventDispatcher creating queue: " + event_queue + ", durable: " + durable);

                // Declare a new consumer
                consumer = new QueueingConsumer(channel);
                channel.basicConsume(event_queue, true, consumer);
        }

        public void kill() throws Exception {
                logger.warn("Stopping EventDispatcher");
                setListeners(null);
                killed = true;
                interrupt();
                channel.close();
        }

        @Override
        public void run() {
                logger.info("EventDispatcher started");
                Delivery delivery;
                Event event;

                while (!killed) {
                        try {
                                // Get the delivery
                                delivery = consumer.nextDelivery();

                                // Get the event
                                event = serializer.deserializeEvent(delivery.getBody());

                                logger.info("Event received -> Topic: " + event.getTopic() + "CorrId: " + event.getCorrId());

                                // Dispatch it
                                dispatch(event.getTopic(), event);
                        } catch (InterruptedException i) {
                                logger.error(i);
                        } catch (ShutdownSignalException e) {
                                logger.error(e);
                                try {
                                        if (channel.isOpen()) {
                                                channel.close();
                                        }
                                        startEventQueue();
                                } catch (Exception e1) {
                                        try {
                                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
                                                Thread.sleep(milis);
                                        } catch (InterruptedException e2) {
                                                logger.error(e2);
                                        }
                                        logger.error(e1);
                                }
                        } catch (ConsumerCancelledException e) {
                                logger.error(e);
                        } catch (Exception e) {
                                logger.error(e);
                        }
                }
        }

        public int addListener(EventListener e) throws Exception {
                // Map<String, ArrayList<EventListener<Event>>> mListeners =
                // listeners.get(e.getTopic());
                // if(mListeners == null){
                // mListeners = new HashMap<String, ArrayList<EventListener<Event>>>();
                //
                // String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE);
                // String reference = e.getTopic();
                // channel.exchangeDeclare(reference, "fanout");
                // channel.queueBind(queueName, reference, "");
                // }

                Vector<EventListener> vListeners = listeners.get(e.getTopic());
                if (vListeners == null) {
                        vListeners = new Vector<EventListener>();

                        String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE);
                        String reference = e.getTopic();

                        logger.info("Declaring fanout -> " + reference + " Binding with: " + queueName);

                        channel.exchangeDeclare(reference, "fanout");
                        channel.queueBind(queueName, reference, "");
                }
                vListeners.add(e);
                listeners.put(e.getTopic(), vListeners);

                return vListeners.size();
        }

        public int removeListener(EventListener e) {
                Vector<EventListener> vListeners = listeners.get(e.getTopic());
                if (vListeners != null) {
                        // TODO: removeListener -> remove(e) override equals?
                        vListeners.remove(e);
                }

                return vListeners.size();
        }

        /**
         * This method dispatches the events. Every time an event is received, this
         * method is launched. This method creates a new thread and executes the
         * notifyEvent function of the listeners associated to this event
         * 
         * @param topic
         * @param event
         */
        public void dispatch(String topic, final Event event) {
                if (listeners.containsKey(topic)) {
                        for (final EventListener listener : listeners.get(topic)) {
                                new Thread() {
                                        @SuppressWarnings("unchecked")
                                        public void run() {
                                                listener.notifyEvent(event);
                                        }
                                }.start();
                        }
                }
        }

        public Map<String, Vector<EventListener>> getListeners() {
                return listeners;
        }

        public void setListeners(Map<String, Vector<EventListener>> listeners) {
                this.listeners = listeners;
        }

}