Index: trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- trunk/objectmq/src/omq/common/broker/Broker.java (revision 23) +++ trunk/objectmq/src/omq/common/broker/Broker.java (revision 24) @@ -23,4 +23,6 @@ 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,4 +38,23 @@ 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,5 +105,5 @@ try { Properties environment = Environment.getEnvironment(); - remote.start(reference, environment); + remote.startRemoteObject(reference, environment); } catch (Exception e) { throw new RemoteException(e); @@ -119,6 +140,6 @@ 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,6 +161,7 @@ if (!message.equalsIgnoreCase(new String(delivery.getBody()))) { - throw new Exception(); + throw new IOException("Ping-pong initialitzation has failed"); } } + } Index: trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java =================================================================== --- trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java (revision 23) +++ trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java (revision 24) @@ -23,4 +23,22 @@ 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; } Index: trunk/objectmq/src/omq/common/util/ParameterQueue.java =================================================================== --- trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 23) +++ trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 24) @@ -14,5 +14,5 @@ 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,5 +31,6 @@ public static String ENABLE_SSL = "revo.enable_ssl"; public static String DEBUGFILE = "revo.debug_file"; - + + public static String RETRY_TIME_CONNECTION = "omq.retry_connection"; /* Index: trunk/objectmq/src/omq/server/remote/request/RemoteObject.java =================================================================== --- trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 23) +++ trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 24) @@ -35,4 +35,5 @@ private String UID; + private Properties env; private transient RemoteWrapper remoteWrapper; private transient Map>> params; @@ -56,6 +57,7 @@ } - 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>>(); @@ -72,25 +74,5 @@ 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,4 +91,19 @@ } 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,4 +206,28 @@ } + 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 { Index: trunk/objectmq/src/omq/ztest/calculator/ClientTest.java =================================================================== --- trunk/objectmq/src/omq/ztest/calculator/ClientTest.java (revision 23) +++ trunk/objectmq/src/omq/ztest/calculator/ClientTest.java (revision 24) @@ -22,5 +22,5 @@ // 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, Index: trunk/objectmq/src/omq/ztest/calculator/ServerTest.java =================================================================== --- trunk/objectmq/src/omq/ztest/calculator/ServerTest.java (revision 23) +++ trunk/objectmq/src/omq/ztest/calculator/ServerTest.java (revision 24) @@ -16,5 +16,5 @@ // 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,4 +25,5 @@ // 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();