Changeset 70 for trunk/src

Timestamp:

06/28/13 12:41:11 (11 years ago)

Author:

stoda

Message:

MultiProxymq? added

Location:

trunk/src

Files:

• main/java/omq/client/proxy/MultiProxymq.java (added)
• main/java/omq/client/proxy/Proxymq.java (modified) (8 diffs)
• main/java/omq/common/broker/Broker.java (modified) (7 diffs)
• test/java/omq/test/event/EventTrigger.java (deleted)
• test/java/omq/test/event/EventTriggerImpl.java (deleted)
• test/java/omq/test/event/Message.java (added)
• test/java/omq/test/event/MessageEvent.java (deleted)
• test/java/omq/test/event/MessageImpl.java (added)
• test/java/omq/test/event/MessageListener.java (deleted)
• test/java/omq/test/event/MessageTest.java (moved) (moved from trunk/src/test/java/omq/test/event/EventTest.java) (3 diffs)

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

@@ -5 +5 @@
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
 import java.util.Collection;
 import java.util.HashMap;
@@ -50 +49 @@
 
         private String uid;
+        private transient String exchange;
+        private transient String multiExchange;
+        private transient String replyQueueName;
         private transient String serializerType;
         private transient Broker broker;
@@ -96 +98 @@
                 // this.channel = Broker.getChannel();
                 env = broker.getEnvironment();
+                exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                multiExchange = multi + exchange;
+                replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
 
                 // set the serializer type
@@ -134 +139 @@
                 if (request.isAsync()) {
                         logger.debug("Publish async request -> " + request.getId());
-                        publishAsyncRequest(request);
+                        publishMessage(request, replyQueueName);
                 } else {
                         logger.debug("Publish sync request -> " + request.getId());
@@ -151 +156 @@
 
                 if (request.isMulti()) {
-                        exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                        exchange = multiExchange;
                         routingkey = "";
                 } else {
-                        exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                        exchange = this.exchange;
                         routingkey = uid;
                 }
@@ -166 +171 @@
         }
 
