You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2016/09/08 07:38:49 UTC
[2/2] camel git commit: CAMEL-10296: fix checkstyle issues and make
varible basicReturnReceived volatile
CAMEL-10296: fix checkstyle issues and make varible basicReturnReceived volatile
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fbca40a6
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fbca40a6
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fbca40a6
Branch: refs/heads/master
Commit: fbca40a6d048870031509066b90105548936a0c9
Parents: 0ce6444
Author: Florian Gessner <fl...@tis.biz>
Authored: Thu Sep 8 09:02:32 2016 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Thu Sep 8 09:36:32 2016 +0200
----------------------------------------------------------------------
.../rabbitmq/RabbitMQMessagePublisher.java | 30 +++++++++-----------
1 file changed, 14 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/fbca40a6/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 248df21..bde0430 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
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ReturnListener;
+
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.NoTypeConversionAvailableException;
@@ -38,22 +39,20 @@ import org.slf4j.LoggerFactory;
* A method object for publishing to RabbitMQ
*/
public class RabbitMQMessagePublisher {
- private final ReturnListener GUARANTEED_DELIVERY_RETURN_LISTENER = new ReturnListener() {
- @Override
- public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
- LOG.warn("Delivery failed for exchange {} and routing key {}; replyCode = {}; replyText = {}", exchange, routingKey, replyCode, replyText);
- basicReturnReceived = true;
- }
- };
-
private static final Logger LOG = LoggerFactory.getLogger(RabbitMQMessagePublisher.class);
private final Exchange camelExchange;
private final Channel channel;
private final String routingKey;
private final RabbitMQEndpoint endpoint;
private final Message message;
-
- private boolean basicReturnReceived = false;
+ private volatile boolean basicReturnReceived;
+ private final ReturnListener guaranteedDeliveryReturnListener = new ReturnListener() {
+ @Override
+ public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
+ LOG.warn("Delivery failed for exchange {} and routing key {}; replyCode = {}; replyText = {}", exchange, routingKey, replyCode, replyText);
+ basicReturnReceived = true;
+ }
+ };
public RabbitMQMessagePublisher(final Exchange camelExchange, final Channel channel, final String routingKey, final RabbitMQEndpoint endpoint) {
this.camelExchange = camelExchange;
@@ -113,9 +112,8 @@ public class RabbitMQMessagePublisher {
channel.confirmSelect();
}
if (endpoint.isGuaranteedDeliveries()) {
- basicReturnReceived = false;
- channel.addReturnListener(GUARANTEED_DELIVERY_RETURN_LISTENER);
-
+ basicReturnReceived = false;
+ channel.addReturnListener(guaranteedDeliveryReturnListener);
}
try {
@@ -125,7 +123,7 @@ public class RabbitMQMessagePublisher {
}
} finally {
if (endpoint.isGuaranteedDeliveries()) {
- channel.removeReturnListener(GUARANTEED_DELIVERY_RETURN_LISTENER);
+ channel.removeReturnListener(guaranteedDeliveryReturnListener);
}
}
}
@@ -138,8 +136,8 @@ public class RabbitMQMessagePublisher {
try {
LOG.debug("Waiting for publisher acknowledgements for {}ms", endpoint.getPublisherAcknowledgementsTimeout());
channel.waitForConfirmsOrDie(endpoint.getPublisherAcknowledgementsTimeout());
- if(basicReturnReceived){
- throw new RuntimeCamelException("Failed to deliver message; basic.return received");
+ if (basicReturnReceived) {
+ throw new RuntimeCamelException("Failed to deliver message; basic.return received");
}
} catch (InterruptedException | TimeoutException e) {
LOG.warn("Acknowledgement error for {}", camelExchange);