Changeset 55

Timestamp:

06/21/13 16:55:58 (11 years ago)

Author:

stoda

Message:

@MultiMethod? implemented and working with @AsyncMethod? annotation.
TODO: @Multi with SyncMethod?(waitNum = x) -> Must return a List<?>
Refactoring in Proxymq if it's necessary

Location:

trunk

Files:

• src/main/java/omq/client/proxy/Proxymq.java (modified) (6 diffs)
• src/main/java/omq/common/message/Request.java (modified) (2 diffs)
• src/main/java/omq/server/RemoteObject.java (modified) (3 diffs)
• src/test/java/omq/test/multiProcess/MultiProcessTest.java (modified) (1 diff)
• target/classes/log4j.xml (added)

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

@@ -48 +48 @@
         private static final long serialVersionUID = 1L;
         private static final Logger logger = Logger.getLogger(Proxymq.class.getName());
+        private static final String multi = "multi#";
         private static Map<String, Object> proxies = new Hashtable<String, Object>();
 
@@ -56 +57 @@
         private transient EventDispatcher dispatcher;
         private transient Serializer serializer;
-        // private transient Channel channel;
         private transient Properties env;
         private transient Map<String, byte[]> results;
@@ -111 +111 @@
         @Override
         public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
-                // long timeStart = (new Date()).getTime();
-
                 // Local methods only
                 String methodName = method.getName();
@@ -151 +149 @@
 
                 // Get the environment properties
-                String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
-                String routingkey = this.uid;
+                String exchange;
+                String routingkey;
+
+                if (request.isMulti()) {
+                        exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                        routingkey = "";
+                } else {
+                        exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                        routingkey = uid;
+                }
 
                 // Add the correlation ID and create a replyTo property
@@ -159 +165 @@
                 // Publish the message
                 byte[] bytesRequest = serializer.serialize(serializerType, request);
-                // TODO See this
-                // channel.basicPublish(exchange, routingkey, props, bytesRequest);
                 broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
         }
@@ -213 +217 @@
                                 timeout = sync.timeout();
                         }
-                        return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout);
+                        return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout, multi);
                 } else {
                         return Request.newAsyncRequest(corrId, methodName, arguments, multi);

trunk/src/main/java/omq/common/message/Request.java

@@ -35 +35 @@
         }
 
-        public Request(String id2, String method2, boolean b, Object[] params2, boolean multi2) {
-                // TODO Auto-generated constructor stub
+        private Request(String id, String method, boolean async, Object[] params, boolean multi) {
+                this.id = id;
+                this.method = method;
+                this.async = async;
+                this.params = params;
+                this.multi = multi;
         }
 
@@ -43 +47 @@
         }
 
-        public static Request newSyncRequest(String id, String method, Object[] params, int retries, long timeout) {
-                Request req = new Request(id, method, false, params);
+        public static Request newSyncRequest(String id, String method, Object[] params, int retries, long timeout, boolean multi) {
+                Request req = new Request(id, method, false, params, multi);
                 req.setRetries(retries);
                 req.setTimeout(timeout);

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

@@ -223 +223 @@
                 String queue = UID;
                 String routingKey = UID;
+                // Multi info
+                String multiExchange = multi + exchange;
+                String multiQueue = UID + System.currentTimeMillis();
+
                 boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
 
@@ -231 +235 @@
                 logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue);
                 channel.exchangeDeclare(exchange, "direct");
-                channel.exchangeDeclare(multi + exchange, "fanout");
                 channel.queueDeclare(queue, durable, false, false, null);
                 channel.queueBind(queue, exchange, routingKey);
-                channel.queueBind(queue, multi + exchange, routingKey);
+
+                channel.exchangeDeclare(multiExchange, "fanout");
+                channel.queueDeclare(multiQueue, durable, false, false, null);
+                channel.queueBind(multiQueue, multiExchange, "");
 
                 // Declare the event topic fanout
@@ -243 +249 @@
                 consumer = new QueueingConsumer(channel);
                 channel.basicConsume(queue, true, consumer);
+                channel.basicConsume(multiQueue, true, consumer);
         }
 

trunk/src/test/java/omq/test/multiProcess/MultiProcessTest.java

@@ -90 +90 @@
                 assertEquals(x, b);
                 remoteNumber.setNumber(0);
+                remoteNumber.setNumber(0);
         }