camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/3] camel git commit: Adding ability to REQUEUE a message on RabbitMQ server instead of REJECTING message / sending it to DLQ.
Date Sun, 22 Feb 2015 15:19:10 GMT
Adding ability to REQUEUE a message on RabbitMQ server instead of REJECTING message / sending
it to DLQ.


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

Branch: refs/heads/master
Commit: 464c06eac4ab7155e016ab72541525d0d2a2e77e
Parents: d5b5964
Author: Andrew Austin <andrew.austin@wgu.edu>
Authored: Sat Feb 21 17:16:24 2015 -0700
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sun Feb 22 16:17:32 2015 +0100

----------------------------------------------------------------------
 .../component/rabbitmq/RabbitMQConstants.java   |   1 +
 .../component/rabbitmq/RabbitMQConsumer.java    |  21 ++--
 .../rabbitmq/RabbitMQRequeueIntTest.java        | 111 +++++++++++++++++++
 3 files changed, 126 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/464c06ea/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConstants.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConstants.java
index 383ac5d..b1f4a0c 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConstants.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConstants.java
@@ -34,6 +34,7 @@ public final class RabbitMQConstants {
     public static final String EXPIRATION = "rabbitmq.EXPIRATION";
     public static final String TIMESTAMP = "rabbitmq.TIMESTAMP";
     public static final String APP_ID = "rabbitmq.APP_ID";
+    public static final String REQUEUE = "rabbitmq.REQUEUE";
     public static final String RABBITMQ_DEAD_LETTER_EXCHANGE = "x-dead-letter-exchange";
     public static final String RABBITMQ_DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
     

http://git-wip-us.apache.org/repos/asf/camel/blob/464c06ea/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 3201887..73aa814 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
@@ -184,6 +184,13 @@ public class RabbitMQConsumer extends DefaultConsumer {
             Exchange exchange = consumer.endpoint.createRabbitExchange(envelope, properties,
body);
             mergeAmqpProperties(exchange, properties);
 
+            Message msg;
+            if (exchange.hasOut()) {
+                msg = exchange.getOut();
+            } else {
+                msg = exchange.getIn();
+            }
+
             boolean sendReply = properties.getReplyTo() != null;
             if (sendReply && !exchange.getPattern().isOutCapable()) {
                 exchange.setPattern(ExchangePattern.InOut);
@@ -200,12 +207,6 @@ public class RabbitMQConsumer extends DefaultConsumer {
             if (!exchange.isFailed()) {
                 // processing success
                 if (sendReply && exchange.getPattern().isOutCapable()) {
-                    Message msg;
-                    if (exchange.hasOut()) {
-                        msg = exchange.getOut();
-                    } else {
-                        msg = exchange.getIn();
-                    }
                     AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                             .headers(msg.getHeaders())
                             .correlationId(properties.getCorrelationId())
@@ -217,9 +218,15 @@ public class RabbitMQConsumer extends DefaultConsumer {
                     channel.basicAck(deliveryTag, false);
                 }
             } else {
+                boolean isRequeueHeaderSet = msg.getHeader(RabbitMQConstants.REQUEUE, false,
boolean.class);
                 // processing failed, then reject and handle the exception
                 if (deliveryTag != 0 && !consumer.endpoint.isAutoAck()) {
-                    channel.basicReject(deliveryTag, false);
+                    log.trace("Rejecting receipt [delivery_tag={}] with requeue={}", deliveryTag,
isRequeueHeaderSet);
+                    if (isRequeueHeaderSet) {
+                        channel.basicReject(deliveryTag, true);
+                    } else {
+                        channel.basicReject(deliveryTag, false);
+                    }
                 }
                 if (exchange.getException() != null) {
                     getExceptionHandler().handleException("Error processing exchange", exchange,
exchange.getException());

http://git-wip-us.apache.org/repos/asf/camel/blob/464c06ea/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
new file mode 100644
index 0000000..534f5b4
--- /dev/null
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
@@ -0,0 +1,111 @@
+package org.apache.camel.component.rabbitmq;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Produce;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+/**
+ * Integration test to confirm REQUEUE header causes message to be re-queued instead of sent
to DLQ.
+ *
+ * Created by Andrew Austin on 2/21/15.
+ */
+public class RabbitMQRequeueIntTest extends CamelTestSupport {
+    public static final String ROUTING_KEY = "rk4";
+
+    @Produce(uri = "direct:rabbitMQ")
+    protected ProducerTemplate directProducer;
+
+    @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?username=cameltest&password=cameltest"
+            + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+    private Endpoint rabbitMQEndpoint;
+
+    @EndpointInject(uri = "mock:producing")
+    private MockEndpoint producingMockEndpoint;
+
+    @EndpointInject(uri = "mock:consuming")
+    private MockEndpoint consumingMockEndpoint;
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+
+            @Override
+            public void configure() throws Exception {
+                from("direct:rabbitMQ")
+                        .id("producingRoute")
+                        .log("Sending message")
+                        .inOnly(rabbitMQEndpoint)
+                        .to(producingMockEndpoint);
+                from(rabbitMQEndpoint)
+                        .id("consumingRoute")
+                        .log("Receiving message")
+                        .inOnly(consumingMockEndpoint)
+                        .choice()
+                        .when(body().isEqualTo("requeue header false"))
+                            .log("Setting REQUEUE flag to false")
+                            .setHeader(RabbitMQConstants.REQUEUE, constant(false))
+                        .when(body().isEqualTo("requeue header true"))
+                            .log("Setting REQUEUE flag to true")
+                            .setHeader(RabbitMQConstants.REQUEUE, constant(true))
+                        .when(body().isEqualTo("non-boolean header"))
+                                .log("Setting REQUEUE flag to non-boolean")
+                                .setHeader(RabbitMQConstants.REQUEUE, constant(4l))
+                        .end()
+                        .throwException(new Exception("Simulated exception"));
+            }
+        };
+    }
+
+    @Test
+    public void testNoRequeueHeaderCausesReject() throws Exception {
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.expectedMessageCount(1);
+
+        directProducer.sendBody("no requeue header");
+
+        Thread.sleep(100);
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+
+    @Test
+    public void testNonBooleanRequeueHeaderCausesReject() throws Exception {
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.expectedMessageCount(1);
+
+        directProducer.sendBody("non-boolean header");
+
+        Thread.sleep(100);
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+
+    @Test
+    public void testFalseRequeueHeaderCausesReject() throws Exception {
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.expectedMessageCount(1);
+
+        directProducer.sendBody("non-boolean header");
+
+        Thread.sleep(100);
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+
+    @Test
+    public void testTrueRequeueHeaderCausesRequeue() throws Exception {
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.setMinimumExpectedMessageCount(2);
+
+        directProducer.sendBody("requeue header true");
+
+        Thread.sleep(100);
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+}


Mime
View raw message