Changeset 77 for trunk/src/main

Timestamp:

07/04/13 16:45:14 (11 years ago)

Author:

stoda

Message:

ParameterQueues? changed, added some properties to modify the queues

Location:

trunk/src/main/java/omq

Files:

• client/proxy/MultiProxymq.java (modified) (1 diff)
• client/proxy/Proxymq.java (modified) (3 diffs)
• common/util/OmqConnectionFactory.java (modified) (1 diff)
• common/util/ParameterQueue.java (modified) (3 diffs)
• common/util/Serializer.java (modified) (3 diffs)
• server/RemoteObject.java (modified) (2 diffs)

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

@@ -41 +41 @@
                 replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
                 exchange = multi + uid;
-                serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA);
+                serializerType = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA);
         }
 

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

@@ -51 +51 @@
         private transient Serializer serializer;
         private transient Properties env;
+        private transient Integer deliveryMode = null;
         private transient Map<String, byte[]> results;
 
@@ -94 +95 @@
 
                 // set the serializer type
-                serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA);
+                serializerType = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA);
+                if (env.getProperty(ParameterQueue.DELIVERY_MODE) != null) {
+                        deliveryMode = Integer.parseInt(env.getProperty(ParameterQueue.DELIVERY_MODE));
+                }
 
                 // Create a new hashmap and registry it in rListener
@@ -143 +147 @@
 
                 // Add the correlation ID and create a replyTo property
-                BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType).build();
+                BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType)
+                                .deliveryMode(deliveryMode).build();
 
                 // Publish the message
                 byte[] bytesRequest = serializer.serialize(serializerType, request);
                 broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
-                logger.debug("Proxymq: " + uid + " invokes " + request.getMethod() + ", corrID" + corrId + ", exchange: " + exchange + ", replyQueue: "
-                                + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync());
+                logger.debug("Proxymq: " + uid + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange + ", replyQueue: "
+                                + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync()
+                                + ", delivery mode: " + deliveryMode);
         }
 

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

@@ -56 +56 @@
 
                 // Get host info of rabbimq (where it is)
-                String host = env.getProperty(ParameterQueue.SERVER_HOST);
-                int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT));
+                String host = env.getProperty(ParameterQueue.RABBIT_HOST);
+                int port = Integer.parseInt(env.getProperty(ParameterQueue.RABBIT_PORT));
 
                 boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL));

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

@@ -12 +12 @@
          */
 
-        public static String SERIALIZER_NAME = "omq.serializer";
+        /**
+         * 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 ENABLECOMPRESSION = "omq.compression";
+        public static String ENABLE_COMPRESSION = "omq.compression";
 
         /**
          * Set the ip where the rabbitmq server is.
          */
-        public static String SERVER_HOST = "omq.host";
+        public static String RABBIT_HOST = "omq.host";
 
         /**
          * Set the port that rabbitmq uses.
          */
-        public static String SERVER_PORT = "omq.port";
+        public static String RABBIT_PORT = "omq.port";
 
         /**
@@ -51 +54 @@
 
         /**
-         * Set the clients event queue. Every client must have a different queue
-         * name.
-         */
-        public static String EVENT_REPLY_QUEUE = "omq.reply_queue_event";
-
-        /**
          * Set if the queues must be durable. The queues won't be lost when rabbitmq
          * crashes if DURABLE_QUEUES is set trues.
@@ -72 +69 @@
         public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue";
 
-        // TODO persisten messages? the messages will be saved in the disk if this
+        // TODO persistent messages? the messages will be saved in the disk if this
         // flag is set true
 
+        /**
+         * Set if the system will use ssl
+         */
         public static String ENABLE_SSL = "omq.enable_ssl";
-        public static String DEBUGFILE = "omq.debug_file";
 
+        /**
+         * 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";
 
-        /*
-         * Values
+        /**
+         * Set how many threads will be created to invoke remote methods
          */
-
-        // Change this!!!
-        public static String RPC_TYPE = "direct";
-
         public static String NUM_THREADS = "omq.num_threads";
 
-        public static String REGISTRY_NAME = "REGISTRY";
+        /**
+         * Set the specific name of a multi queue in a specific object
+         */
+        public static String MULTI_QUEUE_NAME = "omq.multi_queue_name";
+
+        /**
+         * Set if server will delete a queue when is no longer in use
+         */
+        public static String AUTO_DELETE_QUEUE = "omq.auto_delete";
+
+        /**
+         * Set if we are declaring an exclusive queue (restricted to this
+         * connection)
+         */
+        public static String EXCLUSIVE_QUEUE = "omq.exclusive_queue";
+
+        /**
+         * 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";
 
         /**

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

@@ -40 +40 @@
 
         private Boolean getEnableCompression() {
-                return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false"));
+                return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLE_COMPRESSION, "false"));
         }
 
@@ -46 +46 @@
                 if (serializer == null) {
                         try {
-                                String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA);
+                                String className = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA);
 
                                 if (className == null || className.isEmpty()) {
@@ -98 +98 @@
         }
 
-        // TODO: remove this function and think about the event serialization
         public byte[] serialize(Object obj) throws SerializerException {
                 ISerializer instance = getInstance();

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

@@ -218 +218 @@
                 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"));
+                boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "false"));
+                boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false"));
 
                 // Start channel
@@ -227 +233 @@
 
                 // Declares and bindings
-                logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable);
+                logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: "
+                                + exclusive + ", AutoDelete: " + autoDelete);
                 channel.exchangeDeclare(exchange, "direct");
-                channel.queueDeclare(queue, durable, false, false, null);
+                channel.queueDeclare(queue, durable, exclusive, autoDelete, null);
                 channel.queueBind(queue, exchange, routingKey);
 
-                logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + durable);
+                logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + durable
+                                + ", Exclusive: " + exclusive + ", AutoDelete: " + autoDelete);
                 channel.exchangeDeclare(multiExchange, "fanout");
-                channel.queueDeclare(multiQueue, durable, false, false, null);
+                channel.queueDeclare(multiQueue, durable, exclusive, autoDelete, null);
                 channel.queueBind(multiQueue, multiExchange, "");