source: trunk/src/main/java/omq/client/listener/ResponseListener.java @ 50

package omq.client.listener;

import java.io.IOException;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Properties;

import org.apache.log4j.Logger;

import omq.client.proxy.Proxymq;
import omq.common.broker.Broker;
import omq.common.util.ParameterQueue;

import com.rabbitmq.client.AMQP.BasicProperties;
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;

/**
 * Class that inherits from RemoteListener. It's used in the server side. This
 * class gets the deliveries from the server and stores them into the proxies
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class ResponseListener extends Thread {
        private static final Logger logger = Logger.getLogger(ResponseListener.class.getName());
        private static ResponseListener rListener;

        private Channel channel;
        private QueueingConsumer consumer;
        private boolean killed = false;
        private Map<String, Map<String, byte[]>> results;
        private Properties env;

        /**
         * Protected constructor used by the singleton pattern
         * 
         * @param env
         * @throws Exception
         */
        protected ResponseListener(Properties env) throws Exception {
                this.env = env;

                // Init the hashtable (it's concurrent)
                this.results = new Hashtable<String, Map<String, byte[]>>();

                startRPCQueue();
        }

        @Override
        public void run() {
                Delivery delivery;
                String uid_request;

                while (!killed) {
                        try {
                                // Get the delivery

                                delivery = consumer.nextDelivery();

                                BasicProperties props = delivery.getProperties();

                                // Get the response with its uid
                                uid_request = delivery.getProperties().getCorrelationId();
                                logger.debug("Response received -> proxy reference: " + props.getAppId() + ", corrId: " + uid_request);

                                // Stores the new response
                                Map<String, byte[]> proxyResults = results.get(props.getAppId());

                                // Put the result into the proxy results and notify him
                                synchronized (proxyResults) {
                                        // If we haven't received this response before, we store it
                                        if (!proxyResults.containsKey(uid_request)) {
                                                proxyResults.put(uid_request, delivery.getBody());
                                                proxyResults.notifyAll();
                                        }
                                }
                        } catch (InterruptedException i) {
                                logger.error(i.toString(), i);
                        } catch (ShutdownSignalException e) {
                                logger.error(e.toString(), e);
                                try {
                                        if (channel.isOpen()) {
                                                channel.close();
                                        }
                                        startRPCQueue();
                                } catch (Exception e1) {
                                        e1.printStackTrace();
                                        try {
                                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
                                                Thread.sleep(milis);
                                        } catch (InterruptedException e2) {
                                                logger.error(e2.toString(), e2);
                                        }
                                }
                        } catch (ConsumerCancelledException e) {
                                logger.error(e.toString(), e);
                        } catch (Exception e) {
                                logger.error(e.toString(), e);
                        }
                }
        }

        private void startRPCQueue() throws Exception {
                channel = Broker.getNewChannel();

                Map<String, Object> args = null;

                String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));

                int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1"));
                if (ttl > 0) {
                        args = new HashMap<String, Object>();
                        args.put("x-message-ttl", ttl);
                }

                channel.queueDeclare(reply_queue, durable, false, false, args);

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

        /**
         * Static function which initializes the ResponseListener
         * 
         * @param env
         * @throws Exception
         */
        public static void init(Properties env) throws Exception {
                if (rListener == null) {
                        rListener = new ResponseListener(env);
                        rListener.start();
                } else {
                        throw new Exception("Cannot init because it already exists");
                }
        }

        /**
         * Method to retrieve the unique ResponseListener, this function can also
         * initialize a ResponseListener using and environment
         * 
         * @param env
         * @return unique ResponseListener
         * @throws Exception
         */
        public static ResponseListener getRequestListener(Properties env) throws Exception {
                if (rListener == null) {
                        rListener = new ResponseListener(env);
                        rListener.start();
                } else {
                        // TODO: create a new exception to indicate that a response listener
                        // cannot be init
                        throw new Exception("Cannot init because it already exists");
                }
                return rListener;
        }

        public static boolean isVoid() {
                return rListener == null;
        }

        /**
         * Method to retrieve the unique ResponseListener
         * 
         * @return
         * @throws Exception
         */
        public static ResponseListener getRequestListener() throws Exception {
                if (rListener == null) {
                        throw new Exception("Request listener not initialized");
                }
                return rListener;
        }

        /**
         * 
         * @param key
         * @return whether the map has the param key
         */
        public boolean containsKey(String key) {
                return results.containsKey(key);
        }

        /**
         * This method is used to kill the unique responseListener in the system
         * 
         * @throws Exception
         */
        public static void stopResponseListner() throws Exception {
                rListener.kill();
                rListener = null;
        }

        /**
         * Interrupt and kill the Thread
         * 
         * @throws IOException
         */
        public void kill() throws IOException {
                logger.warn("Killing ResponseListener");
                interrupt();
                killed = true;
                channel.close();
        }

        // Revisar això
        public void registerProxy(Proxymq proxy) {
                if (!results.containsKey(proxy.getRef())) {
                        results.put(proxy.getRef(), proxy.getResults());
                }
        }
}