You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/09/26 06:34:35 UTC
[1/4] camel git commit: CAMEL-11791: Enhanced reconnection for
rabbitmq consumer and producer (including queue/exchange deletion)
Repository: camel
Updated Branches:
refs/heads/camel-2.19.x 2c69684cb -> 7bcefc1a9
refs/heads/master 5b08f5034 -> 4e179e647
CAMEL-11791: Enhanced reconnection for rabbitmq consumer and producer (including queue/exchange deletion)
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/c5443cf5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/c5443cf5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/c5443cf5
Branch: refs/heads/master
Commit: c5443cf52f8662a7e19ff40bdde3c3ae8751566c
Parents: 5b08f50
Author: Veiga Ortiz, Héctor <he...@here.com>
Authored: Mon Sep 25 14:47:18 2017 -0400
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Sep 26 08:28:50 2017 +0200
----------------------------------------------------------------------
.../component/rabbitmq/RabbitConsumer.java | 26 ++++++++++++++------
.../component/rabbitmq/RabbitMQConsumer.java | 12 ++++++---
.../component/rabbitmq/RabbitMQEndpoint.java | 2 +-
.../rabbitmq/RabbitMQMessagePublisher.java | 1 +
.../component/rabbitmq/RabbitMQProducer.java | 8 +++++-
.../rabbitmq/pool/PoolableChannelFactory.java | 6 ++++-
6 files changed, 41 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
index 6c20b57..e96367c 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
@@ -237,8 +237,16 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
* the defined consumer tag (client- or server-generated)
*/
public void handleCancel(String consumerTag) throws IOException {
- // no work to do
- log.debug("Received cancel signal on the rabbitMQ channel");
+ log.debug("Received cancel signal on the rabbitMQ channel.");
+
+ try {
+ channel.basicCancel(tag);
+ } catch (Exception e) {
+ //no-op
+ }
+
+ this.consumer.getEndpoint().declareExchangeAndQueue(channel);
+ this.start();
}
/**
@@ -287,12 +295,16 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
if (isChannelOpen()) {
// The connection is good, so nothing to do
return;
+ } else if (!isChannelOpen() && this.consumer.getEndpoint().getAutomaticRecoveryEnabled()) {
+ // Still need to wait for channel to re-open
+ throw new IOException("Waiting for channel to re-open.");
+ } else if (!this.consumer.getEndpoint().getAutomaticRecoveryEnabled()) {
+ log.info("Attempting to open a new rabbitMQ channel");
+ Connection conn = consumer.getConnection();
+ channel = openChannel(conn);
+ // Register the channel to the tag
+ start();
}
- log.info("Attempting to open a new rabbitMQ channel");
- Connection conn = consumer.getConnection();
- channel = openChannel(conn);
- // Register the channel to the tag
- start();
}
private boolean isChannelOpen() {
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/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 9c02cb7..95a6609 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
@@ -71,12 +71,16 @@ public class RabbitMQConsumer extends DefaultConsumer implements Suspendable {
* @throws TimeoutException
*/
protected synchronized Connection getConnection() throws IOException, TimeoutException {
- if (this.conn != null && this.conn.isOpen()) {
+ if (this.conn == null) {
+ openConnection();
+ return this.conn;
+ } else if (!this.conn.isOpen() && this.endpoint.getAutomaticRecoveryEnabled()) {
+ return this.conn;
+ } else {
+ log.debug("The existing connection is closed");
+ openConnection();
return this.conn;
}
- log.debug("The existing connection is closed");
- openConnection();
- return this.conn;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/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 e3cb28d..c3cd8bf 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
@@ -117,7 +117,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
@UriParam(label = "advanced")
private Boolean automaticRecoveryEnabled;
@UriParam(label = "advanced")
- private Integer networkRecoveryInterval;
+ private Integer networkRecoveryInterval = 5000;
@UriParam(label = "advanced")
private Boolean topologyRecoveryEnabled;
@UriParam(label = "consumer")
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
index 85e657f..a61d470 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
@@ -24,6 +24,7 @@ import java.io.Serializable;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ReturnListener;
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/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 3e45c15..759039d 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
@@ -79,9 +79,11 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
channel = channelPool.borrowObject();
}
if (!channel.isOpen()) {
- log.warn("Got a closed channel from the pool");
+ log.warn("Got a closed channel from the pool. Invalidating and borrowing a new one from the pool.");
+ channelPool.invalidateObject(channel);
// Reconnect if another thread hasn't yet
checkConnectionAndChannelPool();
+ attemptDeclaration();
channel = channelPool.borrowObject();
}
try {
@@ -103,6 +105,10 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
log.trace("Creating channel pool...");
channelPool = new GenericObjectPool<Channel>(new PoolableChannelFactory(this.conn), getEndpoint().getChannelPoolMaxSize(),
GenericObjectPool.WHEN_EXHAUSTED_BLOCK, getEndpoint().getChannelPoolMaxWait());
+ attemptDeclaration();
+ }
+
+ private synchronized void attemptDeclaration() throws Exception {
if (getEndpoint().isDeclare()) {
execute(new ChannelCallback<Void>() {
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/c5443cf5/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
index b10201f..ea0e619 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
@@ -41,7 +41,11 @@ public class PoolableChannelFactory implements PoolableObjectFactory<Channel> {
@Override
public void destroyObject(Channel t) throws Exception {
- t.close();
+ try {
+ t.close();
+ } catch (Exception e) {
+ //no-op
+ }
}
@Override
[4/4] camel git commit: CAMEL-11791: Polished. This closes #1957
Posted by da...@apache.org.
CAMEL-11791: Polished. This closes #1957
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7bcefc1a
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7bcefc1a
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7bcefc1a
Branch: refs/heads/camel-2.19.x
Commit: 7bcefc1a9f5e6b082223d02b6cc1c0ba83094c0f
Parents: 99f0f24
Author: Claus Ibsen <da...@apache.org>
Authored: Tue Sep 26 08:31:33 2017 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Sep 26 08:34:03 2017 +0200
----------------------------------------------------------------------
.../camel-rabbitmq/src/main/docs/rabbitmq-component.adoc | 6 +++---
.../apache/camel/component/rabbitmq/RabbitMQEndpoint.java | 8 ++++----
2 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/7bcefc1a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
index cbd0e2b..428f294 100644
--- a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
+++ b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
@@ -108,14 +108,14 @@ with the following path and query parameters:
| **connectionFactory** (advanced) | To use a custom RabbitMQ connection factory. When this option is set all connection options (connectionTimeout requestedChannelMax...) set on URI are not used | | ConnectionFactory
| **exchangeArgs** (advanced) | Key/value args for configuring the exchange parameters when declare=true | | Map
| **exchangeArgsConfigurer** (advanced) | Set the configurer for setting the exchange args in Channel.exchangeDeclare | | ArgsConfigurer
-| **networkRecoveryInterval** (advanced) | Network recovery interval in milliseconds (interval used when recovering from network failure) | | Integer
+| **networkRecoveryInterval** (advanced) | Network recovery interval in milliseconds (interval used when recovering from network failure) | 5000 | Integer
| **queueArgs** (advanced) | Key/value args for configuring the queue parameters when declare=true | | Map
| **queueArgsConfigurer** (advanced) | Set the configurer for setting the queue args in Channel.queueDeclare | | ArgsConfigurer
| **requestedChannelMax** (advanced) | Connection requested channel max (max number of channels offered) | 0 | int
| **requestedFrameMax** (advanced) | Connection requested frame max (max size of frame offered) | 0 | int
| **requestedHeartbeat** (advanced) | Connection requested heartbeat (heart-beat in seconds offered) | 60 | int
-| **requestTimeout** (advanced) | Set timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds) | | long
-| **requestTimeoutChecker Interval** (advanced) | Set requestTimeoutCheckerInterval for inOut exchange | | long
+| **requestTimeout** (advanced) | Set timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds) | 20000 | long
+| **requestTimeoutChecker Interval** (advanced) | Set requestTimeoutCheckerInterval for inOut exchange | 1000 | long
| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
| **topologyRecoveryEnabled** (advanced) | Enables connection topology recovery (should topology recovery be performed) | | Boolean
| **transferException** (advanced) | When true and an inOut Exchange failed on the consumer side send the caused Exception back in the response | false | boolean
http://git-wip-us.apache.org/repos/asf/camel/blob/7bcefc1a/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 2f064cd..9efb4ee 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
@@ -75,7 +75,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private boolean autoDelete = true;
@UriParam(label = "common", defaultValue = "true")
private boolean durable = true;
- @UriParam(label = "common", defaultValue = "false")
+ @UriParam(label = "common")
private boolean exclusive;
@UriParam(label = "producer")
private boolean bridgeEndpoint;
@@ -111,7 +111,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private ConnectionFactory connectionFactory;
@UriParam(label = "advanced")
private Boolean automaticRecoveryEnabled;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "5000")
private Integer networkRecoveryInterval = 5000;
@UriParam(label = "advanced")
private Boolean topologyRecoveryEnabled;
@@ -160,9 +160,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
@UriParam(label = "advanced")
@Deprecated
private ArgsConfigurer exchangeArgsConfigurer;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "20000")
private long requestTimeout = 20000;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "1000")
private long requestTimeoutCheckerInterval = 1000;
@UriParam(label = "advanced")
private boolean transferException;
[2/4] camel git commit: CAMEL-11791: Polished. This closes #1957
Posted by da...@apache.org.
CAMEL-11791: Polished. This closes #1957
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4e179e64
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4e179e64
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4e179e64
Branch: refs/heads/master
Commit: 4e179e64761e7861fd2571505e47471d098d1de9
Parents: c5443cf
Author: Claus Ibsen <da...@apache.org>
Authored: Tue Sep 26 08:31:33 2017 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Sep 26 08:31:33 2017 +0200
----------------------------------------------------------------------
.../camel-rabbitmq/src/main/docs/rabbitmq-component.adoc | 6 +++---
.../apache/camel/component/rabbitmq/RabbitMQEndpoint.java | 10 +++++-----
2 files changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/4e179e64/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
index 0d7a1eb..63070f6 100644
--- a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
+++ b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
@@ -111,14 +111,14 @@ with the following path and query parameters:
| *connectionFactory* (advanced) | To use a custom RabbitMQ connection factory. When this option is set all connection options (connectionTimeout requestedChannelMax...) set on URI are not used | | ConnectionFactory
| *exchangeArgs* (advanced) | *Deprecated* Key/value args for configuring the exchange parameters when declare=true | | Map
| *exchangeArgsConfigurer* (advanced) | *Deprecated* Set the configurer for setting the exchange args in Channel.exchangeDeclare | | ArgsConfigurer
-| *networkRecoveryInterval* (advanced) | Network recovery interval in milliseconds (interval used when recovering from network failure) | | Integer
+| *networkRecoveryInterval* (advanced) | Network recovery interval in milliseconds (interval used when recovering from network failure) | 5000 | Integer
| *queueArgs* (advanced) | *Deprecated* Key/value args for configuring the queue parameters when declare=true | | Map
| *queueArgsConfigurer* (advanced) | *Deprecated* Set the configurer for setting the queue args in Channel.queueDeclare | | ArgsConfigurer
| *requestedChannelMax* (advanced) | Connection requested channel max (max number of channels offered) | 0 | int
| *requestedFrameMax* (advanced) | Connection requested frame max (max size of frame offered) | 0 | int
| *requestedHeartbeat* (advanced) | Connection requested heartbeat (heart-beat in seconds offered) | 60 | int
-| *requestTimeout* (advanced) | Set timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds) | | long
-| *requestTimeoutChecker Interval* (advanced) | Set requestTimeoutCheckerInterval for inOut exchange | | long
+| *requestTimeout* (advanced) | Set timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds) | 20000 | long
+| *requestTimeoutChecker Interval* (advanced) | Set requestTimeoutCheckerInterval for inOut exchange | 1000 | long
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
| *topologyRecoveryEnabled* (advanced) | Enables connection topology recovery (should topology recovery be performed) | | Boolean
| *transferException* (advanced) | When true and an inOut Exchange failed on the consumer side send the caused Exception back in the response | false | boolean
http://git-wip-us.apache.org/repos/asf/camel/blob/4e179e64/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 c3cd8bf..11abfe7 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
@@ -78,9 +78,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private boolean autoDelete = true;
@UriParam(label = "common", defaultValue = "true")
private boolean durable = true;
- @UriParam(label = "common", defaultValue = "false")
+ @UriParam(label = "common")
private boolean exclusive;
- @UriParam(label = "common", defaultValue = "false")
+ @UriParam(label = "common")
private boolean passive;
@UriParam(label = "producer")
private boolean bridgeEndpoint;
@@ -116,7 +116,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private ConnectionFactory connectionFactory;
@UriParam(label = "advanced")
private Boolean automaticRecoveryEnabled;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "5000")
private Integer networkRecoveryInterval = 5000;
@UriParam(label = "advanced")
private Boolean topologyRecoveryEnabled;
@@ -165,9 +165,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
@UriParam(label = "advanced")
@Deprecated
private ArgsConfigurer exchangeArgsConfigurer;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "20000")
private long requestTimeout = 20000;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced", defaultValue = "1000")
private long requestTimeoutCheckerInterval = 1000;
@UriParam(label = "advanced")
private boolean transferException;
[3/4] camel git commit: CAMEL-11791: Enhanced reconnection for
rabbitmq consumer and producer (including queue/exchange deletion)
Posted by da...@apache.org.
CAMEL-11791: Enhanced reconnection for rabbitmq consumer and producer (including queue/exchange deletion)
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/99f0f246
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/99f0f246
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/99f0f246
Branch: refs/heads/camel-2.19.x
Commit: 99f0f2460f2ef159e086c306acee153e7df7e217
Parents: 2c69684
Author: Veiga Ortiz, Héctor <he...@here.com>
Authored: Mon Sep 25 14:47:18 2017 -0400
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Sep 26 08:32:06 2017 +0200
----------------------------------------------------------------------
.../component/rabbitmq/RabbitConsumer.java | 26 ++++++++++++++------
.../component/rabbitmq/RabbitMQConsumer.java | 12 ++++++---
.../component/rabbitmq/RabbitMQEndpoint.java | 2 +-
.../rabbitmq/RabbitMQMessagePublisher.java | 1 +
.../component/rabbitmq/RabbitMQProducer.java | 8 +++++-
.../rabbitmq/pool/PoolableChannelFactory.java | 6 ++++-
6 files changed, 41 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
index 6c20b57..e96367c 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
@@ -237,8 +237,16 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
* the defined consumer tag (client- or server-generated)
*/
public void handleCancel(String consumerTag) throws IOException {
- // no work to do
- log.debug("Received cancel signal on the rabbitMQ channel");
+ log.debug("Received cancel signal on the rabbitMQ channel.");
+
+ try {
+ channel.basicCancel(tag);
+ } catch (Exception e) {
+ //no-op
+ }
+
+ this.consumer.getEndpoint().declareExchangeAndQueue(channel);
+ this.start();
}
/**
@@ -287,12 +295,16 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
if (isChannelOpen()) {
// The connection is good, so nothing to do
return;
+ } else if (!isChannelOpen() && this.consumer.getEndpoint().getAutomaticRecoveryEnabled()) {
+ // Still need to wait for channel to re-open
+ throw new IOException("Waiting for channel to re-open.");
+ } else if (!this.consumer.getEndpoint().getAutomaticRecoveryEnabled()) {
+ log.info("Attempting to open a new rabbitMQ channel");
+ Connection conn = consumer.getConnection();
+ channel = openChannel(conn);
+ // Register the channel to the tag
+ start();
}
- log.info("Attempting to open a new rabbitMQ channel");
- Connection conn = consumer.getConnection();
- channel = openChannel(conn);
- // Register the channel to the tag
- start();
}
private boolean isChannelOpen() {
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/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 9c02cb7..95a6609 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
@@ -71,12 +71,16 @@ public class RabbitMQConsumer extends DefaultConsumer implements Suspendable {
* @throws TimeoutException
*/
protected synchronized Connection getConnection() throws IOException, TimeoutException {
- if (this.conn != null && this.conn.isOpen()) {
+ if (this.conn == null) {
+ openConnection();
+ return this.conn;
+ } else if (!this.conn.isOpen() && this.endpoint.getAutomaticRecoveryEnabled()) {
+ return this.conn;
+ } else {
+ log.debug("The existing connection is closed");
+ openConnection();
return this.conn;
}
- log.debug("The existing connection is closed");
- openConnection();
- return this.conn;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/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 ddfdfe8..2f064cd 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
@@ -112,7 +112,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
@UriParam(label = "advanced")
private Boolean automaticRecoveryEnabled;
@UriParam(label = "advanced")
- private Integer networkRecoveryInterval;
+ private Integer networkRecoveryInterval = 5000;
@UriParam(label = "advanced")
private Boolean topologyRecoveryEnabled;
@UriParam(label = "consumer")
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
index 85e657f..a61d470 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessagePublisher.java
@@ -24,6 +24,7 @@ import java.io.Serializable;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ReturnListener;
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/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 3e45c15..759039d 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
@@ -79,9 +79,11 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
channel = channelPool.borrowObject();
}
if (!channel.isOpen()) {
- log.warn("Got a closed channel from the pool");
+ log.warn("Got a closed channel from the pool. Invalidating and borrowing a new one from the pool.");
+ channelPool.invalidateObject(channel);
// Reconnect if another thread hasn't yet
checkConnectionAndChannelPool();
+ attemptDeclaration();
channel = channelPool.borrowObject();
}
try {
@@ -103,6 +105,10 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
log.trace("Creating channel pool...");
channelPool = new GenericObjectPool<Channel>(new PoolableChannelFactory(this.conn), getEndpoint().getChannelPoolMaxSize(),
GenericObjectPool.WHEN_EXHAUSTED_BLOCK, getEndpoint().getChannelPoolMaxWait());
+ attemptDeclaration();
+ }
+
+ private synchronized void attemptDeclaration() throws Exception {
if (getEndpoint().isDeclare()) {
execute(new ChannelCallback<Void>() {
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/99f0f246/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
index b10201f..ea0e619 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/pool/PoolableChannelFactory.java
@@ -41,7 +41,11 @@ public class PoolableChannelFactory implements PoolableObjectFactory<Channel> {
@Override
public void destroyObject(Channel t) throws Exception {
- t.close();
+ try {
+ t.close();
+ } catch (Exception e) {
+ //no-op
+ }
}
@Override