Index: /trunk/objectmq/src/omq/Remote.java =================================================================== --- /trunk/objectmq/src/omq/Remote.java (revision 14) +++ /trunk/objectmq/src/omq/Remote.java (revision 15) @@ -3,6 +3,8 @@ import java.io.IOException; import java.io.Serializable; +import java.util.Collection; import omq.common.event.Event; +import omq.common.event.EventListener; import omq.exception.SerializerException; @@ -22,3 +24,9 @@ public void notifyEvent(Event event) throws IOException, SerializerException; + + public void addListener(EventListener eventListener) throws Exception; + + public void removeListener(EventListener eventListener) throws Exception; + + public Collection getListeners() throws Exception; } Index: /trunk/objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 14) +++ /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 15) @@ -5,4 +5,5 @@ import java.lang.reflect.Method; import java.lang.reflect.Proxy; +import java.util.Collection; import java.util.HashMap; import java.util.Hashtable; @@ -15,4 +16,6 @@ import omq.client.remote.response.ResponseListener; import omq.common.event.Event; +import omq.common.event.EventDispatcher; +import omq.common.event.EventListener; import omq.common.message.Request; import omq.common.message.Response; @@ -44,8 +47,9 @@ private String uid; private transient ResponseListener rListener; + private transient EventDispatcher dispatcher; private transient Channel channel; private transient Properties env; private transient Map results; - // private transient Map methodTypes; + private transient Map listeners; private static final Map> primitiveClasses = new HashMap>(); @@ -79,7 +83,10 @@ this.uid = uid; this.rListener = ResponseListener.getRequestListener(); + this.dispatcher = EventDispatcher.getDispatcher(); this.channel = rListener.getChannel(); this.env = env; + + listeners = new HashMap(); // Create a new hashmap and registry it in rListener @@ -97,4 +104,12 @@ if (methodName.equals("getRef")) { return getRef(); + } else if (methodName.equals("addListener")) { + addListener((EventListener) arguments[0]); + return null; + } else if (methodName.equals("removeListener")) { + removeListener((EventListener) arguments[0]); + return null; + } else if (methodName.equals("getListeners")) { + return getListeners(); } } @@ -270,5 +285,24 @@ @Override public void notifyEvent(Event event) throws IOException, SerializerException { - + } + + @Override + public void addListener(EventListener eventListener) throws Exception { + if (eventListener.getTopic() == null) { + eventListener.setTopic(uid); + } + listeners.put(eventListener.getTopic(), eventListener); + dispatcher.addListener(eventListener); + } + + @Override + public void removeListener(EventListener eventListener) throws Exception { + listeners.remove(eventListener.getTopic()); + dispatcher.removeListener(eventListener); + } + + @Override + public Collection getListeners() throws Exception { + return listeners.values(); } Index: /trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- /trunk/objectmq/src/omq/common/broker/Broker.java (revision 14) +++ /trunk/objectmq/src/omq/common/broker/Broker.java (revision 15) @@ -7,4 +7,5 @@ import omq.client.proxy.Proxymq; import omq.client.remote.response.ResponseListener; +import omq.common.event.EventDispatcher; import omq.common.util.Environment; import omq.common.util.OmqConnectionFactory; @@ -19,4 +20,5 @@ private static Connection connection; private static Channel channel; + private static boolean clientStarted = false; public static void initBroker(Properties env) throws Exception { @@ -47,7 +49,9 @@ Properties environment = Environment.getEnvironment(); - if (ResponseListener.isVoid()) { - ResponseListener.init(environment); + if (!clientStarted) { + initClient(environment); + clientStarted = true; } + if (!Proxymq.containsProxy(reference)) { Proxymq proxy = new Proxymq(reference, contract, environment); @@ -79,3 +83,12 @@ } + private static void initClient(Properties environment) throws Exception { + if (ResponseListener.isVoid()) { + ResponseListener.init(environment); + } + if (EventDispatcher.isVoid()) { + EventDispatcher.init(environment); + } + } + } Index: /trunk/objectmq/src/omq/common/event/Event.java =================================================================== --- /trunk/objectmq/src/omq/common/event/Event.java (revision 14) +++ /trunk/objectmq/src/omq/common/event/Event.java (revision 15) @@ -18,7 +18,6 @@ private String topic; - public Event(String corrId, String topic) { + public Event(String corrId) { this.corrId = corrId; - this.topic = topic; } Index: /trunk/objectmq/src/omq/common/event/EventDispatcher.java =================================================================== --- /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 14) +++ /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 15) @@ -161,3 +161,7 @@ } + public static boolean isVoid() { + return dispatcher == null; + } + } Index: /trunk/objectmq/src/omq/server/remote/request/RemoteObject.java =================================================================== --- /trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 14) +++ /trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 15) @@ -4,4 +4,5 @@ import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -12,4 +13,5 @@ import omq.common.broker.Broker; import omq.common.event.Event; +import omq.common.event.EventListener; import omq.common.util.ParameterQueue; import omq.common.util.Serializer; @@ -202,3 +204,16 @@ } + @Override + 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/src/omq/ztest/calculator/Calculator.java =================================================================== --- /trunk/objectmq/src/omq/ztest/calculator/Calculator.java (revision 14) +++ /trunk/objectmq/src/omq/ztest/calculator/Calculator.java (revision 15) @@ -13,4 +13,7 @@ @AsyncMethod public void mult(int x, int y); + + @AsyncMethod + public void divideByZero(); } Index: /trunk/objectmq/src/omq/ztest/calculator/CalculatorImpl.java =================================================================== --- /trunk/objectmq/src/omq/ztest/calculator/CalculatorImpl.java (revision 14) +++ /trunk/objectmq/src/omq/ztest/calculator/CalculatorImpl.java (revision 15) @@ -28,3 +28,7 @@ } + public void divideByZero() { + + } + } Index: /trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java =================================================================== --- /trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java (revision 14) +++ /trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java (revision 15) @@ -12,30 +12,6 @@ public class CalculatorTest { - private static CalculatorImpl calc; private static Calculator remoteCalc; - - @BeforeClass - public static void startServer() throws Exception { - Properties env = new Properties(); - env.setProperty(ParameterQueue.USER_NAME, "guest"); - env.setProperty(ParameterQueue.USER_PASS, "guest"); - - // Get host info of rabbimq (where it is) - env.setProperty(ParameterQueue.SERVER_HOST, "10.30.239.228"); - env.setProperty(ParameterQueue.SERVER_PORT, "5672"); - env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.KryoImp"); - env.setProperty(ParameterQueue.ENABLECOMPRESSION, "true"); - - // Get info about the queue & the exchange where the RemoteListener will - // listen to. - env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); - env.setProperty(ParameterQueue.RPC_QUEUE, "rpc_queue"); - env.setProperty(ParameterQueue.RPC_ROUTING_KEY, "rpc"); - - calc = new CalculatorImpl(); - - Broker.initBroker(env); - Broker.bind(Calculator.class.getSimpleName(), calc); - } + private static Calculator remoteCalc2; @BeforeClass @@ -46,12 +22,11 @@ // Set host info of rabbimq (where it is) - env.setProperty(ParameterQueue.SERVER_HOST, "10.30.239.228"); + 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.ENABLECOMPRESSION, "true"); - + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + // Set info about where the message will be sent env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); - env.setProperty(ParameterQueue.RPC_ROUTING_KEY, "rpc"); // Set info about the queue & the exchange where the ResponseListener @@ -61,5 +36,6 @@ Broker.initBroker(env); - remoteCalc = (Calculator) Broker.lookup(Calculator.class.getSimpleName(), Calculator.class); + remoteCalc = (Calculator) Broker.lookup("calculator1", Calculator.class); + remoteCalc2 = (Calculator) Broker.lookup("calculator2", Calculator.class); } @@ -76,4 +52,15 @@ @Test + public void add2() throws Exception { + int x = 10; + int y = 20; + + int sync = remoteCalc2.add(x, y); + int sum = x + y; + + assertEquals(sum, sync); + } + + @Test public void mult() throws Exception { int x = 5; @@ -82,8 +69,4 @@ remoteCalc.mult(x, y); Thread.sleep(200); - - int mult = x * y; - - assertEquals(mult, calc.getMult()); } } Index: /trunk/objectmq/src/omq/ztest/calculator/ServerTest.java =================================================================== --- /trunk/objectmq/src/omq/ztest/calculator/ServerTest.java (revision 15) +++ /trunk/objectmq/src/omq/ztest/calculator/ServerTest.java (revision 15) @@ -0,0 +1,33 @@ +package omq.ztest.calculator; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +public class ServerTest { + private static CalculatorImpl calc; + private static CalculatorImpl calc2; + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + 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.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + calc = new CalculatorImpl(); + calc2 = new CalculatorImpl(); + + Broker.initBroker(env); + Broker.bind("calculator1", calc); + Broker.bind("calculator2", calc2); + } +} Index: /trunk/objectmq/src/omq/ztest/calculator/ZeroEvent.java =================================================================== --- /trunk/objectmq/src/omq/ztest/calculator/ZeroEvent.java (revision 15) +++ /trunk/objectmq/src/omq/ztest/calculator/ZeroEvent.java (revision 15) @@ -0,0 +1,15 @@ +package omq.ztest.calculator; + +import omq.common.event.Event; + +public class ZeroEvent extends Event { + /** + * + */ + private static final long serialVersionUID = 1L; + + public ZeroEvent(String corrId) { + super(corrId); + } + +}