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 2016/08/16 14:28:34 UTC
[3/6] camel git commit: Fixed CS. Polished. This closes #1119
Fixed CS. Polished. This closes #1119
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f39b83ee
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f39b83ee
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f39b83ee
Branch: refs/heads/master
Commit: f39b83eebb6133086e01d98c1f3fb3af38f2dd09
Parents: 7ee0977
Author: Claus Ibsen <da...@apache.org>
Authored: Tue Aug 16 16:27:35 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Aug 16 16:27:35 2016 +0200
----------------------------------------------------------------------
.../component/rabbitmq/RabbitConsumer.java | 55 ++++++++++----------
.../component/rabbitmq/RabbitMQConsumer.java | 3 +-
2 files changed, 28 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f39b83ee/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 93e2499..6c20b57 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
@@ -26,7 +26,6 @@ import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
-
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
@@ -42,7 +41,7 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
/** Consumer tag for this consumer. */
private volatile String consumerTag;
private volatile boolean stopping;
-
+
private final Semaphore lock = new Semaphore(1);
/**
@@ -59,29 +58,29 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
log.warn("Unable to open channel for RabbitMQConsumer. Continuing and will try again", e);
}
}
+
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- try {
+ try {
if (!consumer.getEndpoint().isAutoAck()) {
- lock.acquire();
+ lock.acquire();
}
//Channel might be open because while we were waiting for the lock, stop() has been succesfully called.
- if (!channel.isOpen()) return;
-
+ if (!channel.isOpen()) {
+ return;
+ }
+
try {
doHandleDelivery(consumerTag, envelope, properties, body);
} finally {
if (!consumer.getEndpoint().isAutoAck()) {
- lock.release();
+ lock.release();
}
- }
-
- } catch (InterruptedException e) {
- log.error("Thread Interrupted!");
-
- }
-
-
+ }
+
+ } catch (InterruptedException e) {
+ log.warn("Thread Interrupted!");
+ }
}
public void doHandleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
@@ -186,25 +185,25 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
channel.basicCancel(tag);
}
try {
- lock.acquire();
+ lock.acquire();
if (isChannelOpen()) {
channel.close();
}
- } catch (TimeoutException e) {
+ } catch (TimeoutException e) {
log.error("Timeout occured");
throw e;
} catch (InterruptedException e1) {
- log.error("Thread Interrupted!");
+ log.error("Thread Interrupted!");
} finally {
lock.release();
-
- }
+
+ }
}
/**
* Stores the most recently passed-in consumerTag - semantically, there
* should be only one.
- *
+ *
* @see Consumer#handleConsumeOk
*/
public void handleConsumeOk(String consumerTag) {
@@ -213,7 +212,7 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
/**
* Retrieve the consumer tag.
- *
+ *
* @return the most recently notified consumer tag.
*/
public String getConsumerTag() {
@@ -222,31 +221,31 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
/**
* No-op implementation of {@link Consumer#handleCancelOk}.
- *
+ *
* @param consumerTag
* the defined consumer tag (client- or server-generated)
*/
public void handleCancelOk(String consumerTag) {
// no work to do
- log.debug("Recieved cancelOk signal on the rabbitMQ channel");
+ log.debug("Received cancelOk signal on the rabbitMQ channel");
}
/**
* No-op implementation of {@link Consumer#handleCancel(String)}
- *
+ *
* @param consumerTag
* the defined consumer tag (client- or server-generated)
*/
public void handleCancel(String consumerTag) throws IOException {
// no work to do
- log.debug("Recieved cancel signal on the rabbitMQ channel");
+ log.debug("Received cancel signal on the rabbitMQ channel");
}
/**
* No-op implementation of {@link Consumer#handleShutdownSignal}.
*/
public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
- log.info("Recieved shutdown signal on the rabbitMQ channel");
+ log.info("Received shutdown signal on the rabbitMQ channel");
// Check if the consumer closed the connection or something else
if (!sig.isInitiatedByApplication()) {
@@ -277,7 +276,7 @@ class RabbitConsumer implements com.rabbitmq.client.Consumer {
*/
public void handleRecoverOk(String consumerTag) {
// no work to do
- log.debug("Recieved recover ok signal on the rabbitMQ channel");
+ log.debug("Received recover ok signal on the rabbitMQ channel");
}
/**
http://git-wip-us.apache.org/repos/asf/camel/blob/f39b83ee/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 9faffc2..9c02cb7 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
@@ -131,8 +131,7 @@ public class RabbitMQConsumer extends DefaultConsumer implements Suspendable {
try {
consumer.stop();
} catch (TimeoutException e) {
- log.error("Timeout occured");
- throw e;
+ log.warn("Timeout occurred while stopping consumer. This exception is ignored", e);
}
}
this.consumers.clear();