Changeset 75 for trunk/src

Timestamp:

07/02/13 11:50:11 (11 years ago)

Author:

stoda

Message:

Multi problem solved

Location:

trunk/src

Files:

• main/java/omq/client/proxy/MultiProxymq.java (modified) (4 diffs)
• main/java/omq/client/proxy/Proxymq.java (modified) (1 diff)
• main/java/omq/common/message/Request.java (modified) (2 diffs)
• main/java/omq/common/util/Serializers/GsonImp.java (modified) (2 diffs)
• main/java/omq/server/RemoteObject.java (modified) (2 diffs)
• main/resources/log4j.xml (modified) (2 diffs)
• test/java/omq/test/workspace/WorkspaceTest.java (modified) (4 diffs)

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

@@ -4 +4 @@
 import java.lang.reflect.Method;
 import java.util.Properties;
+
+import org.apache.log4j.Logger;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
@@ -20 +22 @@
  */
 public class MultiProxymq implements InvocationHandler {
+        private static final Logger logger = Logger.getLogger(MultiProxymq.class.getName());
         private static final String multi = "multi#";
 
@@ -37 +40 @@
                 Properties env = broker.getEnvironment();
                 replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
-                exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                exchange = multi + uid;
                 serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA);
         }
@@ -55 +58 @@
                 broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
 
+                logger.debug("Proxymq: " + uid + " invokes " + methodName + ", corrID" + corrId + ", exchange: " + exchange + ", replyQueue: " + replyQueueName
+                                + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync());
+
                 return null;
         }

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

@@ -90 +90 @@
                 env = broker.getEnvironment();
                 exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
-                multiExchange = multi + exchange;
+                multiExchange = multi + uid;
                 replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
 

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

@@ -23 +23 @@
         }
 
-        public Request(String id, String method, Object[] params) {
-                this.id = id;
-                this.method = method;
-                this.params = params;
-        }
-
-        private Request(String id, String method, boolean async, Object[] params) {
+        public Request(String id, String method, boolean async, Object[] params) {
                 this.id = id;
                 this.method = method;
@@ -36 +30 @@
         }
 
-        private Request(String id, String method, boolean async, Object[] params, boolean multi) {
+        public Request(String id, String method, boolean async, Object[] params, boolean multi) {
                 this.id = id;
                 this.method = method;

trunk/src/main/java/omq/common/util/Serializers/GsonImp.java

@@ -33 +33 @@
                 String id = jsonObj.get("id").getAsString();
                 String method = jsonObj.get("method").getAsString();
+                boolean async = jsonObj.get("async").getAsBoolean();
 
                 List<Class<?>> types = obj.getParams(method);
@@ -48 +49 @@
                                 i++;
                         }
-
-                        return new Request(id, method, arguments);
+                        return new Request(id, method, async, arguments);
                 } catch (NullPointerException e) {
-                        return new Request(id, method, null);
+                        return new Request(id, method, async, null);
                 }
         }

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

@@ -97 +97 @@
                                 Delivery delivery = consumer.nextDelivery();
 
-                                logger.debug(UID + " has received a message");
+                                logger.debug(UID + " has received a message, serializer: " + delivery.getProperties().getType());
 
                                 remoteWrapper.notifyDelivery(delivery);
@@ -219 +219 @@
                 String routingKey = UID;
                 // Multi info
-                String multiExchange = multi + exchange;
+                String multiExchange = multi + UID;
 
                 boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));

trunk/src/main/resources/log4j.xml

@@ -4 +4 @@
 <log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
     <appender name="consoleAppender" class="org.apache.log4j.ConsoleAppender">
-        <param name="Threshold" value="INFO" />
+        <param name="Threshold" value="DEBUG" />
         <layout class="org.apache.log4j.PatternLayout">
             <param name="ConversionPattern" value="%d{[yyyy-MM-dd HH:mm:ss]} %-5p %c:%L - %m%n" />
@@ -30 +30 @@
 
   <root> 
-    <priority value ="info" /> 
+    <priority value ="DEBUG" /> 
     <appender-ref ref="consoleAppender" />
     <appender-ref ref="A2" />  

trunk/src/test/java/omq/test/workspace/WorkspaceTest.java

@@ -20 +20 @@
 @RunWith(value = Parameterized.class)
 public class WorkspaceTest {
+        private String type;
         private static Broker serverBroker;
         private static Broker clientBroker1;
         private static Broker clientBroker2;
+        private static RemoteWorkspaceImpl w1C1;
+        private static RemoteWorkspaceImpl w3C1;
+        private static RemoteWorkspaceImpl w1C2;
+        private static RemoteWorkspaceImpl w2C2;
         private static String[] workspaces = { "w1", "w2", "w3" };
         private static RemoteWorkspace[] remoteWorks = new RemoteWorkspace[3];
@@ -28 +33 @@
         // In this case the Constructor acts as a server
         public WorkspaceTest(String type) throws Exception {
+                this.type = type;
                 Properties env = new Properties();
                 env.setProperty(ParameterQueue.USER_NAME, "guest");
@@ -75 +81 @@
                 clientBroker2 = new Broker(env);
 
+                // Client 1 will subscribe to changes in the workspaces w1 and w3
+                w1C1 = new RemoteWorkspaceImpl();
+                w3C1 = new RemoteWorkspaceImpl();
+
+                clientBroker1.bind(workspaces[0], w1C1);
+                clientBroker1.bind(workspaces[2], w3C1);
+
+                // Client 2 will subscribe to changes in the workspaces w1 and w2
+                w1C2 = new RemoteWorkspaceImpl();
+                w2C2 = new RemoteWorkspaceImpl();
+
+                clientBroker2.bind(workspaces[0], w1C2);
+                clientBroker2.bind(workspaces[1], w2C2);
+
                 System.out.println("Client 1 & client2 started");
         }
@@ -85 +105 @@
         @Test
         public void test() throws Exception {
+                System.out.println("Starting test: " + type);
                 String expected = null;
                 String actual = null;
-
-                // Client 1 will subscribe to changes in the workspaces w1 and w3
-                RemoteWorkspaceImpl w1C1 = new RemoteWorkspaceImpl();
-                RemoteWorkspaceImpl w3C1 = new RemoteWorkspaceImpl();
-
-                clientBroker1.bind(workspaces[0], w1C1);
-                clientBroker1.bind(workspaces[2], w3C1);
-
-                // Client 2 will subscribe to changes in the workspaces w1 and w2
-                RemoteWorkspaceImpl w1C2 = new RemoteWorkspaceImpl();
-                RemoteWorkspaceImpl w2C2 = new RemoteWorkspaceImpl();
-
-                clientBroker2.bind(workspaces[0], w1C2);
-                clientBroker2.bind(workspaces[2], w2C2);
 
                 // The server will notify a change in the w2