Index: /trunk/objectmq/src/omq/client/listener/ResponseListener.java =================================================================== --- /trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 35) +++ /trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 36) @@ -8,10 +8,9 @@ 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,9 +28,9 @@ private static ResponseListener rListener; - private Connection connection; private Channel channel; private QueueingConsumer consumer; private boolean killed = false; private Map> results; + private Properties env; /** @@ -42,26 +41,10 @@ */ 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>(); - Map 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(); - 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,4 +81,18 @@ } 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,4 +101,25 @@ } } + } + + private void startRPCQueue() throws Exception { + channel = Broker.getNewChannel(); + + Map 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(); + 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,8 +176,4 @@ } - public synchronized Channel getChannel() throws Exception { - return connection.createChannel(); - } - /** * @@ -190,5 +204,4 @@ killed = true; channel.close(); - connection.close(); } Index: /trunk/objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 35) +++ /trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 36) @@ -15,4 +15,5 @@ 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,5 +29,4 @@ import com.rabbitmq.client.AMQP.BasicProperties; -import com.rabbitmq.client.Channel; /** @@ -48,5 +48,5 @@ private transient ResponseListener rListener; private transient EventDispatcher dispatcher; - private transient Channel channel; + // private transient Channel channel; private transient Properties env; private transient Map results; @@ -85,5 +85,6 @@ 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,6 +98,6 @@ @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,6 +121,7 @@ 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,15 +130,16 @@ 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,9 +153,11 @@ // 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,6 +226,6 @@ } 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 Index: /trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- /trunk/objectmq/src/omq/common/broker/Broker.java (revision 35) +++ /trunk/objectmq/src/omq/common/broker/Broker.java (revision 36) @@ -14,5 +14,4 @@ import omq.common.util.ParameterQueue; import omq.common.util.Serializer; -import omq.exception.EnvironmentException; import omq.exception.InitBrokerException; import omq.exception.RemoteException; @@ -33,7 +32,5 @@ 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,9 +112,9 @@ 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); } Index: /trunk/objectmq/src/omq/common/event/EventDispatcher.java =================================================================== --- /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 35) +++ /trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 36) @@ -40,4 +40,9 @@ listeners = new HashMap>(); + startEventQueue(); + + } + + private void startEventQueue() throws Exception { // Get a new connection and a new channel channel = Broker.getNewChannel(); @@ -112,4 +117,18 @@ 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); Index: /trunk/objectmq/src/omq/common/util/Environment.java =================================================================== --- /trunk/objectmq/src/omq/common/util/Environment.java (revision 35) +++ /trunk/objectmq/src/omq/common/util/Environment.java (revision 36) @@ -4,5 +4,4 @@ import omq.exception.EnvironmentException; - /** @@ -12,5 +11,5 @@ */ public class Environment { - private static Properties env; + private static Properties env = null; /** @@ -36,3 +35,7 @@ env = environment; } + + public static boolean isVoid() { + return env == null; + } } Index: /trunk/objectmq/test/clientToleranceTest/ClientTest.java =================================================================== --- /trunk/objectmq/test/clientToleranceTest/ClientTest.java (revision 36) +++ /trunk/objectmq/test/clientToleranceTest/ClientTest.java (revision 36) @@ -0,0 +1,67 @@ +package clientToleranceTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; +import org.junit.Test; + +import calculatorTest.Calculator; + +public class ClientTest { + private static Calculator remoteCalc; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "5000"); + + Broker.initBroker(env); + remoteCalc = (Calculator) Broker.lookup("calculator1", Calculator.class); + } + + @Test + public void toleranceTest() throws Exception { + int x = 10; + int y = 20; + int sum = 10 + 20; + + int sync = remoteCalc.add(x, y); + + String password = "unpc"; + String[] command = { "/bin/bash", "-c", "echo " + password + " | sudo -S service rabbitmq-server restart" }; + + Runtime runtime = Runtime.getRuntime(); + runtime.exec(command); + + Thread.sleep(15000); + int resp = remoteCalc.add(x, y); + + assertEquals(sum, sync); + assertEquals(sum, resp); + } + +} Index: /trunk/objectmq/test/clientToleranceTest/ServerTest.java =================================================================== --- /trunk/objectmq/test/clientToleranceTest/ServerTest.java (revision 36) +++ /trunk/objectmq/test/clientToleranceTest/ServerTest.java (revision 36) @@ -0,0 +1,36 @@ +package clientToleranceTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import calculatorTest.CalculatorImpl; + +public class ServerTest { + private static CalculatorImpl calc; + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // 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(); + + Broker.initBroker(env); + Broker.bind("calculator1", calc); + + System.out.println("Server started"); + } +}