Changeset 96 for branches/supervisor/src/main/java/omq/server/RemoteObject.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/RemoteObject.java (modified) (7 diffs)

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

@@ -12 +12 @@
 import omq.common.broker.Broker;
 import omq.common.util.ParameterQueue;
-import omq.exception.SerializerException;
 
 import org.apache.log4j.Logger;
-
-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;
 
 /**
@@ -32 +25 @@
  * 
  */
-public abstract class RemoteObject extends Thread implements Remote {
+public abstract class RemoteObject implements Remote {
 
         private static final long serialVersionUID = -1778953938739846450L;
-        private static final String multi = "multi#";
         private static final Logger logger = Logger.getLogger(RemoteObject.class.getName());
 
@@ -41 +33 @@
         private Properties env;
         private transient Broker broker;
-        private transient String multiQueue;
-        private transient RemoteWrapper remoteWrapper;
         private transient Map<String, List<Class<?>>> params;
-        private transient Channel channel;
-        private transient QueueingConsumer consumer;
-        private transient boolean killed = false;
+        private transient List<InvocationThread> invocationList;
 
         private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();
@@ -60 +48 @@
         }
 
-        public RemoteObject() {
-        }
-
         /**
          * This method starts a remoteObject.
@@ -90 +75 @@
                 // Get num threads to use
                 int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1"));
-                this.remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer());
-
-                startQueues();
-
-                // Start this listener
-                this.start();
-        }
-
-        @Override
-        public void run() {
-                while (!killed) {
-                        try {
-                                Delivery delivery = consumer.nextDelivery();
-
-                                logger.debug(UID + " has received a message, serializer: " + delivery.getProperties().getType());
-
-                                remoteWrapper.notifyDelivery(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) {
-                                logger.error(e);
-                        }
-                }
+                invocationList = new ArrayList<InvocationThread>(numThreads);
+
+                // Start invocation threads
+                for (int i = 0; i < numThreads; i++) {
+                        InvocationThread iThread = new InvocationThread(this, broker);
+                        invocationList.add(iThread);
+                        iThread.start();
+                }
+
         }
 
@@ -147 +98 @@
          */
         public void kill() throws IOException {
-                logger.warn("Killing objectmq: " + this.getRef());
-                killed = true;
-                interrupt();
-                channel.close();
-                remoteWrapper.stopRemoteWrapper();
+                logger.info("Killing objectmq: " + this.getRef());
+                for (InvocationThread iThread : invocationList) {
+                        iThread.kill();
+                }
         }
 
@@ -250 +200 @@
         }
 
-        public Channel getChannel() {
-                return channel;
-        }
-
         public Broker getBroker() {
                 return broker;
         }
 
-        /**
-         * 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;
-                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;
-
-                //TODO see if this is useless
-                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 Properties getEnv() {
+                return env;
         }