Changeset 108 for branches/supervisor

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

Location:

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

Files:

• AInvocationThread.java (added)
• InvocationThread.java (modified) (8 diffs)
• MultiInvocationThread.java (modified) (1 diff)
• RemoteThreadPool.java (modified) (4 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.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;
 
-import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConsumerCancelledException;
 import com.rabbitmq.client.QueueingConsumer;
@@ -28 +17 @@
  * 
  */
-public class InvocationThread extends Thread {
+public class InvocationThread extends AInvocationThread {
 
         private static final Logger logger = Logger.getLogger(InvocationThread.class.getName());
-        private static final String multi = "multi#";
 
         // RemoteObject
-        private RemoteObject obj;
-        private String reference;
-        private String UID;
-        private Properties env;
         private boolean idle;
         private long lastExec;
 
-        private RemoteThreadPool pool;
-
-        // Broker
-        private Broker broker;
-        private Serializer serializer;
-
-        // Consumer
-        private Channel channel;
-        private QueueingConsumer consumer;
-        private String multiQueue;
-        private boolean killed = false;
-
         public InvocationThread(RemoteObject obj) throws Exception {
-                this.obj = obj;
-                this.UID = obj.getUID();
-                this.reference = obj.getRef();
-                this.env = obj.getEnv();
-                this.broker = obj.getBroker();
-                this.pool = obj.getPool();
-                this.serializer = broker.getSerializer();
+                super(obj);
                 this.lastExec = 0;
                 this.idle = true;
-        }
-
-        @Override
-        public synchronized void start() {
-                try {
-                        startQueues();
-                        super.start();
-                } catch (Exception e) {
-                        logger.error("Cannot start a remoteObject", e);
-                }
-
         }
 
@@ -87 +42 @@
                                 idle = false;
 
-                                String serializerType = delivery.getProperties().getType();
-
-                                // Deserialize the request
-                                Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj);
-                                String methodName = request.getMethod();
-                                String requestID = request.getId();
-
-                                logger.debug("Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + ", serializerType: " + serializerType);
-
-                                // Invoke the method
-                                Object result = null;
-                                OmqException error = null;
-                                try {
-                                        result = obj.invokeMethod(request.getMethod(), request.getParams());
-                                } catch (InvocationTargetException e) {
-                                        Throwable throwable = e.getTargetException();
-                                        logger.error("Object: " + obj.getRef() + " at method: " + methodName + ", corrID" + requestID, throwable);
-                                        error = new OmqException(throwable.getClass().getCanonicalName(), throwable.getMessage());
-                                } catch (NoSuchMethodException e) {
-                                        logger.error("Object: " + obj.getRef() + " cannot find method: " + methodName);
-                                        error = new OmqException(e.getClass().getCanonicalName(), e.getMessage());
-                                }
-
-                                // Reply if it's necessary
-                                if (!request.isAsync()) {
-                                        Response resp = new Response(request.getId(), obj.getRef(), result, error);
-
-                                        BasicProperties props = delivery.getProperties();
-
-                                        BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build();
-
-                                        byte[] bytesResponse = serializer.serialize(serializerType, resp);
-                                        channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse);
-                                        logger.debug("Publish sync response -> Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + " replyTo: "
-                                                        + props.getReplyTo());
-                                }
-
-                                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+                                executeTask(delivery);
 
                                 // The thread is now idle
@@ -168 +86 @@
          * @throws Exception
          */
-        private void startQueues() throws Exception {
+        protected void startQueues() throws Exception {
                 // Start channel
                 channel = broker.getNewChannel();
@@ -190 +108 @@
                         channel.queueBind(queue, exchange, routingKey);
                 }
-                logger.info("RemoteObject: " + reference + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: "
-                                + exclusive + ", AutoDelete: " + autoDelete);
+                logger.info("RemoteObject: " + reference + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable
+                                + ", Exclusive: " + exclusive + ", AutoDelete: " + autoDelete);
 
                 /*
@@ -207 +125 @@
                                 channel.queueBind(UID, exchange, UID);
                         }
+                        // TODO logger queue
+                        // TODO UID queue should be reference + UID
                 }
-
-                /*
-                 * 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);
 
                 /*
@@ -238 +133 @@
                  */
 
-                // Disable Round Robin behavior 
+                // Disable Round Robin behavior
                 boolean autoAck = false;
 
@@ -247 +142 @@
                 consumer = new QueueingConsumer(channel);
                 channel.basicConsume(queue, autoAck, consumer);
-                channel.basicConsume(multiQueue, autoAck, consumer);
                 if (UID != null) {
                         channel.basicConsume(UID, autoAck, consumer);
                 }
-        }
-
-        public void kill() throws IOException {
-                logger.info("Killing objectmq: " + reference + " thread id");
-                killed = true;
-                interrupt();
-                channel.close();
-        }
-
-        public RemoteObject getObj() {
-                return obj;
-        }
-
-        public void setObj(RemoteObject obj) {
-                this.obj = obj;
         }
 

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);
+        }
 
 }

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

@@ -23 +23 @@
         private static final Logger logger = Logger.getLogger(RemoteThreadPool.class.getName());
         private List<InvocationThread> workers;
+        private MultiInvocationThread multiWorker;
         private AtomicInteger busy;
         private int minPoolThreads;
@@ -34 +35 @@
         private boolean killed = false;
 
-        public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread, RemoteObject obj, Broker broker) {
+        public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread,
+                        RemoteObject obj, Broker broker) {
                 this.minPoolThreads = minPoolThreads;
                 this.maxPoolThreads = maxPoolThreads;
@@ -53 +55 @@
                  * Create and start minPoolThreads
                  */
-                logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads + ", refresh time: "
-                                + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread);
+                logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads
+                                + ", refresh time: " + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread);
+
+                try {
+                        multiWorker = new MultiInvocationThread(obj);
+                        multiWorker.start();
+                } catch (Exception e1) {
+                        // TODO Auto-generated catch block
+                        e1.printStackTrace();
+                }
 
                 for (int i = 0; i < minPoolThreads; i++) {
@@ -86 +96 @@
                                 } else if (numWorkers > minPoolThreads && busy.get() < numWorkers) {
                                         // Kill idle threads
-                                        System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = " + busy.get());
+                                        System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = "
+                                                        + busy.get());
                                         stopIdleThreads();
                                 }