Changeset 53 for trunk/src/main/java/omq/client

Timestamp:

06/20/13 16:57:39 (11 years ago)

Author:

stoda

Message:

Non static broker
TODO: change all test to see whether the new broker configuration works

Location:

trunk/src/main/java/omq/client

Files:

• listener/ResponseListener.java (modified) (5 diffs)
• proxy/Proxymq.java (modified) (4 diffs)

trunk/src/main/java/omq/client/listener/ResponseListener.java

@@ -28 +28 @@
  */
 public class ResponseListener extends Thread {
-        private static final Logger logger = Logger.getLogger(ResponseListener.class.getName());
-        private static ResponseListener rListener;
+        private final Logger logger = Logger.getLogger(ResponseListener.class.getName());
 
+        private Broker broker;
         private Channel channel;
         private QueueingConsumer consumer;
@@ -43 +43 @@
          * @throws Exception
          */
-        protected ResponseListener(Properties env) throws Exception {
-                this.env = env;
+        public ResponseListener(Broker broker) throws Exception {
+                this.broker = broker;
+                env = broker.getEnvironment();
 
                 // Init the hashtable (it's concurrent)
-                this.results = new Hashtable<String, Map<String, byte[]>>();
+                results = new Hashtable<String, Map<String, byte[]>>();
 
                 startRPCQueue();
@@ -107 +108 @@
 
         private void startRPCQueue() throws Exception {
-                channel = Broker.getNewChannel();
+                channel = broker.getNewChannel();
 
                 Map<String, Object> args = null;
@@ -128 +129 @@
 
         /**
-         * Static function which initializes the ResponseListener
-         * 
-         * @param env
-         * @throws Exception
-         */
-        public static void init(Properties env) throws Exception {
-                if (rListener == null) {
-                        rListener = new ResponseListener(env);
-                        rListener.start();
-                } else {
-                        throw new Exception("Cannot init because it already exists");
-                }
-        }
-
-        /**
-         * Method to retrieve the unique ResponseListener, this function can also
-         * initialize a ResponseListener using and environment
-         * 
-         * @param env
-         * @return unique ResponseListener
-         * @throws Exception
-         */
-        public static ResponseListener getRequestListener(Properties env) throws Exception {
-                if (rListener == null) {
-                        rListener = new ResponseListener(env);
-                        rListener.start();
-                } else {
-                        // TODO: create a new exception to indicate that a response listener
-                        // cannot be init
-                        throw new Exception("Cannot init because it already exists");
-                }
-                return rListener;
-        }
-
-        public static boolean isVoid() {
-                return rListener == null;
-        }
-
-        /**
-         * Method to retrieve the unique ResponseListener
-         * 
-         * @return
-         * @throws Exception
-         */
-        public static ResponseListener getRequestListener() throws Exception {
-                if (rListener == null) {
-                        throw new Exception("Request listener not initialized");
-                }
-                return rListener;
-        }
-
-        /**
          * 
          * @param key
@@ -186 +135 @@
         public boolean containsKey(String key) {
                 return results.containsKey(key);
-        }
-
-        /**
-         * This method is used to kill the unique responseListener in the system
-         * 
-         * @throws Exception
-         */
-        public static void stopResponseListner() throws Exception {
-                rListener.kill();
-                rListener = null;
         }
 

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

@@ -51 +51 @@
         private String uid;
         private transient String serializerType;
+        private transient Broker broker;
         private transient ResponseListener rListener;
         private transient EventDispatcher dispatcher;
+        private transient Serializer serializer;
         // private transient Channel channel;
         private transient Properties env;
@@ -85 +87 @@
          * @throws Exception
          */
-        public Proxymq(String uid, Class<?> clazz, Properties env) throws Exception {
+        public Proxymq(String uid, Class<?> clazz, Broker broker) throws Exception {
                 this.uid = uid;
-                this.rListener = ResponseListener.getRequestListener();
-                this.dispatcher = EventDispatcher.getDispatcher();
+                this.broker = broker;
+                rListener = broker.getResponseListener();
+                dispatcher = broker.getEventDispatcher();
+                serializer = broker.getSerializer();
 
                 // TODO what is better to use a new channel or to use the same?
                 // this.channel = Broker.getChannel();
-                this.env = env;
+                env = broker.getEnvironment();
 
                 // set the serializer type
@@ -160 +164 @@
 
                 // Publish the message
-                byte[] bytesRequest = Serializer.serialize(serializerType, request);
+                byte[] bytesRequest = serializer.serialize(serializerType, request);
                 // TODO See this
                 // channel.basicPublish(exchange, routingkey, props, bytesRequest);
-                Broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
+                broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
                 // Log.saveLog("Client-Serialize", bytesRequest);
         }
@@ -232 +236 @@
                                 throw new TimeoutException("Timeout exception time: " + timeout);
                         }
-                        resp = Serializer.deserializeResponse(results.get(corrId), type);
+                        resp = serializer.deserializeResponse(results.get(corrId), type);
                         // Log.saveLog("Client-Deserialize", results.get(corrId));