Changeset 99 for branches/supervisor

Timestamp:

10/08/13 12:41:19 (11 years ago)

Author:

stoda

Message:

 

Location:

branches/supervisor

Files:

• pom.xml (modified) (1 diff)
• src/main/java/omq/client/proxy/MultiProxymq.java (modified) (1 diff)
• src/main/java/omq/client/proxy/Proxymq.java (modified) (1 diff)
• src/main/java/omq/common/broker/Broker.java (modified) (3 diffs)

branches/supervisor/pom.xml

@@ -4 +4 @@
         <groupId>objectmq</groupId>
         <artifactId>objectmq</artifactId>
-        <version>0.5.5</version>
+        <version>0.6.0</version>
         <name>objectmq</name>
         <description>Middleware based on AMQP</description>

branches/supervisor/src/main/java/omq/client/proxy/MultiProxymq.java

@@ -55 +55 @@
 
                 byte[] bytesRequest = serializer.serialize(serializerType, request);
-                broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
+                broker.publishMessge(exchange, routingkey, props, bytesRequest);
 
                 logger.debug("Proxymq: " + uid + " invokes " + methodName + ", corrID" + corrId + ", exchange: " + exchange + ", replyQueue: " + replyQueueName

branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java

@@ -172 +172 @@
                 // Publish the message
                 byte[] bytesRequest = serializer.serialize(serializerType, request);
-                broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest);
+                broker.publishMessge(exchange, routingkey, props, bytesRequest);
                 logger.debug("Proxymq: " + uid + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange + ", replyQueue: "
                                 + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync()

branches/supervisor/src/main/java/omq/common/broker/Broker.java

@@ -28 +28 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
 import com.rabbitmq.client.ShutdownListener;
@@ -129 +130 @@
          * @throws Exception
          */
-        public Channel getChannel() throws Exception {
+        public synchronized Channel getChannel() throws Exception {
                 return channel;
         }
@@ -139 +140 @@
          * @throws IOException
          */
-        public Channel getNewChannel() throws IOException {
+        public synchronized Channel getNewChannel() throws IOException {
                 return connection.createChannel();
+        }
+
+        /**
+         * Publishes a request using a thread safe channel -if the channel is closed
+         * it tries to open it once
+         * 
+         * @param exchange
+         * @param routingKey
+         * @param props
+         * @param bytesRequest
+         * @throws IOException
+         */
+        public synchronized void publishMessge(String exchange, String routingKey, BasicProperties props, byte[] bytesRequest) throws IOException {
+                if (!channel.isOpen()) {
+                        logger.error("Broker's channel is closed opening a new one", channel.getCloseReason());
+                        channel = getNewChannel();
+                }
+                channel.basicPublish(exchange, routingKey, props, bytesRequest);
         }