source: trunk/objectmq/src/omq/client/remote/response/ResponseListener.java @ 9

package omq.client.remote.response;

import java.util.Hashtable;
import java.util.Map;
import java.util.Properties;

import omq.client.proxy.Proxymq;
import omq.common.remote.RemoteListener;
import omq.common.util.ParameterQueue;

import com.rabbitmq.client.AMQP.BasicProperties;
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 RemoteListener {
        private static ResponseListener rListener;
        private Map<String, Map<String, byte[]>> results;

        /**
         * Protected constructor used by the singleton pattern
         * 
         * @param env
         * @throws Exception
         */
        protected ResponseListener(Properties env) throws Exception {
                super(env);
                // Init the hashtable (it's concurrent)
                this.results = new Hashtable<String, Map<String, byte[]>>();

                String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
                channel.queueDeclare(reply_queue, false, false, false, null);

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

        @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();

                                // 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) {
                                i.printStackTrace();
                        } catch (ShutdownSignalException e) {
                                e.printStackTrace();
                        } catch (ConsumerCancelledException e) {
                                e.printStackTrace();
                        } catch (Exception e) {
                                e.printStackTrace();
                        }
                }
        }

        /**
         * 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;
        }

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