-        private void publishAsyncRequest(Request request) throws Exception {
-                // Get the environment properties
-                String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
-                publishMessage(request, replyQueueName);
-        }
-
         private Object publishSyncRequest(Request request, Class<?> type) throws Exception {
                 String corrId = request.getId();
@@ -177 +176 @@
                 int retries = request.getRetries();
                 long timeout = request.getTimeout();
-
-                // Get the environment properties
-                String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
 
                 // Publish the message
@@ -296 +292 @@
 
         /**
-         * Returns an instance of a proxy class for the specified interfaces that
-         * dispatches method invocations to the specified invocation handler. * @param
-         * loader
-         * 
-         * @param loader
-         *            the class loader to define the proxy class
-         * 
-         * @param interfaces
-         *            the list of interfaces for the proxy class to implement
-         * @param proxy
-         *            the invocation handler to dispatch method invocations to
-         * @return a proxy instance with the specified invocation handler of a proxy
-         *         class that is defined by the specified class loader and that
-         *         implements the specified interfaces
-         */
-        public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, Proxymq proxy) {
-                return Proxy.newProxyInstance(loader, interfaces, proxy);
-        }
-
-        /**
          * Gets the Map used internally to retreive the response of the server
          * 

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

@@ -2 +2 @@
 
 import java.io.IOException;
+import java.lang.reflect.Proxy;
 import java.net.URL;
 import java.util.HashMap;
@@ -10 +11 @@
 import omq.Remote;
 import omq.client.listener.ResponseListener;
+import omq.client.proxy.MultiProxymq;
 import omq.client.proxy.Proxymq;
 import omq.common.event.Event;
@@ -46 +48 @@
         private Map<String, RemoteObject> remoteObjs;
         private Map<String, Object> proxies = new Hashtable<String, Object>();
+        private Map<String, Object> multiProxies = new Hashtable<String, Object>();
 
         public Broker(Properties env) throws Exception {
@@ -73 +76 @@
                         responseListener.kill();
                         eventDispatcher.kill();
-                        //TODO proxies = null; ??
+                        // TODO proxies = null; ??
                 }
                 // Stop all the remote objects working
@@ -136 +139 @@
                                 Proxymq proxy = new Proxymq(reference, contract, this);
                                 Class<?>[] array = { contract };
-                                Object newProxy = Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy);
+                                Object newProxy = Proxy.newProxyInstance(contract.getClassLoader(), array, proxy);
                                 proxies.put(reference, newProxy);
                                 return (T) newProxy;
@@ -147 +150 @@
         }
 
+        @SuppressWarnings("unchecked")
+        public synchronized <T extends Remote> T lookupMulti(String reference, Class<T> contract) throws RemoteException {
+                try {
+                        if (!multiProxies.containsKey(reference)) {
+                                MultiProxymq proxy = new MultiProxymq(reference, contract, this);
+                                Class<?>[] array = { contract };
+                                Object newProxy = Proxy.newProxyInstance(contract.getClassLoader(), array, proxy);
+                                multiProxies.put(reference, newProxy);
+                                return (T) newProxy;
+                        }
+                        return (T) multiProxies.get(reference);
+
+                } catch (Exception e) {
+                        throw new RemoteException(e);
+                }
+        }
+
         public void bind(String reference, RemoteObject remote) throws RemoteException {
                 try {
@@ -155 +175 @@
                 }
         }
-        
+
         public void startTriggerEvent(String reference, RemoteObject remote) throws RemoteException {
                 try {

trunk/src/test/java/omq/test/event/MessageTest.java

@@ -3 +3 @@
 import static org.junit.Assert.assertEquals;
 
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Properties;
 
@@ -9 +11 @@
 import omq.common.util.Serializer;
 
+import org.junit.After;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
 
-public class EventTest {
-        private static EventTriggerImpl trigger;
+@RunWith(value = Parameterized.class)
+public class MessageTest {
+        private static final String NAME = "message";
+
+        private static Broker broker;
+        private static Message serverProxy;
+        private static MessageImpl clientM1;
+        private static MessageImpl clientM2;
+
+        public MessageTest(String type) throws Exception {
+                Properties env = new Properties();
+                env.setProperty(ParameterQueue.USER_NAME, "guest");
+                env.setProperty(ParameterQueue.USER_PASS, "guest");
+
+                // Set 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.DURABLE_QUEUES, "false");
+                env.setProperty(ParameterQueue.SERIALIZER_NAME, type);
+                env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false");
+
+                // Set info about where the message will be sent
+                env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange");
+
+                // Set info about the queue & the exchange where the ResponseListener
+                // will listen to.
+                env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue");
+                env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue");
+
+                broker = new Broker(env);
+                serverProxy = broker.lookupMulti(NAME, Message.class);
+        }
+
+        @Parameters
+        public static Collection<Object[]> data() {
+                Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } };
+                return Arrays.asList(data);
+        }
 
         @BeforeClass
-        public static void server() throws Exception {
+        public static void client() throws Exception {
                 Properties env = new Properties();
                 env.setProperty(ParameterQueue.USER_NAME, "guest");
@@ -31 +73 @@
                 env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000");
 
-                trigger = new EventTriggerImpl();
                 Broker broker = new Broker(env);
+                clientM1 = new MessageImpl();
+                broker.bind(NAME, clientM1);
 
-                broker.startTriggerEvent("trigger", trigger);
+                Broker broker2 = new Broker(env);
+                clientM2 = new MessageImpl();
+                broker2.bind(NAME, clientM2);
+        }
 
-                System.out.println("Server started");
+        @After
+        public void stop() throws Exception {
+                broker.stopBroker();
         }
 
         @Test
-        public void eventTest() throws Exception {
-                String expected = "This is an event";
-                String actual = null;
+        public void test() throws Exception {
+                String expected = "Hello";
 
-                Properties env = new Properties();
-                env.setProperty(ParameterQueue.USER_NAME, "guest");
-                env.setProperty(ParameterQueue.USER_PASS, "guest");
+                serverProxy.setMessage(expected);
+                Thread.sleep(200);
+                assertEquals(expected, clientM1.getMessage());
+                assertEquals(expected, clientM2.getMessage());
+                serverProxy.setMessage("");
+                Thread.sleep(200);
+        }
 
-                // Set 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.DURABLE_QUEUES, "false");
-                env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA);
-                env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false");
-
-                // Set info about where the message will be sent
-                env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange");
-                // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug");
-
-                // Set info about the queue & the exchange where the ResponseListener
-                // will listen to.
-                env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue");
-                env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue");
-
-                Broker broker = new Broker(env);
-                EventTrigger et = broker.lookup("trigger", EventTrigger.class);
-
-                MessageListener ms = new MessageListener();
-                et.addListener(ms);
-                trigger.triggerEvent(expected);
-                Thread.sleep(500);
-                actual = ms.getMessage();
-
-                assertEquals(expected, actual);
-        }
 }