Changeset 24 for trunk

Timestamp:

05/24/13 16:23:16 (12 years ago)

Author:

stoda

Message:

fault tolerance in server if the rabbitmq server falls added

Location:

trunk/objectmq/src/omq

Files:

• common/broker/Broker.java (modified) (5 diffs)
• common/util/OmqConnectionFactory.java (modified) (1 diff)
• common/util/ParameterQueue.java (modified) (2 diffs)
• server/remote/request/RemoteObject.java (modified) (5 diffs)
• ztest/calculator/ClientTest.java (modified) (1 diff)
• ztest/calculator/ServerTest.java (modified) (2 diffs)

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

@@ -23 +23 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.ShutdownListener;
+import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
 
@@ -36 +38 @@
                         Environment.setEnvironment(env);
                         connection = OmqConnectionFactory.getNewConnection(env);
+                        connection.addShutdownListener(new ShutdownListener() {
+                                @Override
+                                public void shutdownCompleted(ShutdownSignalException cause) {
+                                        if (connection.isOpen()) {
+                                                try {
+                                                        connection.close();
+                                                } catch (IOException e) {
+                                                        e.printStackTrace();
+                                                }
+                                        }
+                                        try {
+                                                Properties env = Environment.getEnvironment();
+                                                connection = OmqConnectionFactory.getNewWorkingConnection(env);
+                                                channel = connection.createChannel();
+                                        } catch (Exception e) {
+                                                e.printStackTrace();
+                                        }
+                                }
+                        });
                         channel = connection.createChannel();
                         try {
@@ -84 +105 @@
                 try {
                         Properties environment = Environment.getEnvironment();
-                        remote.start(reference, environment);
+                        remote.startRemoteObject(reference, environment);
                 } catch (Exception e) {
                         throw new RemoteException(e);
@@ -119 +140 @@
                 String message = "ping";
 
-                String exchange = env.getProperty(ParameterQueue.USER_NAME);
-                String queueName = exchange + "ping";
+                String exchange = env.getProperty(ParameterQueue.USER_NAME) + "ping";
+                String queueName = exchange;
                 String routingKey = "routingKey";
 
@@ -140 +161 @@
 
                 if (!message.equalsIgnoreCase(new String(delivery.getBody()))) {
-                        throw new Exception();
+                        throw new IOException("Ping-pong initialitzation has failed");
                 }
         }
+
 }

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

@@ -23 +23 @@
                         connection = getNewConnection(env);
                 }
+        }
+
+        public static Connection getNewWorkingConnection(Properties env) throws Exception {
+                Connection connection = null;
+                boolean working = false;
+
+                while (!working) {
+                        try {
+                                connection = getNewConnection(env);
+                                working = true;
+                        } catch (Exception e) {
+                                e.printStackTrace();
+                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
+                                Thread.sleep(milis);
+                        }
+                }
+
+                return connection;
         }
 

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

@@ -14 +14 @@
         public static String SERIALIZERNAME = "revo.serializer";
         public static String ENABLECOMPRESSION = "revo.compression";
-        
+
         public static String SERVER_HOST = "revo.host";
         public static String SERVER_PORT = "revo.port";
@@ -31 +31 @@
         public static String ENABLE_SSL = "revo.enable_ssl";
         public static String DEBUGFILE = "revo.debug_file";
-        
+
+        public static String RETRY_TIME_CONNECTION = "omq.retry_connection";
 
         /*

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

@@ -35 +35 @@
 
         private String UID;
+        private Properties env;
         private transient RemoteWrapper remoteWrapper;
         private transient Map<String, List<Class<?>>> params;
@@ -56 +57 @@
         }
 
-        public void start(String reference, Properties env) throws Exception {
+        public void startRemoteObject(String reference, Properties env) throws Exception {
                 this.UID = reference;
+                this.env = env;
 
                 params = new HashMap<String, List<Class<?>>>();
@@ -72 +74 @@
                 remoteWrapper = new RemoteWrapper(this, numThreads);
 
-                // Get info about which exchange and queue will use
-                String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
-                String queue = UID;
-                String routingKey = UID;
-
-                // Start channel
-                channel = Broker.getNewChannel();
-
-                // Declares and bindings
-                System.out.println("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue);
-                channel.exchangeDeclare(exchange, "direct");
-                channel.queueDeclare(queue, false, false, false, null);
-                channel.queueBind(queue, exchange, routingKey);
-
-                // Declare the event topic fanout
-                System.out.println("RemoteObject: " + UID + " declaring fanout exchange: " + UID);
-                channel.exchangeDeclare(UID, "fanout");
-
-                // Declare a new consumer
-                consumer = new QueueingConsumer(channel);
-                channel.basicConsume(queue, true, consumer);
+                startQueues();
 
                 // Start this listener
@@ -109 +91 @@
                         } catch (ShutdownSignalException e) {
                                 e.printStackTrace();
+                                try {
+                                        if (channel.isOpen()) {
+                                                channel.close();
+                                        }
+                                        startQueues();
+                                } catch (Exception e1) {
+                                        try {
+                                                long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
+                                                Thread.sleep(milis);
+                                        } catch (InterruptedException e2) {
+                                                // TODO Auto-generated catch block
+                                                e2.printStackTrace();
+                                        }
+                                        e1.printStackTrace();
+                                }
                         } catch (ConsumerCancelledException e) {
                                 e.printStackTrace();
@@ -209 +206 @@
         }
 
+        private void startQueues() throws Exception {
+                // Get info about which exchange and queue will use
+                String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
+                String queue = UID;
+                String routingKey = UID;
+
+                // Start channel
+                channel = Broker.getNewChannel();
+
+                // Declares and bindings
+                System.out.println("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue);
+                channel.exchangeDeclare(exchange, "direct");
+                channel.queueDeclare(queue, false, false, false, null);
+                channel.queueBind(queue, exchange, routingKey);
+
+                // Declare the event topic fanout
+                System.out.println("RemoteObject: " + UID + " declaring fanout exchange: " + UID);
+                channel.exchangeDeclare(UID, "fanout");
+
+                // Declare a new consumer
+                consumer = new QueueingConsumer(channel);
+                channel.basicConsume(queue, true, consumer);
+        }
+
         @Override
         public void addListener(EventListener eventListener) throws Exception {

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

@@ -22 +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,

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

@@ -16 +16 @@
 
                 // Get 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,
@@ -25 +25 @@
                 // Set info about where the message will be sent
                 env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange");
+                env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000");
 
                 calc = new CalculatorImpl();