Changeset 36

Timestamp:

06/11/13 18:25:36 (11 years ago)

Author:

stoda

Message:

Fault tolerance in the client side added.
Some refactoring in proxymq, broker, environment and the client listeners

Location:

trunk/objectmq

Files:

• src/omq/client/listener/ResponseListener.java (modified) (7 diffs)
• src/omq/client/proxy/Proxymq.java (modified) (9 diffs)
• src/omq/common/broker/Broker.java (modified) (3 diffs)
• src/omq/common/event/EventDispatcher.java (modified) (2 diffs)
• src/omq/common/util/Environment.java (modified) (3 diffs)
• test/clientToleranceTest (added)
• test/clientToleranceTest/ClientTest.java (added)
• test/clientToleranceTest/ServerTest.java (added)

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

@@ -8 +8 @@
 
 import omq.client.proxy.Proxymq;
-import omq.common.util.OmqConnectionFactory;
+import omq.common.broker.Broker;
 import omq.common.util.ParameterQueue;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConsumerCancelledException;
 import com.rabbitmq.client.QueueingConsumer;
@@ -29 +28 @@
         private static ResponseListener rListener;
 
-        private Connection connection;
         private Channel channel;
         private QueueingConsumer consumer;
         private boolean killed = false;
         private Map<String, Map<String, byte[]>> results;
+        private Properties env;
 
         /**
@@ -42 +41 @@
          */
         protected ResponseListener(Properties env) throws Exception {
-                connection = OmqConnectionFactory.getNewConnection(env);
-                channel = connection.createChannel();
+                this.env = env;
 
                 // Init the hashtable (it's concurrent)
                 this.results = new Hashtable<String, Map<String, byte[]>>();
 
-                Map<String, Object> args = null;
-
-                String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
-                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
-
-                int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1"));
-                if (ttl > 0) {
-                        args = new HashMap<String, Object>();
-                        args.put("x-message-ttl", ttl);
-                }
-
-                channel.queueDeclare(reply_queue, durable, false, false, args);
-
-                // Declare a new consumer
-                consumer = new QueueingConsumer(channel);
-                channel.basicConsume(reply_queue, true, consumer);
+                startRPCQueue();
         }
 
@@ -98 +81 @@
                         } catch (ShutdownSignalException e) {
                                 e.printStackTrace();
+                                try {
+                                        if (channel.isOpen()) {
+                                                channel.close();
+                                        }
+                                        startRPCQueue();
+                                } catch (Exception e1) {
+                                        e1.printStackTrace();
+                                        try {
+                                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
+                                                Thread.sleep(milis);
+                                        } catch (InterruptedException e2) {
+                                                e2.printStackTrace();
+                                        }
+                                }
                         } catch (ConsumerCancelledException e) {
                                 e.printStackTrace();
@@ -104 +101 @@
                         }
                 }
+        }
+
+        private void startRPCQueue() throws Exception {
+                channel = Broker.getNewChannel();
+
+                Map<String, Object> args = null;
+
+                String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
+                boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));
+
+                int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1"));
+                if (ttl > 0) {
+                        args = new HashMap<String, Object>();
+                        args.put("x-message-ttl", ttl);
+                }
+
+                channel.queueDeclare(reply_queue, durable, false, false, args);
+
+                // Declare a new consumer
+                consumer = new QueueingConsumer(channel);
+                channel.basicConsume(reply_queue, true, consumer);
         }
 
@@ -158 +176 @@
         }
 
