Changeset 96 for branches/supervisor/src/main/java/omq/server/InvocationThread.java

Timestamp:

10/02/13 17:31:26 (11 years ago)

Author:

stoda

Message:

ObjectMQ without RemoteWrapper? - one consumer per thread -

File:

• branches/supervisor/src/main/java/omq/server/InvocationThread.java (modified) (7 diffs)

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

@@ -1 +1 @@
 package omq.server;
 
+import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
-import java.util.concurrent.BlockingQueue;
-
+import java.util.Properties;
+
+import omq.common.broker.Broker;
 import omq.common.message.Request;
 import omq.common.message.Response;
+import omq.common.util.ParameterQueue;
 import omq.common.util.Serializer;
 import omq.exception.OmqException;
+import omq.exception.SerializerException;
 
 import org.apache.log4j.Logger;
@@ -13 +17 @@
 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;
 
 /**
@@ -22 +29 @@
  */
 public class InvocationThread extends Thread {
+
         private static final Logger logger = Logger.getLogger(InvocationThread.class.getName());
+        private static final String multi = "multi#";
+
+        // RemoteObject
         private RemoteObject obj;
+        private String UID;
+        private Properties env;
+
+        // Broker
+        private Broker broker;
         private Serializer serializer;
-        // private RemoteWrapper wrapper;
-        private BlockingQueue<Delivery> deliveryQueue;
+
+        // Consumer
+        private Channel channel;
+        private QueueingConsumer consumer;
+        private String multiQueue;
         private boolean killed = false;
 
-        public InvocationThread(RemoteObject obj, RemoteWrapper wrapper, Serializer serializer) {
+        public InvocationThread(RemoteObject obj, Broker broker) throws Exception {
                 this.obj = obj;
-                // this.wrapper = wrapper;
-                this.deliveryQueue = wrapper.getDeliveryQueue();
-                this.serializer = serializer;
+                this.UID = obj.getRef();
+                this.env = obj.getEnv();
+                this.broker = broker;
+                this.serializer = broker.getSerializer();
+        }
+
+        @Override
+        public synchronized void start() {
+                try {
+                        startQueues();
+                        super.start();
+                } catch (Exception e) {
+                        logger.error("Cannot start a remoteObject", e);
+                }
+
         }
 
@@ -41 +72 @@
                         try {
                                 // Get the delivery
-                                Delivery delivery = deliveryQueue.take();
-
-                                // // Indicate this thread is not available
-                                // wrapper.increaseBusy();
+                                Delivery delivery = consumer.nextDelivery();
 
                                 String serializerType = delivery.getProperties().getType();
@@ -69 +97 @@
                                 }
 
-                                
-                                Channel channel = obj.getChannel();
-                                
-                                
                                 // Reply if it's necessary
                                 if (!request.isAsync()) {
                                         Response resp = new Response(request.getId(), obj.getRef(), result, error);
-
-                                        
 
                                         BasicProperties props = delivery.getProperties();
@@ -88 +110 @@
                                                         + props.getReplyTo());
                                 }
-                                
+
                                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
-                                
-                                // // Indicate this thread is available
-                                // wrapper.decreaseBusy();
                         } catch (InterruptedException i) {
                                 logger.error(i);
-                                killed = true;
+                        } 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) {
-                                logger.error("Object: " + obj.getRef(), e);
+                                logger.error(e);
                         }
 
                 }
+        }
+
+        /**
+         * This method starts the queues using the information got in the
+         * environment.
+         * 
+         * @throws Exception
+         */
+        private void startQueues() throws Exception {
+                // Start channel
+                channel = broker.getNewChannel();
+
+                /*
+                 * Default queue, Round Robin behaviour
+                 */
+
+                // Get info about which exchange and queue will use
+                String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, "");
+                String queue = UID;
+                String routingKey = UID;
+
+                // RemoteObject default queue
+                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUE, "false"));
+                boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "false"));
+                boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false"));
+
+                // Declares and bindings
+                if (!exchange.equalsIgnoreCase("")) { // Default exchange case
+                        channel.exchangeDeclare(exchange, "direct");
+                }
+                channel.queueDeclare(queue, durable, exclusive, autoDelete, null);
+                if (!exchange.equalsIgnoreCase("")) { // Default exchange case
+                        channel.queueBind(queue, exchange, routingKey);
+                }
+                logger.info("RemoteObject: " + UID + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: "
+                                + exclusive + ", AutoDelete: " + autoDelete);
+
+                /*
+                 * Multi queue, exclusive per each instance
+                 */
+
+                // Get info about the multiQueue
+                String multiExchange = multi + UID;
+                // TODO:String multiExchange = multi + exchange + UID;
+                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: " + UID + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + multiDurable
+                                + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete);
+
+                /*
+                 * Consumer
+                 */
+
+                boolean autoAck = false;
+
+                int prefetchCount = 1;
+                channel.basicQos(prefetchCount);
+
+                // Declare a new consumer
+                consumer = new QueueingConsumer(channel);
+                channel.basicConsume(queue, autoAck, consumer);
+                channel.basicConsume(multiQueue, autoAck, consumer);
+        }
+
+        public void kill() throws IOException {
+                logger.info("Killing objectmq: " + UID + " thread id");
+                killed = true;
+                interrupt();
+                channel.close();
         }
 
@@ -111 +231 @@
         }
 
-        public BlockingQueue<Delivery> getDeliveryQueue() {
-                return deliveryQueue;
-        }
-
-        public void setDeliveryQueue(BlockingQueue<Delivery> deliveryQueue) {
-                this.deliveryQueue = deliveryQueue;
-        }
 }