Changeset 15 for trunk

Timestamp:

05/20/13 19:45:19 (11 years ago)

Author:

stoda

Message:

Listeners added (not tested)
little test to shut up Vilella

Location:

trunk/objectmq/src/omq

Files:

• Remote.java (modified) (2 diffs)
• client/proxy/Proxymq.java (modified) (6 diffs)
• common/broker/Broker.java (modified) (4 diffs)
• common/event/Event.java (modified) (1 diff)
• common/event/EventDispatcher.java (modified) (1 diff)
• server/remote/request/RemoteObject.java (modified) (3 diffs)
• ztest/calculator/Calculator.java (modified) (1 diff)
• ztest/calculator/CalculatorImpl.java (modified) (1 diff)
• ztest/calculator/CalculatorTest.java (modified) (5 diffs)
• ztest/calculator/ServerTest.java (added)
• ztest/calculator/ZeroEvent.java (added)

trunk/objectmq/src/omq/Remote.java

@@ -3 +3 @@
 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 +24 @@
 
         public void notifyEvent(Event event) throws IOException, SerializerException;
+
+        public void addListener(EventListener eventListener) throws Exception;
+
+        public void removeListener(EventListener eventListener) throws Exception;
+
+        public Collection<EventListener> getListeners() throws Exception;
 }

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

@@ -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 +16 @@
 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 +47 @@
         private String uid;
         private transient ResponseListener rListener;
+        private transient EventDispatcher dispatcher;
         private transient Channel channel;
         private transient Properties env;
         private transient Map<String, byte[]> results;
-        // private transient Map<Method, CallType> methodTypes;
+        private transient Map<String, EventListener> listeners;
 
         private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();
@@ -79 +83 @@
                 this.uid = uid;
                 this.rListener = ResponseListener.getRequestListener();
+                this.dispatcher = EventDispatcher.getDispatcher();
 
                 this.channel = rListener.getChannel();
                 this.env = env;
+
+                listeners = new HashMap<String, EventListener>();
 
                 // Create a new hashmap and registry it in rListener
@@ -97 +104 @@
                         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 +285 @@
         @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<EventListener> getListeners() throws Exception {
+                return listeners.values();
         }
 

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

@@ -7 +7 @@
 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 +20 @@
         private static Connection connection;
         private static Channel channel;
+        private static boolean clientStarted = false;
 
         public static void initBroker(Properties env) throws Exception {
@@ -47 +49 @@
                         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 +83 @@
         }
 
+        private static void initClient(Properties environment) throws Exception {
+                if (ResponseListener.isVoid()) {
+                        ResponseListener.init(environment);
+                }
+                if (EventDispatcher.isVoid()) {
+                        EventDispatcher.init(environment);
+                }
+        }
+
 }

trunk/objectmq/src/omq/common/event/Event.java

@@ -18 +18 @@
         private String topic;
 
-        public Event(String corrId, String topic) {
+        public Event(String corrId) {
                 this.corrId = corrId;
-                this.topic = topic;
         }
 

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

@@ -161 +161 @@
         }
 
+        public static boolean isVoid() {
+                return dispatcher == null;
+        }
+
 }

trunk/objectmq/src/omq/server/remote/request/RemoteObject.java

@@ -4 +4 @@
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -12 +13 @@
 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 +204 @@
         }
 
+        @Override
+        public void addListener(EventListener eventListener) throws Exception {
+        }
+
+        @Override
+        public void removeListener(EventListener eventListener) throws Exception {
+        }
+
+        @Override
+        public Collection<EventListener> getListeners() throws Exception {
+                return null;
+        }
+
 }

trunk/objectmq/src/omq/ztest/calculator/Calculator.java

@@ -13 +13 @@
         @AsyncMethod
         public void mult(int x, int y);
+        
+        @AsyncMethod
+        public void divideByZero();
 
 }

trunk/objectmq/src/omq/ztest/calculator/CalculatorImpl.java

@@ -28 +28 @@
         }
 
+        public void divideByZero() {
+                
+        }
+
 }

trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java

@@ -12 +12 @@
 
 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 +22 @@
 
                 // 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 +36 @@
 
                 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 +52 @@
 
         @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 +69 @@
                 remoteCalc.mult(x, y);
                 Thread.sleep(200);
-
-                int mult = x * y;
-
-                assertEquals(mult, calc.getMult());
         }
 }