-        public synchronized Channel getChannel() throws Exception {
-                return connection.createChannel();
-        }
-
         /**
          * 
@@ -190 +204 @@
                 killed = true;
                 channel.close();
-                connection.close();
         }
 

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

@@ -15 +15 @@
 import omq.client.annotation.SyncMethod;
 import omq.client.listener.ResponseListener;
+import omq.common.broker.Broker;
 import omq.common.event.Event;
 import omq.common.event.EventDispatcher;
@@ -28 +29 @@
 
 import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.client.Channel;
 
 /**
@@ -48 +48 @@
         private transient ResponseListener rListener;
         private transient EventDispatcher dispatcher;
-        private transient Channel channel;
+        // private transient Channel channel;
         private transient Properties env;
         private transient Map<String, byte[]> results;
@@ -85 +85 @@
                 this.dispatcher = EventDispatcher.getDispatcher();
 
-                this.channel = rListener.getChannel();
+                // TODO what is better to use a new channel or to use the same?
+                // this.channel = Broker.getChannel();
                 this.env = env;
 
@@ -97 +98 @@
         @Override
         public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
-                //long timeStart = (new Date()).getTime();
-                
+                // long timeStart = (new Date()).getTime();
+
                 // Local methods only
                 String methodName = method.getName();
@@ -120 +121 @@
                 Request request = createRequest(method, arguments);
 
-                //Log.saveTimeSendRequestLog("Client-time-request", request.getId(), method.getName(), timeStart);
-                
+                // Log.saveTimeSendRequestLog("Client-time-request", request.getId(),
+                // method.getName(), timeStart);
+
                 Object response = null;
                 // Publish the request
@@ -128 +130 @@
                         publishAsyncRequest(request);
                 } else {
-                        System.out.println("Publish sync request -> " + request.getId());                       
+                        System.out.println("Publish sync request -> " + request.getId());
                         response = publishSyncRequest(request, method.getReturnType());
-                        
-                        //long timeEnd = (new Date()).getTime();
-                        //Log.saveTimeSendRequestLog("Client-time-response", request.getId(), method.getName(), timeEnd);
-                }
-                
+
+                        // long timeEnd = (new Date()).getTime();
+                        // Log.saveTimeSendRequestLog("Client-time-response",
+                        // request.getId(), method.getName(), timeEnd);
+                }
+
                 return response;
         }
 
-        private void publishMessage(Request request, String replyQueueName) throws IOException, SerializerException {
+        private void publishMessage(Request request, String replyQueueName) throws Exception {
                 String corrId = request.getId();
 
@@ -150 +153 @@
                 // Publish the message
                 byte[] bytesRequest = Serializer.serialize(request);
-                channel.basicPublish(exchange, routingkey, props, bytesRequest);
-                //Log.saveLog("Client-Serialize", bytesRequest);
-        }
-
-        private void publishAsyncRequest(Request request) throws IOException, SerializerException {
+                // TODO See this
+                // channel.basicPublish(exchange, routingkey, props, bytesRequest);
+                Broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
+                // Log.saveLog("Client-Serialize", bytesRequest);
+        }
+
+        private void publishAsyncRequest(Request request) throws Exception {
                 // Get the environment properties
                 String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
@@ -221 +226 @@
                         }
                         resp = Serializer.deserializeResponse(results.get(corrId), type);
-                        //Log.saveLog("Client-Deserialize", results.get(corrId));
-                        
+                        // Log.saveLog("Client-Deserialize", results.get(corrId));
+
                         // Remove and indicate the key exists (a hashmap can contain a null
                         // object, using this we'll know whether a response has been

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

@@ -14 +14 @@
 import omq.common.util.ParameterQueue;
 import omq.common.util.Serializer;
-import omq.exception.EnvironmentException;
 import omq.exception.InitBrokerException;
 import omq.exception.RemoteException;
@@ -33 +32 @@
 
         public static void initBroker(Properties env) throws Exception {
-                try {
-                        Environment.getEnvironment();
-                } catch (EnvironmentException ex) { // environment not set.
+                if (Environment.isVoid()) {
                         Environment.setEnvironment(env);
                         connection = OmqConnectionFactory.getNewConnection(env);
@@ -115 +112 @@
                 System.out.println("EventTrigger fanout exchange: " + UID + " Event topic: " + UID + " Event corrID: " + event.getCorrId());
                 channel.exchangeDeclare(UID, "fanout");
-                
+
                 byte[] bytesResponse = Serializer.serialize(wrapper);
                 channel.basicPublish(UID, "", null, bytesResponse);
 
-                //Log.saveLog("Server-Serialize", bytesResponse);                       
+                // Log.saveLog("Server-Serialize", bytesResponse);
         }
 

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

@@ -40 +40 @@
                 listeners = new HashMap<String, Vector<EventListener>>();
 
+                startEventQueue();
+
+        }
+
+        private void startEventQueue() throws Exception {
                 // Get a new connection and a new channel
                 channel = Broker.getNewChannel();
@@ -112 +117 @@
                                 System.out.println("ShutdownSignalException e: " + e);
                                 e.printStackTrace();
+                                try {
+                                        if (channel.isOpen()) {
+                                                channel.close();
+                                        }
+                                        startEventQueue();
+                                } catch (Exception e1) {
+                                        try {
+                                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
+                                                Thread.sleep(milis);
+                                        } catch (InterruptedException e2) {
+                                                e2.printStackTrace();
+                                        }
+                                        e1.printStackTrace();
+                                }
                         } catch (ConsumerCancelledException e) {
                                 System.out.println("ConsumerCancelledException e: " + e);

trunk/objectmq/src/omq/common/util/Environment.java

@@ -4 +4 @@
 
 import omq.exception.EnvironmentException;
-
 
 /**
@@ -12 +11 @@
  */
 public class Environment {
-        private static Properties env;
+        private static Properties env = null;
 
         /**
@@ -36 +35 @@
                 env = environment;
         }
+
+        public static boolean isVoid() {
+                return env == null;
+        }
 }