Changeset 84 for trunk/src/main/java/omq/server/RemoteObject.java

Timestamp:

07/10/13 13:14:30 (11 years ago)

Author:

stoda

Message:

Default queues added, default exchange enabled, more control in remote queues added.
Tests verified and changed Persistent test to show how to make persistent messages.

File:

• trunk/src/main/java/omq/server/RemoteObject.java (modified) (3 diffs)

trunk/src/main/java/omq/server/RemoteObject.java

@@ -39 +39 @@
 
         private String UID;
-        private String multiQueue;
         private Properties env;
         private transient Broker broker;
+        private transient String multiQueue;
         private transient RemoteWrapper remoteWrapper;
         private transient Map<String, List<Class<?>>> params;
@@ -77 +77 @@
                 this.broker = broker;
                 this.UID = reference;
-                this.multiQueue = UID + System.currentTimeMillis();
                 this.env = env;
 
@@ -262 +261 @@
          */
         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 exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, "");
                 String queue = UID;
                 String routingKey = UID;
 
-                // Multi info
-                String multiExchange = multi + UID;
-                if (env.getProperty(ParameterQueue.MULTI_QUEUE_NAME) != null) {
-                        multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME);
-                }
-
-                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
+                // 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"));
 
-                // Start channel
-                channel = broker.getNewChannel();
-
                 // Declares and bindings
-                logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: "
+                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);
-                channel.exchangeDeclare(exchange, "direct");
-                channel.queueDeclare(queue, durable, exclusive, autoDelete, null);
-                channel.queueBind(queue, exchange, routingKey);
-
-                logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", 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");
-                channel.queueDeclare(multiQueue, durable, exclusive, autoDelete, null);
+                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
+                 */
 
                 // Declare a new consumer