Changeset 55 for trunk/src/main/java/omq/client/proxy/Proxymq.java

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

File:

• trunk/src/main/java/omq/client/proxy/Proxymq.java (modified) (6 diffs)

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);