Changeset 58 for trunk/src/main/java

Timestamp:

06/25/13 16:38:31 (11 years ago)

Author:

stoda

Message:

@MultiMethod? + @SyncMethod? implemented and tested

Location:

trunk/src/main/java/omq

Files:

• client/proxy/Proxymq.java (modified) (9 diffs)
• common/message/Request.java (modified) (3 diffs)
• server/RemoteObject.java (modified) (3 diffs)

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

@@ -2 +2 @@
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Hashtable;
-import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-
-import org.apache.log4j.Logger;
 
 import omq.Remote;
@@ -34 +31 @@
 import omq.exception.TimeoutException;
 
+import org.apache.log4j.Logger;
+
 import com.rabbitmq.client.AMQP.BasicProperties;
 
@@ -190 +189 @@
                         try {
                                 publishMessage(request, replyQueueName);
-                                return getResult(corrId, timeout, type);
+                                if (request.isMulti()) {
+                                        return getResults(corrId, 2, timeout, type);
+                                } else {
+                                        return getResult(corrId, timeout, type);
+                                }
+
                         } catch (TimeoutException te) {
                                 logger.error(te);
@@ -203 +207 @@
                 String methodName = method.getName();
                 boolean multi = false;
+                int wait = 0;
 
                 if (method.getAnnotation(MultiMethod.class) != null) {
                         multi = true;
+                        wait = method.getAnnotation(MultiMethod.class).waitNum();
                 }
 
@@ -219 +225 @@
                                 timeout = sync.timeout();
                         }
-                        return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout, multi);
+                        return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout, multi, wait);
                 } else {
                         return Request.newAsyncRequest(corrId, methodName, arguments, multi);
@@ -229 +235 @@
 
                 // Wait for the results.
-                long localTimeout = 0;
+                long localTimeout = timeout;
                 long start = System.currentTimeMillis();
                 synchronized (results) {
                         // Due to we are using notifyAll(), we need to control the real time
                         while (!results.containsKey(corrId) && (timeout - localTimeout) >= 0) {
-                                results.wait(timeout);
+                                results.wait(localTimeout);
                                 localTimeout = System.currentTimeMillis() - start;
                         }
@@ -258 +264 @@
         }
 
-        @SuppressWarnings("unused")
         private Object getResults(String corrId, int wait, long timeout, Class<?> type) throws Exception {
                 Response resp = null;
-                List<Object> list = new ArrayList<Object>();
+                Class<?> actualType = type.getComponentType();
+
+                Object array = Array.newInstance(actualType, wait);
 
                 int i = 0;
@@ -279 +286 @@
                                 }
                                 // Remove the corrId to receive new replies
-                                resp = serializer.deserializeResponse(results.remove(corrId), type);
-                                list.add(resp.getResult());
-
+                                resp = serializer.deserializeResponse(results.remove(corrId), actualType);
+                                System.out.println("/n/n/n/n/nResult type: "+resp.getResult()+" /n/n/n/n/n");
+                                Array.set(array, i, resp.getResult());
                         }
                         i++;
@@ -289 +296 @@
                 }
 
-                return list;
+                return array;
         }
 

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

@@ -16 +16 @@
 
         private transient boolean multi;
+        private transient int wait;
         private transient long timeout;
         private transient int retries;
@@ -47 +48 @@
         }
 
-        public static Request newSyncRequest(String id, String method, Object[] params, int retries, long timeout, boolean multi) {
+        public static Request newSyncRequest(String id, String method, Object[] params, int retries, long timeout, boolean multi, int wait) {
                 Request req = new Request(id, method, false, params, multi);
                 req.setRetries(retries);
                 req.setTimeout(timeout);
+                req.setWait(wait);
                 return req;
         }
@@ -113 +115 @@
                 this.multi = multi;
         }
+
+        public int getWait() {
+                return wait;
+        }
+
+        public void setWait(int wait) {
+                this.wait = wait;
+        }
 }

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

@@ -39 +39 @@
 
         private String UID;
+        private String multiQueue;
         private Properties env;
         private transient Broker broker;
@@ -66 +67 @@
                 this.broker = broker;
                 UID = reference;
+                multiQueue = UID + System.currentTimeMillis();
                 env = broker.getEnvironment();
                 serializer = broker.getSerializer();
@@ -225 +227 @@
                 // Multi info
                 String multiExchange = multi + exchange;
-                //TODO: change this
-                String multiQueue = UID + System.currentTimeMillis();
 
                 boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));