camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ningji...@apache.org
Subject [2/4] git commit: Reconnect RabbitMQ Producer/Consumer after broker was unavailable
Date Fri, 06 Jun 2014 05:02:55 GMT
Reconnect RabbitMQ Producer/Consumer after broker was unavailable


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1191e19f
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1191e19f
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1191e19f

Branch: refs/heads/master
Commit: 1191e19fc194b545931530011511a55a8d6d6431
Parents: 0762be7
Author: Gerald Quintana <gerald.quintana@zenika.com>
Authored: Mon May 5 15:47:01 2014 +0200
Committer: Gerald Quintana <gerald.quintana@zenika.com>
Committed: Mon May 5 15:47:01 2014 +0200

----------------------------------------------------------------------
 .../component/rabbitmq/RabbitMQConsumer.java    | 112 +++++++++++++++----
 .../component/rabbitmq/RabbitMQEndpoint.java    |  22 ++++
 .../component/rabbitmq/RabbitMQProducer.java    |  33 +++++-
 .../rabbitmq/RabbitMQEndpointTest.java          |  30 +++--
 .../rabbitmq/RabbitMQReConnectionIntTest.java   |  77 +++++++++++++
 5 files changed, 232 insertions(+), 42 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/1191e19f/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
index 501f2a1..0bdc93b 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
@@ -17,8 +17,9 @@
 package org.apache.camel.component.rabbitmq;
 
 import java.io.IOException;
