Changeset 84 for trunk/src/main/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.

Location:

trunk/src/main/java/omq

Files:

• client/listener/ResponseListener.java (modified) (2 diffs)
• client/proxy/Proxymq.java (modified) (1 diff)
• common/util/ParameterQueue.java (modified) (4 diffs)
• server/RemoteObject.java (modified) (3 diffs)

trunk/src/main/java/omq/client/listener/ResponseListener.java

@@ -120 +120 @@
 
                 String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
-                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
+                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUE, "false"));
+                boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "true"));
+                boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "true"));
 
                 int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1"));
@@ -128 +130 @@
                 }
 
-                channel.queueDeclare(reply_queue, durable, false, false, args);
-                logger.info("ResponseListener creating queue: " + reply_queue + ", durable: " + durable + "TTL: " + (ttl > 0 ? ttl : "not set"));
+                if (reply_queue == null) {
+                        reply_queue = channel.queueDeclare().getQueue();
+                        env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, reply_queue);
+                } else {
+                        channel.queueDeclare(reply_queue, durable, exclusive, autoDelete, args);
+                }
+                logger.info("ResponseListener creating queue: " + reply_queue + ", durable: " + durable + ", exclusive: " + exclusive + ", autoDelete: " + autoDelete
+                                + ", TTL: " + (ttl > 0 ? ttl : "not set"));
 
                 // Declare a new consumer

trunk/src/main/java/omq/client/proxy/Proxymq.java

@@ -92 +92 @@
                 // this.channel = Broker.getChannel();
                 env = broker.getEnvironment();
-                exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, "");
                 multiExchange = multi + uid;
                 replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);

trunk/src/main/java/omq/common/util/ParameterQueue.java

@@ -9 +9 @@
 public class ParameterQueue {
 
-        /**
-         * Set the proxy's serializer method
+        /*
+         * Connection info
          */
-        public static String PROXY_SERIALIZER = "omq.serializer";
 
         /**
-         * Set whether the messages must be compressed or not
+         * Set the clients username
          */
-        public static String ENABLE_COMPRESSION = "omq.compression";
+        public static String USER_NAME = "omq.username";
+
+        /**
+         * Set the clients password
+         */
+        public static String USER_PASS = "omq.pass";
 
         /**
@@ -30 +34 @@
 
         /**
-         * Set the clients username
+         * Set if the system will use ssl
          */
-        public static String USER_NAME = "omq.username";
+        public static String ENABLE_SSL = "omq.enable_ssl";
 
         /**
-         * Set the clients password
+         * Set how many time we have to wait to retry the connection with the server
+         * when this goes down
          */
-        public static String USER_PASS = "omq.pass";
+        public static String RETRY_TIME_CONNECTION = "omq.retry_connection";
+
+        /*
+         * Queues info
+         */
 
         /**
@@ -51 +60 @@
 
         /**
-         * Set if the queues must be durable. The queues won't be lost when rabbitmq
-         * crashes if DURABLE_QUEUES is set trues.
-         */
-        public static String DURABLE_QUEUES = "omq.durable_queue";
-
-        /**
-         * The MESSAGE_TTL_IN_QUEUES controls for how long a message published to
-         * the queues can live before it is discarded. A message that has been in
-         * the queue for longer than the configured TTL is said to be dead.
-         * 
-         * This property must be a non-negative 32 bit integer (0 <= n <= 2^32-1),
-         * describing the TTL period in milliseconds.
-         */
-        public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue";
-
-        /**
-         * Set if the system will use ssl
-         */
-        public static String ENABLE_SSL = "omq.enable_ssl";
-
-        /**
-         * Set how many time we have to wait to retry the connection with the server
-         * when this goes down
-         */
-        public static String RETRY_TIME_CONNECTION = "omq.retry_connection";
-
-        /**
-         * Set how many threads will be created to invoke remote methods
-         */
-        public static String NUM_THREADS = "omq.num_threads";
-
-        /**
          * Set the specific name of a multi queue in a specific object
          */
         public static String MULTI_QUEUE_NAME = "omq.multi_queue_name";
+
+        /**
+         * Set if a queue must be durable. The queue won't be lost when RabbitMQ
+         * crashes if DURABLE_QUEUE is set true.
+         */
+        public static String DURABLE_QUEUE = "omq.durable_queue";
 
         /**
@@ -99 +82 @@
 
         /**
+         * Set if a queue must be durable. The queue won't be lost when RabbitMQ
+         * crashes if DURABLE_QUEUE is set true.
+         */
+        public static String DURABLE_MQUEUE = "omq.durable_mqueue";
+
+        /**
+         * Set if server will delete a queue when is no longer in use
+         */
+        public static String AUTO_DELETE_MQUEUE = "omq.auto_mdelete";
+
+        /**
+         * Set if we are declaring an exclusive queue (restricted to this
+         * connection)
+         */
+        public static String EXCLUSIVE_MQUEUE = "omq.exclusive_mqueue";
+
+        /**
+         * The MESSAGE_TTL_IN_QUEUES controls for how long a message published to
+         * the queues can live before it is discarded. A message that has been in
+         * the queue for longer than the configured TTL is said to be dead.
+         * 
+         * This property must be a non-negative 32 bit integer (0 <= n <= 2^32-1),
+         * describing the TTL period in milliseconds.
+         */
+        public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue";
+
+        /*
+         * Message info
+         */
+
+        /**
+         * Set the proxy's serializer method
+         */
+        public static String PROXY_SERIALIZER = "omq.serializer";
+
+        /**
+         * Set whether the messages must be compressed or not
+         */
+        public static String ENABLE_COMPRESSION = "omq.compression";
+
+        /**
          * Set 1 to indicate the message will be nonpersistent and 2 to indicate it
          * will be persistent
          */
         public static String DELIVERY_MODE = "omq.delivery_mode";
+
+        /*
+         * ObjectMQ info
+         */
+
+        /**
+         * Set how many threads will be created to invoke remote methods
+         */
+        public static String NUM_THREADS = "omq.num_threads";
 
         /**

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