source: branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java

package omq.server;

import omq.common.util.ParameterQueue;
import omq.exception.SerializerException;

import org.apache.log4j.Logger;

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

public class MultiInvocationThread extends AInvocationThread {

        private static final Logger logger = Logger.getLogger(MultiInvocationThread.class.getName());
        private static final String multi = "multi#";

        // Consumer
        private String multiQueue;

        public MultiInvocationThread(RemoteObject obj) throws Exception {
                super(obj);
        }

        @Override
        public void run() {
                while (!killed) {
                        try {
                                // Get the delivery
                                Delivery delivery = consumer.nextDelivery();
                                // This thread does not need to set busy because it's mandatory
                                // to exist
                                executeTask(delivery);
                        } catch (InterruptedException i) {
                                logger.error(i);
                        } catch (ShutdownSignalException e) {
                                logger.error(e);
                                try {
                                        if (channel.isOpen()) {
                                                channel.close();
                                        }
                                        startQueues();
                                } 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 (SerializerException e) {
                                logger.error(e);
                        } catch (Exception e) {
                                e.printStackTrace();
                                logger.error(e);
                        }

                }
                logger.info("ObjectMQ ('" + obj.getRef() + "') InvocationThread " + Thread.currentThread().getId() + " is killed");
        }

        @Override
        protected void startQueues() throws Exception {
                // Start channel
                channel = broker.getNewChannel();

                /*
                 * Multi queue, exclusive per each instance
                 */

                // Get info about the multiQueue
                String multiExchange = multi + reference;
                // TODO:String multiExchange = multi + exchange + reference;
                multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME);

                // Multi queue (exclusive queue per remoteObject)
                boolean multiDurable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_MQUEUE, "false"));
                boolean multiExclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_MQUEUE, "true"));
                boolean multiAutoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_MQUEUE, "true"));

                // Declares and bindings
                channel.exchangeDeclare(multiExchange, "fanout");
                if (multiQueue == null) {
                        multiQueue = channel.queueDeclare().getQueue();
                } else {
                        channel.queueDeclare(multiQueue, multiDurable, multiExclusive, multiAutoDelete, null);
                }
                channel.queueBind(multiQueue, multiExchange, "");
                logger.info("RemoteObject: " + reference + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: "
                                + multiDurable + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete);

                /*
                 * Consumer
                 */

                // Disable Round Robin behavior
                boolean autoAck = false;

                // Declare a new consumer
                consumer = new QueueingConsumer(channel);
                channel.basicConsume(multiQueue, autoAck, consumer);
        }

}