Changeset 108 for branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java

Timestamp:

10/21/13 15:27:22 (11 years ago)

Author:

stoda

Message:

Refactoring to enable multinvocation thread done.
Now there's a multiinvocation thread which listens to the multiqueue

File:

• branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java (modified) (1 diff)

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

@@ -1 +1 @@
 package omq.server;
 
-public class MultiInvocationThread {
+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);
+        }
 
 }