Index: /trunk/objectmq/src/omq/Remote.java =================================================================== --- /trunk/objectmq/src/omq/Remote.java (revision 34) +++ /trunk/objectmq/src/omq/Remote.java (revision 35) @@ -25,8 +25,8 @@ public void notifyEvent(Event event) throws IOException, SerializerException; - public void addListener(EventListener eventListener) throws Exception; + public void addListener(EventListener eventListener) throws Exception; - public void removeListener(EventListener eventListener) throws Exception; + public void removeListener(EventListener eventListener) throws Exception; - public Collection getListeners() throws Exception; + public Collection> getListeners() throws Exception; } Index: /trunk/objectmq/src/omq/client/listener/ResponseListener.java =================================================================== --- /trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 34) +++ /trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 35) @@ -2,4 +2,5 @@ import java.io.IOException; +import java.util.HashMap; import java.util.Hashtable; import java.util.Map; @@ -47,6 +48,16 @@ this.results = new Hashtable>(); + Map args = null; + String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); - channel.queueDeclare(reply_queue, false, false, false, null); + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + + int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1")); + if (ttl > 0) { + args = new HashMap(); + args.put("x-message-ttl", ttl); + } + + channel.queueDeclare(reply_queue, durable, false, false, args); // Declare a new consumer @@ -73,6 +84,5 @@ // Stores the new response - Map proxyResults = results - .get(props.getAppId()); + Map proxyResults = results.get(props.getAppId()); // Put the result into the proxy results and notify him @@ -119,6 +129,5 @@ * @throws Exception */ - public static ResponseListener getRequestListener(Properties env) - throws Exception { + public static ResponseListener getRequestListener(Properties env) throws Exception { if (rListener == null) { rListener = new ResponseListener(env); @@ -148,5 +157,5 @@ return rListener; } - + public synchronized Channel getChannel() throws Exception { return connection.createChannel(); Index: /trunk/objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 34) +++ /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 35) @@ -51,5 +51,5 @@ private transient Properties env; private transient Map results; - private transient Map listeners; + private transient Map> listeners; private static final Map> primitiveClasses = new HashMap>(); @@ -88,5 +88,5 @@ this.env = env; - listeners = new HashMap(); + listeners = new HashMap>(); // Create a new hashmap and registry it in rListener @@ -107,8 +107,8 @@ return getRef(); } else if (methodName.equals("addListener")) { - addListener((EventListener) arguments[0]); + addListener((EventListener) arguments[0]); return null; } else if (methodName.equals("removeListener")) { - removeListener((EventListener) arguments[0]); + removeListener((EventListener) arguments[0]); return null; } else if (methodName.equals("getListeners")) { @@ -303,5 +303,5 @@ @Override - public void addListener(EventListener eventListener) throws Exception { + public void addListener(EventListener eventListener) throws Exception { if (eventListener.getTopic() == null) { eventListener.setTopic(uid); @@ -312,5 +312,5 @@ @Override - public void removeListener(EventListener eventListener) throws Exception { + public void removeListener(EventListener eventListener) throws Exception { listeners.remove(eventListener.getTopic()); dispatcher.removeListener(eventListener); @@ -318,5 +318,5 @@ @Override - public Collection getListeners() throws Exception { + public Collection> getListeners() throws Exception { return listeners.values(); } Index: /trunk/objectmq/src/omq/common/event/EventDispatcher.java =================================================================== --- /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 34) +++ /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 35) @@ -24,4 +24,5 @@ * */ +@SuppressWarnings("rawtypes") public class EventDispatcher extends Thread { private static EventDispatcher dispatcher; @@ -43,5 +44,6 @@ String event_queue = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); - channel.queueDeclare(event_queue, false, false, false, null); + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + channel.queueDeclare(event_queue, durable, false, false, null); // Declare a new consumer @@ -96,9 +98,10 @@ System.out.println("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); - + // Log.saveLog("Client-Deserialize", delivery.getBody()); + + // long timeEnd = (new Date()).getTime(); + // Log.saveTimeSendRequestLog("Client-time-response", + // event.getCorrId(), "Event!", timeEnd); + // Dispatch it dispatch(event.getTopic(), event); @@ -171,4 +174,5 @@ for (final EventListener listener : listeners.get(topic)) { new Thread() { + @SuppressWarnings("unchecked") public void run() { listener.notifyEvent(event); Index: /trunk/objectmq/src/omq/common/util/ParameterQueue.java =================================================================== --- /trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 34) +++ /trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 35) @@ -12,18 +12,66 @@ */ - public static String SERIALIZERNAME = "omq.serializer"; + public static String SERIALIZER_NAME = "omq.serializer"; + + /** + * Set whether the messages must be compressed or not + */ public static String ENABLECOMPRESSION = "omq.compression"; + /** + * Set the ip where the rabbitmq server is. + */ public static String SERVER_HOST = "omq.host"; + + /** + * Set the port that rabbitmq uses. + */ public static String SERVER_PORT = "omq.port"; - public static String SERVER_REGISTRY = "omq.registry"; + /** + * Set the clients username + */ public static String USER_NAME = "omq.username"; + + /** + * Set the clients password + */ public static String USER_PASS = "omq.pass"; + /** + * Set the exchange where the objectmq are listening + */ public static String RPC_EXCHANGE = "omq.rpc_exchange"; + /** + * Set the clients reply queue. Every client must have a different queue + * name. + */ public static String RPC_REPLY_QUEUE = "omq.reply_queue_rpc"; + + /** + * Set the clients event queue. Every client must have a different queue + * name. + */ public static String EVENT_REPLY_QUEUE = "omq.reply_queue_event"; + + /** + * Set if the queues must be durable. The queues won't be lost when rabbitmq + * crashes if DURABLE_QUEUES is set trues. + */ + public static String DURABLE_QUEUES = "omq.durable_queue"; + + /** + * The MESSAGE_TTL_IN_QUEUES controls for how long a message published to + * the queues can live before it is discarded. A message that has been in + * the queue for longer than the configured TTL is said to be dead. + * + * This property must be a non-negative 32 bit integer (0 <= n <= 2^32-1), + * describing the TTL period in milliseconds. + */ + public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue"; + + // TODO persisten messages? the messages will be saved in the disk if this + // flag is set true public static String ENABLE_SSL = "omq.enable_ssl"; Index: /trunk/objectmq/src/omq/common/util/Serializer.java =================================================================== --- /trunk/objectmq/src/omq/common/util/Serializer.java (revision 34) +++ /trunk/objectmq/src/omq/common/util/Serializer.java (revision 35) @@ -7,5 +7,8 @@ import omq.common.message.Request; import omq.common.message.Response; +import omq.common.util.Serializers.GsonImp; import omq.common.util.Serializers.ISerializer; +import omq.common.util.Serializers.JavaImp; +import omq.common.util.Serializers.KryoImp; import omq.exception.EnvironmentException; import omq.exception.SerializerException; @@ -18,4 +21,8 @@ */ public class Serializer { + public static String kryo = KryoImp.class.getCanonicalName(); + public static String java = JavaImp.class.getCanonicalName(); + public static String gson = GsonImp.class.getCanonicalName(); + public static ISerializer serializer; @@ -36,5 +43,5 @@ try { Properties env = Environment.getEnvironment(); - String className = env.getProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.JavaImp"); + String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); if (className == null || className.isEmpty()) { Index: /trunk/objectmq/src/omq/server/InvocationThread.java =================================================================== --- /trunk/objectmq/src/omq/server/InvocationThread.java (revision 34) +++ /trunk/objectmq/src/omq/server/InvocationThread.java (revision 35) @@ -42,18 +42,7 @@ System.out.println("Invoke method: " + methodName + " CorrID: " + requestID); - // Changed --------------------------------------- - Object result = null; - if ("commit".equalsIgnoreCase(methodName)) { - Object[] arguments = request.getArguments(); - arguments[1] = ((String) arguments[1]) + "@@" + requestID; - result = obj.invokeMethod(methodName, arguments); - } else { - result = obj.invokeMethod(request.getMethod(), request.getArguments()); - } - // ----------------------------------------------- - - // // Invoke the method - // Object result = obj.invokeMethod(request.getMethod(), - // request.getArguments()); + // Invoke the method + Object result = obj.invokeMethod(request.getMethod(), + request.getArguments()); if (!request.isAsync()) { Index: /trunk/objectmq/src/omq/server/RemoteObject.java =================================================================== --- /trunk/objectmq/src/omq/server/RemoteObject.java (revision 34) +++ /trunk/objectmq/src/omq/server/RemoteObject.java (revision 35) @@ -101,5 +101,4 @@ Thread.sleep(milis); } catch (InterruptedException e2) { - // TODO Auto-generated catch block e2.printStackTrace(); } @@ -211,4 +210,5 @@ String queue = UID; String routingKey = UID; + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); // Start channel @@ -218,5 +218,5 @@ System.out.println("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue); channel.exchangeDeclare(exchange, "direct"); - channel.queueDeclare(queue, false, false, false, null); + channel.queueDeclare(queue, durable, false, false, null); channel.queueBind(queue, exchange, routingKey); @@ -231,13 +231,13 @@ @Override - public void addListener(EventListener eventListener) throws Exception { - } - - @Override - public void removeListener(EventListener eventListener) throws Exception { - } - - @Override - public Collection getListeners() throws Exception { + public void addListener(EventListener eventListener) throws Exception { + } + + @Override + public void removeListener(EventListener eventListener) throws Exception { + } + + @Override + public Collection> getListeners() throws Exception { return null; } Index: /trunk/objectmq/test/calculatorTest/ClientTest.java =================================================================== --- /trunk/objectmq/test/calculatorTest/ClientTest.java (revision 34) +++ /trunk/objectmq/test/calculatorTest/ClientTest.java (revision 35) @@ -7,4 +7,5 @@ import omq.common.broker.Broker; import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; import org.junit.BeforeClass; @@ -24,7 +25,6 @@ env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); env.setProperty(ParameterQueue.SERVER_PORT, "5672"); - // env.setProperty(ParameterQueue.SERIALIZERNAME, - // "omq.common.util.Serializers.KryoImp"); - env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.kryo); env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); Index: /trunk/objectmq/test/calculatorTest/ServerTest.java =================================================================== --- /trunk/objectmq/test/calculatorTest/ServerTest.java (revision 34) +++ /trunk/objectmq/test/calculatorTest/ServerTest.java (revision 35) @@ -5,4 +5,5 @@ import omq.common.broker.Broker; import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; public class ServerTest { @@ -18,7 +19,6 @@ env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); env.setProperty(ParameterQueue.SERVER_PORT, "5672"); - // env.setProperty(ParameterQueue.SERIALIZERNAME, - // "omq.common.util.Serializers.KryoImp"); - env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "true"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.kryo); env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false");