camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject camel git commit: CAMEL-9505: RabbitMQConsumer don't use Camel ExceptionHandler BEFORE requeing message
Date Wed, 13 Jan 2016 13:12:35 GMT
Repository: camel
Updated Branches:
  refs/heads/master d83fbc581 -> 0832667f2


CAMEL-9505: RabbitMQConsumer don't use Camel ExceptionHandler BEFORE requeing message


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

Branch: refs/heads/master
Commit: 0832667f2c43aee8f22606ea0ef672e1ec98a1ab
Parents: d83fbc5
Author: Andrea Cosentino <ancosen@gmail.com>
Authored: Wed Jan 13 14:05:34 2016 +0100
Committer: Andrea Cosentino <ancosen@gmail.com>
Committed: Wed Jan 13 14:09:03 2016 +0100

----------------------------------------------------------------------
 .../component/rabbitmq/RabbitMQConsumer.java    |  6 +-
 .../RabbitMQRequeueHandledExceptionIntTest.java | 81 ++++++++++++++++++++
 .../rabbitmq/RabbitMQRequeueIntTest.java        |  4 +-
 ...abbitMQRequeueUnhandledExceptionIntTest.java | 81 ++++++++++++++++++++
 4 files changed, 167 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/0832667f/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 eaf2b6c..c3569e7 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
@@ -253,6 +253,9 @@ public class RabbitMQConsumer extends DefaultConsumer {
                     channel.basicAck(deliveryTag, false);
                 }
             } else {
+                if (exchange.getException() != null) {
+                    getExceptionHandler().handleException("Error processing exchange", exchange,
exchange.getException());
+                }
                 boolean isRequeueHeaderSet = msg.getHeader(RabbitMQConstants.REQUEUE, false,
boolean.class);
                 // processing failed, then reject and handle the exception
                 if (deliveryTag != 0 && !consumer.endpoint.isAutoAck()) {
@@ -263,9 +266,6 @@ public class RabbitMQConsumer extends DefaultConsumer {
                         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/0832667f/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueHandledExceptionIntTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueHandledExceptionIntTest.java
b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueHandledExceptionIntTest.java
new file mode 100644
index 0000000..a6e1d73
--- /dev/null
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueHandledExceptionIntTest.java
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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 not to be re-queued when an
handled exception occurs.
+ */
+public class RabbitMQRequeueHandledExceptionIntTest extends CamelTestSupport {
+    public static final String ROUTING_KEY = "rk4";
+
+    @Produce(uri = "direct:rabbitMQ")
+    protected ProducerTemplate directProducer;
+
+    @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
+            + "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)
+                        .onException(Exception.class)
+                        .handled(true)
+                        .end()
+                        .id("consumingRoute")
+                        .log("Receiving message")
+                        .inOnly(consumingMockEndpoint)
+                        .throwException(new Exception("Simulated handled exception"));
+            }
+        };
+    }
+
+    @Test
+    public void testTrueRequeueHeaderWithHandleExceptionNotCausesRequeue() throws Exception
{
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.setMinimumExpectedMessageCount(1);
+
+        directProducer.sendBodyAndHeader("Hello, World!", RabbitMQConstants.REQUEUE, true);
+
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/0832667f/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
index 97798fc..e0c8d74 100644
--- 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
@@ -34,8 +34,8 @@ public class RabbitMQRequeueIntTest extends CamelTestSupport {
     @Produce(uri = "direct:rabbitMQ")
     protected ProducerTemplate directProducer;
 
-    @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?username=cameltest&password=cameltest"
-            + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+    @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
+            + "autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
     private Endpoint rabbitMQEndpoint;
 
     @EndpointInject(uri = "mock:producing")

http://git-wip-us.apache.org/repos/asf/camel/blob/0832667f/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueUnhandledExceptionIntTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueUnhandledExceptionIntTest.java
b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueUnhandledExceptionIntTest.java
new file mode 100644
index 0000000..4b7c6fa
--- /dev/null
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueUnhandledExceptionIntTest.java
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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 when an unhandled
exception occurs.
+ */
+public class RabbitMQRequeueUnhandledExceptionIntTest extends CamelTestSupport {
+    public static final String ROUTING_KEY = "rk4";
+
+    @Produce(uri = "direct:rabbitMQ")
+    protected ProducerTemplate directProducer;
+
+    @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
+            + "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)
+                        .onException(Exception.class)
+                        .handled(false)
+                        .end()
+                        .id("consumingRoute")
+                        .log("Receiving message")
+                        .inOnly(consumingMockEndpoint)
+                        .throwException(new Exception("Simulated unhandled exception"));
+            }
+        };
+    }
+
+    @Test
+    public void testTrueRequeueHeaderWithUnandleExceptionCausesRequeue() throws Exception
{
+        producingMockEndpoint.expectedMessageCount(1);
+        consumingMockEndpoint.setMinimumExpectedMessageCount(2);
+
+        directProducer.sendBodyAndHeader("Hello, World!", RabbitMQConstants.REQUEUE, true);
+
+        producingMockEndpoint.assertIsSatisfied();
+        consumingMockEndpoint.assertIsSatisfied();
+    }
+}


Mime
View raw message