Changeset 54 for trunk/src/main/java/omq

Timestamp:

06/21/13 12:42:25 (11 years ago)

Author:

stoda

Message:

Adding @MultiMethod?
Broker is not a singleton.

Location:

trunk/src/main/java/omq

Files:

• client/annotation/MultiMethod.java (added)
• client/listener/ResponseListener.java (modified) (2 diffs)
• client/proxy/Proxymq.java (modified) (7 diffs)
• common/broker/Broker.java (modified) (2 diffs)
• common/event/EventDispatcher.java (modified) (3 diffs)
• common/message/Request.java (modified) (4 diffs)
• server/InvocationThread.java (modified) (2 diffs)
• server/RemoteObject.java (modified) (2 diffs)

trunk/src/main/java/omq/client/listener/ResponseListener.java

@@ -55 +55 @@
         @Override
         public void run() {
+                logger.info("ResponseListener started");
                 Delivery delivery;
                 String uid_request;
@@ -122 +123 @@
 
                 channel.queueDeclare(reply_queue, durable, false, false, args);
+                logger.info("ResponseListener creating queue: " + reply_queue + ", durable: " + durable);
 
                 // Declare a new consumer

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

@@ -15 +15 @@
 import omq.Remote;
 import omq.client.annotation.AsyncMethod;
+import omq.client.annotation.MultiMethod;
 import omq.client.annotation.SyncMethod;
 import omq.client.listener.ResponseListener;
@@ -133 +134 @@
                 Request request = createRequest(method, arguments);
 
-                // Log.saveTimeSendRequestLog("Client-time-request", request.getId(),
-                // method.getName(), timeStart);
-
                 Object response = null;
                 // Publish the request
@@ -144 +142 @@
                         logger.debug("Publish sync request -> " + request.getId());
                         response = publishSyncRequest(request, method.getReturnType());
-
-                        // long timeEnd = (new Date()).getTime();
-                        // Log.saveTimeSendRequestLog("Client-time-response",
-                        // request.getId(), method.getName(), timeEnd);
                 }
 
@@ -168 +162 @@
                 // channel.basicPublish(exchange, routingkey, props, bytesRequest);
                 broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
-                // Log.saveLog("Client-Serialize", bytesRequest);
         }
 
@@ -203 +196 @@
                 String corrId = java.util.UUID.randomUUID().toString();
                 String methodName = method.getName();
+                boolean multi = false;
+
+                if (method.getAnnotation(MultiMethod.class) != null) {
+                        multi = true;
+                }
 
                 // Since we need to know whether the method is async and if it has to
@@ -217 +215 @@
                         return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout);
                 } else {
-                        return Request.newAsyncRequest(corrId, methodName, arguments);
+                        return Request.newAsyncRequest(corrId, methodName, arguments, multi);
                 }
         }
@@ -237 +235 @@
                         }
                         resp = serializer.deserializeResponse(results.get(corrId), type);
-                        // Log.saveLog("Client-Deserialize", results.get(corrId));
 
                         // Remove and indicate the key exists (a hashmap can contain a null

trunk/src/main/java/omq/common/broker/Broker.java

@@ -176 +176 @@
                 if (responseListener == null) {
                         responseListener = new ResponseListener(this);
+                        responseListener.start();
                 }
                 if (eventDispatcher == null) {
                         eventDispatcher = new EventDispatcher(this);
+                        eventDispatcher.start();
                 }
         }
@@ -197 +199 @@
                 byte[] bytesResponse = serializer.serialize(wrapper);
                 channel.basicPublish(UID, "", null, bytesResponse);
-
-                // Log.saveLog("Server-Serialize", bytesResponse);
         }
 

trunk/src/main/java/omq/common/event/EventDispatcher.java

@@ -56 +56 @@
                 boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
                 channel.queueDeclare(event_queue, durable, false, false, null);
+                logger.info("EventDispatcher creating queue: " + event_queue + ", durable: " + durable);
 
                 // Declare a new consumer
@@ -72 +73 @@
         @Override
         public void run() {
+                logger.info("EventDispatcher started");
                 Delivery delivery;
                 Event event;
@@ -84 +86 @@
 
                                 logger.info("Event received -> Topic: " + event.getTopic() + "CorrId: " + event.getCorrId());
-                                // Log.saveLog("Client-Deserialize", delivery.getBody());
-
-                                // long timeEnd = (new Date()).getTime();
-                                // Log.saveTimeSendRequestLog("Client-time-response",
-                                // event.getCorrId(), "Event!", timeEnd);
 
                                 // Dispatch it

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

@@ -15 +15 @@
         private boolean async = false;
 
+        private transient boolean multi;
         private transient long timeout;
         private transient int retries;
@@ -34 +35 @@
         }
 
+        public Request(String id2, String method2, boolean b, Object[] params2, boolean multi2) {
+                // TODO Auto-generated constructor stub
+        }
+
         public static Request newSyncRequest(String id, String method, Object[] params) {
                 return new Request(id, method, false, params);
@@ -45 +50 @@
         }
 
-        public static Request newAsyncRequest(String id, String method, Object[] params) {
-                return new Request(id, method, true, params);
+        public static Request newAsyncRequest(String id, String method, Object[] params, boolean multi) {
+                return new Request(id, method, true, params, multi);
         }
 
@@ -97 +102 @@
         }
 
+        public boolean isMulti() {
+                return multi;
+        }
+
+        public void setMulti(boolean multi) {
+                this.multi = multi;
+        }
 }

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

@@ -44 +44 @@
                                 // Deserialize the json
                                 Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj);
-                                // Log.saveLog("Server-Deserialize", delivery.getBody());
-
                                 String methodName = request.getMethod();
                                 String requestID = request.getId();
@@ -77 +75 @@
                                         byte[] bytesResponse = serializer.serialize(serializerType, resp);
                                         channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse);
-
-                                        // Log.saveLog("Server-Serialize", bytesResponse);
+                                        logger.debug("Publish sync response -> Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + " replyTo: "
+                                                        + props.getReplyTo());
                                 }
 

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

@@ -35 +35 @@
 
         private static final long serialVersionUID = -1778953938739846450L;
+        private static final String multi = "multi#";
         private static final Logger logger = Logger.getLogger(RemoteObject.class.getName());
 
@@ -230 +231 @@
                 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);
 
                 // Declare the event topic fanout