-import java.util.HashMap;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
@@ -36,6 +37,10 @@ public class RabbitMQConsumer extends DefaultConsumer {
     private int closeTimeout = 30 * 1000;
     
     private final RabbitMQEndpoint endpoint;
+    /**
+     * Task in charge of starting consumer
+     */
+    private StartConsumerCallable startConsumerCallable;
 
     public RabbitMQConsumer(RabbitMQEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -43,37 +48,56 @@ public class RabbitMQConsumer extends DefaultConsumer {
     }
 
     @Override
-    protected void doStart() throws Exception {
-        executor = endpoint.createExecutor();
-        log.debug("Using executor {}", executor);
-
-        conn = endpoint.connect(executor);
-        log.debug("Using conn {}", conn);
-
-        channel = conn.createChannel();
-        log.debug("Using channel {}", channel);
+    public RabbitMQEndpoint getEndpoint() {
+        return (RabbitMQEndpoint) super.getEndpoint();
+    }
 
-        channel.exchangeDeclare(endpoint.getExchangeName(),
-                endpoint.getExchangeType(),
-                endpoint.isDurable(),
-                endpoint.isAutoDelete(),
-                new HashMap<String, Object>());
+    /**
+     * Open connection and channel
+     */
+    private void openConnectionAndChannel() throws IOException {
+        log.trace("Creating connection...");
+        this.conn = getEndpoint().connect(executor);
+        log.debug("Created connection: {}", conn);
+
+        log.trace("Creating channel...");
+        this.channel = conn.createChannel();
+        log.debug("Created channel: {}", channel);
+    }
 
-        // need to make sure the queueDeclare is same with the exchange declare
-        channel.queueDeclare(endpoint.getQueue(), endpoint.isDurable(), false,
-                endpoint.isAutoDelete(), null);
-        channel.queueBind(
-                endpoint.getQueue(),
-                endpoint.getExchangeName(),
-                endpoint.getRoutingKey() == null ? "" : endpoint
-                        .getRoutingKey());
+    /**
+     * If needed, create Exchange and Queue, then add message listener
+     */
+    private void addConsumer() throws IOException {
+        getEndpoint().declareExchangeAndQueue(channel);
 
         channel.basicConsume(endpoint.getQueue(), endpoint.isAutoAck(),
                 new RabbitConsumer(this, channel));
     }
 
     @Override
-    protected void doStop() throws Exception {
+    protected void doStart() throws Exception {
+        executor = endpoint.createExecutor();
+        log.debug("Using executor {}", executor);
+        try {
+            openConnectionAndChannel();
+            addConsumer();
+        } catch (Exception e) {
+            // Open connection, and start message listener in background
+            Integer networkRecoveryInterval = getEndpoint().getNetworkRecoveryInterval();
+            final long connectionRetryInterval= networkRecoveryInterval!=null &&
networkRecoveryInterval >0? networkRecoveryInterval :100L;
+            startConsumerCallable=new StartConsumerCallable(connectionRetryInterval);
+            executor.submit(startConsumerCallable);
+        }
+    }
+
+    /**
+     * If needed, close Connection and Channel
+     */
+    private void closeConnectionAndChannel() throws IOException {
+        if (startConsumerCallable!=null) {
+            startConsumerCallable.stop();
+        }
         if (channel != null) {
             log.debug("Closing channel: {}", channel);
             channel.close();
@@ -84,6 +108,12 @@ public class RabbitMQConsumer extends DefaultConsumer {
             conn.close(closeTimeout);
             conn = null;
         }
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        closeConnectionAndChannel();
+
         if (executor != null) {
             if (endpoint != null && endpoint.getCamelContext() != null) {
                 endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(executor);
@@ -178,4 +208,38 @@ public class RabbitMQConsumer extends DefaultConsumer {
 
     }
 
+    /**
+     * Task in charge of opening connection and adding listener when consumer is started
+     * and broker is not avaiblable.
+     */
+    private class StartConsumerCallable implements Callable<Void> {
+        private final long connectionRetryInterval;
+        private final AtomicBoolean running=new AtomicBoolean(true);
+        public StartConsumerCallable(long connectionRetryInterval) {
+            this.connectionRetryInterval = connectionRetryInterval;
+        }
+        public void stop() {
+            running.set(false);
+            RabbitMQConsumer.this.startConsumerCallable=null;
+        }
+        @Override
+        public Void call() throws Exception {
+            boolean connectionFailed=true;
+            // Reconnection loop
+            while (running.get() && connectionFailed) {
+                try {
+                    openConnectionAndChannel();
+                    connectionFailed=false;
+                } catch (Exception e) {
+                    log.debug("Connection failed, will retry in "+connectionRetryInterval+"ms",
e);
+                    Thread.sleep(connectionRetryInterval);
+                }
+            }
+            if (!connectionFailed) {
+                addConsumer();
+            }
+            stop();
+            return null;
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/1191e19f/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index 646a633..7f87f53 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -29,6 +30,7 @@ import javax.net.ssl.TrustManager;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Envelope;
@@ -126,6 +128,26 @@ public class RabbitMQEndpoint extends DefaultEndpoint {
         }
     }
 
+	/**
+     * If needed, declare Exchange, declare Queue and bind them with Routing Key
+     */
+    public void declareExchangeAndQueue(Channel channel) throws IOException {
+        channel.exchangeDeclare(getExchangeName(),
+                getExchangeType(),
+                isDurable(),
+                isAutoDelete(),
+                new HashMap<String, Object>());
+        if (getQueue()!=null) {
+            // need to make sure the queueDeclare is same with the exchange declare
+            channel.queueDeclare(getQueue(), isDurable(), false,
+                    isAutoDelete(), null);
+            channel.queueBind(
+                    getQueue(),
+                    getExchangeName(),
+                    getRoutingKey() == null ? "" : getRoutingKey());
+        }
+    }
+
     private ConnectionFactory getOrCreateConnectionFactory() {
         if (connectionFactory == null) {
             ConnectionFactory factory = new ConnectionFactory();

http://git-wip-us.apache.org/repos/asf/camel/blob/1191e19f/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
index 7763423..03177e2 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
@@ -45,11 +45,10 @@ public class RabbitMQProducer extends DefaultProducer {
     public RabbitMQEndpoint getEndpoint() {
         return (RabbitMQEndpoint) super.getEndpoint();
     }
-
-    @Override
-    protected void doStart() throws Exception {
-        this.executorService = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
"CamelRabbitMQProducer[" + getEndpoint().getQueue() + "]");
-
+    /**
+     * Open connection and channel
+     */
+    private void openConnectionAndChannel() throws IOException {
         log.trace("Creating connection...");
         this.conn = getEndpoint().connect(executorService);
         log.debug("Created connection: {}", conn);
@@ -60,7 +59,20 @@ public class RabbitMQProducer extends DefaultProducer {
     }
 
     @Override
-    protected void doStop() throws Exception {
+    protected void doStart() throws Exception {
+        this.executorService = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
"CamelRabbitMQProducer[" + getEndpoint().getQueue() + "]");
+
+        try {
+            openConnectionAndChannel();
+        } catch (IOException e) {
+            log.warn("Failed to create connection", e);
+        }
+    }
+
+    /**
+     * If needed, close Connection and Channel
+     */
+    private void closeConnectionAndChannel() throws IOException {
         if (channel != null) {
             log.debug("Closing channel: {}", channel);
             channel.close();
@@ -71,6 +83,11 @@ public class RabbitMQProducer extends DefaultProducer {
             conn.close(closeTimeout);
             conn = null;
         }
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        closeConnectionAndChannel();
         if (executorService != null) {
             getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
             executorService = null;
@@ -95,6 +112,10 @@ public class RabbitMQProducer extends DefaultProducer {
         byte[] messageBodyBytes = exchange.getIn().getMandatoryBody(byte[].class);
         AMQP.BasicProperties.Builder properties = buildProperties(exchange);
 
+        if (channel==null) {
+            // Open connection and channel lazily
+            openConnectionAndChannel();
+        }
         channel.basicPublish(exchangeName, key, properties.build(), messageBodyBytes);
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/1191e19f/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
index 11b3675..2db5005 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.rabbitmq;
 
+import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.Date;
 import java.util.HashMap;
@@ -26,7 +27,6 @@ import java.util.concurrent.ThreadPoolExecutor;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
-import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.impl.LongStringHelper;
@@ -132,7 +132,13 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
     }
 
     private ConnectionFactory createConnectionFactory(String uri) {
-        RabbitMQEndpoint endpoint = context.getEndpoint(uri, RabbitMQEndpoint.class); 
+        RabbitMQEndpoint endpoint = context.getEndpoint(uri, RabbitMQEndpoint.class);
+		try {
+			endpoint.connect(Executors.newSingleThreadExecutor());
+		} catch(IOException ioExc) {
+			// Doesn't matter if RabbitMQ is not available
+			log.debug("RabbitMQ not available", ioExc);
+		}
         return endpoint.getConnectionFactory();
     }
 
@@ -158,16 +164,16 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
     @Test
     public void testCreateConnectionFactoryCustom() throws Exception {
         ConnectionFactory connectionFactory  = createConnectionFactory("rabbitmq:localhost:1234/exchange"
-                                                                      + "?username=userxxx"
-                                                                      + "&password=passxxx"
-                                                                      + "&connectionTimeout=123"
-                                                                      + "&requestedChannelMax=456"
-                                                                      + "&requestedFrameMax=789"
-                                                                      + "&requestedHeartbeat=987"
-                                                                      + "&sslProtocol=true"
-                                                                      + "&automaticRecoveryEnabled=true"
-                                                                      + "&networkRecoveryInterval=654"
-                                                                      + "&topologyRecoveryEnabled=false");
+				+ "?username=userxxx"
+				+ "&password=passxxx"
+				+ "&connectionTimeout=123"
+				+ "&requestedChannelMax=456"
+				+ "&requestedFrameMax=789"
+				+ "&requestedHeartbeat=987"
+				+ "&sslProtocol=true"
+				+ "&automaticRecoveryEnabled=true"
+				+ "&networkRecoveryInterval=654"
+				+ "&topologyRecoveryEnabled=false");
 
         assertEquals("localhost", connectionFactory.getHost());
         assertEquals(1234, connectionFactory.getPort());

http://git-wip-us.apache.org/repos/asf/camel/blob/1191e19f/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
new file mode 100644
index 0000000..6a63948
--- /dev/null
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
@@ -0,0 +1,77 @@
+package org.apache.camel.component.rabbitmq;
+
+import com.rabbitmq.client.AlreadyClosedException;
+import org.apache.camel.*;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import java.net.ConnectException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Integration test to check that RabbitMQ Endpoint is able to reconnect to broker when broker
+ * is not avaibable.
+ * <ul>
+ *     <li>Stop the broker</li>
+ *     <li>Run the test: the producer complains it can not send messages, the consumer
is silent</li>
+ *     <li>Start the broker: the producer sends messages, and the consumer receives
messages</li>
+ *     <li>Stop the broker: the producer complains it can not send messages, the consumer
is silent</li>
+ *     <li>Start the broker: the producer sends messages, and the consumer receives
messages</li>
+ * </ul>
+ */
+public class RabbitMQReConnectionIntTest extends CamelTestSupport {
+    private static final String EXCHANGE = "ex3";
+
+    @Produce(uri = "direct:rabbitMQ")
+    protected ProducerTemplate directProducer;
+
+    @EndpointInject(uri = "rabbitmq:localhost:5672/" + EXCHANGE + "?username=cameltest&password=cameltest"
+
+            "&queue=q3&routingKey=rk3"+
+            "&automaticRecoveryEnabled=true" +
+            "&requestedHeartbeat=1000" +
+            "&connectionTimeout=5000")
+    private Endpoint rabbitMQEndpoint;
+
+    @EndpointInject(uri = "mock:result")
+    private MockEndpoint mockEndpoint;
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+
+            @Override
+            public void configure() throws Exception {
+                from("direct:rabbitMQ")
+                        .id("producingRoute")
+                        .onException(AlreadyClosedException.class, ConnectException.class)
+                            .maximumRedeliveries(10)
+                            .redeliveryDelay(500L)
+                            .end()
+                        .log("Sending message")
+                        .inOnly(rabbitMQEndpoint);
+                from(rabbitMQEndpoint)
+                        .id("consumingRoute")
+                        .log("Receiving message")
+                        .to(mockEndpoint);
+            }
+        };
+    }
+    @Test
+    public void testSendEndReceive() throws Exception {
+        int nbMessages=100;
+        int failedMessages=0;
+        for(int i=0;i<nbMessages;i++) {
+            try {
+                directProducer.sendBodyAndHeader("Message #"+i, RabbitMQConstants.ROUTING_KEY,
"rk3");
+            } catch (CamelExecutionException e) {
+                log.debug("Can not send message", e);
+                failedMessages++;
+            }
+            Thread.sleep(500L);
+        }
+        mockEndpoint.assertExchangeReceived(nbMessages-failedMessages);
+        assertMockEndpointsSatisfied(5, TimeUnit.SECONDS);
+    }
+}


Mime
View raw message