Index: trunk/src/main/java/omq/client/annotation/MultiMethod.java
===================================================================
--- trunk/src/main/java/omq/client/annotation/MultiMethod.java (revision 54)
+++ trunk/src/main/java/omq/client/annotation/MultiMethod.java (revision 54)
@@ -0,0 +1,12 @@
+package omq.client.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface MultiMethod {
+ int waitNum() default 1;
+}
Index: trunk/src/main/java/omq/client/listener/ResponseListener.java
===================================================================
--- trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 53)
+++ trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 54)
@@ -55,4 +55,5 @@
@Override
public void run() {
+ logger.info("ResponseListener started");
Delivery delivery;
String uid_request;
@@ -122,4 +123,5 @@
channel.queueDeclare(reply_queue, durable, false, false, args);
+ logger.info("ResponseListener creating queue: " + reply_queue + ", durable: " + durable);
// Declare a new consumer
Index: trunk/src/main/java/omq/client/proxy/Proxymq.java
===================================================================
--- trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 53)
+++ trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 54)
@@ -15,4 +15,5 @@
import omq.Remote;
import omq.client.annotation.AsyncMethod;
+import omq.client.annotation.MultiMethod;
import omq.client.annotation.SyncMethod;
import omq.client.listener.ResponseListener;
@@ -133,7 +134,4 @@
Request request = createRequest(method, arguments);
- // Log.saveTimeSendRequestLog("Client-time-request", request.getId(),
- // method.getName(), timeStart);
-
Object response = null;
// Publish the request
@@ -144,8 +142,4 @@
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,5 +162,4 @@
// channel.basicPublish(exchange, routingkey, props, bytesRequest);
broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
- // Log.saveLog("Client-Serialize", bytesRequest);
}
@@ -203,4 +196,9 @@
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,5 +215,5 @@
return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout);
} else {
- return Request.newAsyncRequest(corrId, methodName, arguments);
+ return Request.newAsyncRequest(corrId, methodName, arguments, multi);
}
}
@@ -237,5 +235,4 @@
}
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
Index: trunk/src/main/java/omq/common/broker/Broker.java
===================================================================
--- trunk/src/main/java/omq/common/broker/Broker.java (revision 53)
+++ trunk/src/main/java/omq/common/broker/Broker.java (revision 54)
@@ -176,7 +176,9 @@
if (responseListener == null) {
responseListener = new ResponseListener(this);
+ responseListener.start();
}
if (eventDispatcher == null) {
eventDispatcher = new EventDispatcher(this);
+ eventDispatcher.start();
}
}
@@ -197,6 +199,4 @@
byte[] bytesResponse = serializer.serialize(wrapper);
channel.basicPublish(UID, "", null, bytesResponse);
-
- // Log.saveLog("Server-Serialize", bytesResponse);
}
Index: trunk/src/main/java/omq/common/event/EventDispatcher.java
===================================================================
--- trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 53)
+++ trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 54)
@@ -56,4 +56,5 @@
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,4 +73,5 @@
@Override
public void run() {
+ logger.info("EventDispatcher started");
Delivery delivery;
Event event;
@@ -84,9 +86,4 @@
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
Index: trunk/src/main/java/omq/common/message/Request.java
===================================================================
--- trunk/src/main/java/omq/common/message/Request.java (revision 53)
+++ trunk/src/main/java/omq/common/message/Request.java (revision 54)
@@ -15,4 +15,5 @@
private boolean async = false;
+ private transient boolean multi;
private transient long timeout;
private transient int retries;
@@ -34,4 +35,8 @@
}
+ 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,6 +50,6 @@
}
- 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,3 +102,10 @@
}
+ public boolean isMulti() {
+ return multi;
+ }
+
+ public void setMulti(boolean multi) {
+ this.multi = multi;
+ }
}
Index: trunk/src/main/java/omq/server/InvocationThread.java
===================================================================
--- trunk/src/main/java/omq/server/InvocationThread.java (revision 53)
+++ trunk/src/main/java/omq/server/InvocationThread.java (revision 54)
@@ -44,6 +44,4 @@
// 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,6 +75,6 @@
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());
}
Index: trunk/src/main/java/omq/server/RemoteObject.java
===================================================================
--- trunk/src/main/java/omq/server/RemoteObject.java (revision 53)
+++ trunk/src/main/java/omq/server/RemoteObject.java (revision 54)
@@ -35,4 +35,5 @@
private static final long serialVersionUID = -1778953938739846450L;
+ private static final String multi = "multi#";
private static final Logger logger = Logger.getLogger(RemoteObject.class.getName());
@@ -230,6 +231,8 @@
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
Index: trunk/src/main/resources/log4j.xml
===================================================================
--- trunk/src/main/resources/log4j.xml (revision 53)
+++ trunk/src/main/resources/log4j.xml (revision 54)
@@ -6,5 +6,5 @@
-
+
@@ -25,5 +25,5 @@
-
+