Changeset 70 for trunk/src/main

Timestamp:

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

Author:

stoda

Message:

MultiProxymq? added

Location:

trunk/src/main/java/omq

Files:

• client/proxy/MultiProxymq.java (added)
• client/proxy/Proxymq.java (modified) (8 diffs)
• common/broker/Broker.java (modified) (7 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 {