Changeset 58

Timestamp:

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

Author:

stoda

Message:

@MultiMethod? + @SyncMethod? implemented and tested

Location:

trunk/src

Files:

• main/java/omq/client/proxy/Proxymq.java (modified) (9 diffs)
• main/java/omq/common/message/Request.java (modified) (3 diffs)
• main/java/omq/server/RemoteObject.java (modified) (3 diffs)
• test/java/omq/test/multiProcess/MultiProcessTest.java (modified) (4 diffs)
• test/java/omq/test/multiProcess/Number.java (modified) (1 diff)
• test/java/omq/test/multiProcess/NumberClient.java (added)
• test/java/omq/test/multiProcess/NumberImpl.java (modified) (2 diffs)
• test/java/omq/test/multiThread (deleted)

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

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

@@ -21 +21 @@
 public class MultiProcessTest {
         public static Broker broker;
-        public static Number remoteNumber;
+        public static NumberClient remoteNumber;
 
         public MultiProcessTest(String type) throws Exception {
@@ -44 +44 @@
 
                 broker = new Broker(env);
-                remoteNumber = broker.lookup("number", Number.class);
+                remoteNumber = broker.lookup("number", NumberClient.class);
         }
 
@@ -85 +85 @@
                 int x = 10;
                 remoteNumber.setNumber(x);
-                int a = remoteNumber.getNumer();
+                int a = remoteNumber.getNumber();
                 assertEquals(0, a);
-                int b = remoteNumber.getNumer();
+                int b = remoteNumber.getNumber();
                 assertEquals(x, b);
                 remoteNumber.setNumber(0);
@@ -98 +98 @@
                 remoteNumber.setMultiNumber(x);
                 Thread.sleep(200);
-                int a = remoteNumber.getNumer();
-                int b = remoteNumber.getNumer();
+                int a = remoteNumber.getNumber();
+                int b = remoteNumber.getNumber();
                 assertEquals(x, a);
                 assertEquals(x, b);
+                int[] number = remoteNumber.getMultiNumber();
+                assertEquals(x, number[0]);
+                assertEquals(x, number[1]);
                 remoteNumber.setMultiNumber(0);
                 Thread.sleep(200);

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

@@ -2 +2 @@
 
 import omq.Remote;
-import omq.client.annotation.AsyncMethod;
-import omq.client.annotation.MultiMethod;
-import omq.client.annotation.RemoteInterface;
-import omq.client.annotation.SyncMethod;
 
-@RemoteInterface
 public interface Number extends Remote {
-        @SyncMethod(timeout = 1000)
         public void setNumber(int x);
 
-        @SyncMethod(timeout = 1000)
-        public int getNumer();
+        public int getNumber();
 
-        @MultiMethod
-        @AsyncMethod
         public void setMultiNumber(int x);
+
+        public int getMultiNumber();
 }

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

@@ -1 +1 @@
 package omq.test.multiProcess;
 
-import omq.client.annotation.AsyncMethod;
-import omq.client.annotation.MultiMethod;
-import omq.client.annotation.SyncMethod;
 import omq.server.RemoteObject;
 
@@ -20 +17 @@
         }
 
-        @Override
-        @SyncMethod
         public void setNumber(int x) {
                 this.x = x;
         }
 
-        @Override
-        @SyncMethod(timeout = 1000)
-        public int getNumer() {
+        public int getNumber() {
                 return x;
         }
 
-        @Override
-        @MultiMethod
-        @AsyncMethod
         public void setMultiNumber(int x) {
                 this.x = x;
         }
 
+        public int getMultiNumber() {
+                return x;
+        }
+
 }