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 2018/03/07 09:40:25 UTC
[camel] branch master updated (c0894b1 -> fde3332)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from c0894b1 CAMEL-12324: Improve docs
new 972e124 CAMEL-12329: camel-rabbitmq. Use a new header to override sending to another destination name in the producer to avoid issue with from->to sending to itself in from. This is now aligned with how eg the JMS component does it with a special OVERRIDE header for overrding the destination.
new 93dd1b6 CAMEL-12329: camel-rabbitmq. Use a new header to override sending to another destination name in the producer to avoid issue with from->to sending to itself in from. This is now aligned with how eg the JMS component does it with a special OVERRIDE header for overrding the destination.
new fde3332 CAMEL-12329: Fixed rabbitmq integration tests
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
components/camel-rabbitmq/readme.txt | 21 +++++
.../src/main/docs/rabbitmq-component.adoc | 24 +++++-
.../component/rabbitmq/RabbitMQConstants.java | 3 +
.../camel/component/rabbitmq/RabbitMQEndpoint.java | 13 ---
.../rabbitmq/RabbitMQMessagePublisher.java | 22 +++--
.../camel/component/rabbitmq/RabbitMQProducer.java | 18 ++++-
.../rabbitmq/reply/CorrelationTimeoutMap.java | 4 +-
.../rabbitmq/reply/TemporaryQueueReplyHandler.java | 4 +-
.../rabbitmq/reply/TemporaryQueueReplyManager.java | 6 +-
.../component/rabbitmq/RabbitMQBasicIntTest.java | 93 ++++++++++++++++++++++
.../rabbitmq/RabbitMQConsumerIntTestReplyTo.java | 2 +-
.../RabbitMQRequeueHandledExceptionIntTest.java | 4 +-
.../component/rabbitmq/RabbitMQRequeueIntTest.java | 4 +-
.../RabbitMQRequeueUnhandledExceptionIntTest.java | 4 +-
.../camel-rabbitmq/src/test/resources/README.md | 5 --
15 files changed, 180 insertions(+), 47 deletions(-)
create mode 100644 components/camel-rabbitmq/readme.txt
create mode 100644 components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQBasicIntTest.java
delete mode 100644 components/camel-rabbitmq/src/test/resources/README.md
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 02/03: CAMEL-12329: camel-rabbitmq. Use a new header to
override sending to another destination name in the producer to avoid issue
with from->to sending to itself in from. This is now aligned with how eg
the JMS component does it with a special OVERRIDE header for overrding the
destination.
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 93dd1b6c4dd1351c3baf4f317fa72d7adae4c2c2
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Mar 7 10:17:12 2018 +0100
CAMEL-12329: camel-rabbitmq. Use a new header to override sending to another destination name in the producer to avoid issue with from->to sending to itself in from. This is now aligned with how eg the JMS component does it with a special OVERRIDE header for overrding the destination.
---
components/camel-rabbitmq/readme.txt | 13 +++
.../camel/component/rabbitmq/RabbitMQEndpoint.java | 13 ---
.../rabbitmq/RabbitMQMessagePublisher.java | 13 ++-
.../camel/component/rabbitmq/RabbitMQProducer.java | 2 +-
.../rabbitmq/reply/CorrelationTimeoutMap.java | 4 +-
.../rabbitmq/reply/TemporaryQueueReplyHandler.java | 4 +-
.../rabbitmq/reply/TemporaryQueueReplyManager.java | 6 +-
.../component/rabbitmq/RabbitMQIntBasicTest.java | 94 ++++++++++++++++++++++
8 files changed, 125 insertions(+), 24 deletions(-)
diff --git a/components/camel-rabbitmq/readme.txt b/components/camel-rabbitmq/readme.txt
new file mode 100644
index 0000000..b95b015
--- /dev/null
+++ b/components/camel-rabbitmq/readme.txt
@@ -0,0 +1,13 @@
+Integration testing
+===================
+
+The camel-rabbitmq component has both unit tests and integration tests.
+The integration tests requires a running RabbitMQ broker. This broker can be run via Docker
+
+ docker run -d -p 5672:5672 -e RABBITMQ_DEFAULT_USER=cameltest -e RABBITMQ_DEFAULT_PASS=cameltest --hostname my-rabbit --name some-rabbit rabbitmq:3
+
+And then the integration tests can be run via Maven
+
+ mvn test -P itest
+
+
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 11abfe7..d3567b5 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
@@ -223,19 +223,6 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
new RabbitMQMessagePublisher(camelExchange, channel, routingKey, this).publish();
}
- /**
- * Extracts name of the rabbitmq exchange
- */
- protected String getExchangeName(Message msg) {
- String exchangeName = msg.getHeader(RabbitMQConstants.EXCHANGE_NAME, String.class);
- // If it is BridgeEndpoint we should ignore the message header of
- // EXCHANGE_NAME
- if (exchangeName == null || isBridgeEndpoint()) {
- exchangeName = getExchangeName();
- }
- return exchangeName;
- }
-
@Override
public Consumer createConsumer(Processor processor) throws Exception {
RabbitMQConsumer consumer = new RabbitMQConsumer(this, processor);
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 7e59eec..f5f3bb9 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
@@ -105,12 +105,19 @@ public class RabbitMQMessagePublisher {
}
private void publishToRabbitMQ(final AMQP.BasicProperties properties, final byte[] body) throws IOException {
- String rabbitExchange = endpoint.getExchangeName(message);
+ // remove the OVERRIDE header so it does not propagate
+ String exchangeName = (String) message.removeHeader(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME);
+ // If it is BridgeEndpoint we should ignore the message header of EXCHANGE_OVERRIDE_NAME
+ if (exchangeName == null || endpoint.isBridgeEndpoint()) {
+ exchangeName = endpoint.getExchangeName();
+ } else {
+ LOG.debug("Overriding header: {} detected sending message to exchange: {}", RabbitMQConstants.EXCHANGE_OVERRIDE_NAME, exchangeName);
+ }
Boolean mandatory = camelExchange.getIn().getHeader(RabbitMQConstants.MANDATORY, endpoint.isMandatory(), Boolean.class);
Boolean immediate = camelExchange.getIn().getHeader(RabbitMQConstants.IMMEDIATE, endpoint.isImmediate(), Boolean.class);
- LOG.debug("Sending message to exchange: {} with CorrelationId: {}", rabbitExchange, properties.getCorrelationId());
+ LOG.debug("Sending message to exchange: {} with CorrelationId: {}", exchangeName, properties.getCorrelationId());
if (isPublisherAcknowledgements()) {
channel.confirmSelect();
@@ -121,7 +128,7 @@ public class RabbitMQMessagePublisher {
}
try {
- channel.basicPublish(rabbitExchange, routingKey, mandatory, immediate, properties, body);
+ channel.basicPublish(exchangeName, routingKey, mandatory, immediate, properties, body);
if (isPublisherAcknowledgements()) {
waitForConfirmation();
}
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 a1cc075..996d352 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
@@ -372,7 +372,7 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
String name = "RabbitMQReplyManagerTimeoutChecker[" + getEndpoint().getExchangeName() + "]";
ScheduledExecutorService replyManagerExecutorService = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(name, name);
replyManager.setScheduledExecutorService(replyManagerExecutorService);
- log.info("Starting reply manager service " + name);
+ log.debug("Staring ReplyManager: {}", name);
ServiceHelper.startService(replyManager);
return replyManager;
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/CorrelationTimeoutMap.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/CorrelationTimeoutMap.java
index 8f052a0..7ed1089 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/CorrelationTimeoutMap.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/CorrelationTimeoutMap.java
@@ -86,13 +86,13 @@ public class CorrelationTimeoutMap extends DefaultTimeoutMap<String, ReplyHandle
} else {
result = super.put(key, value, timeoutMillis);
}
- log.info("Added correlationID: {} to timeout after: {} millis", key, timeoutMillis);
+ log.debug("Added correlationID: {} to timeout after: {} millis", key, timeoutMillis);
return result;
}
@Override
public ReplyHandler putIfAbsent(String key, ReplyHandler value, long timeoutMillis) {
- log.info("in putIfAbsent with key {}", key);
+ log.trace("putIfAbsent with key {}", key);
try {
if (listener != null) {
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyHandler.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyHandler.java
index 542e8c0..10234a4 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyHandler.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyHandler.java
@@ -53,7 +53,7 @@ public class TemporaryQueueReplyHandler implements ReplyHandler {
public void onReply(String correlationId, AMQP.BasicProperties properties, byte[] reply) {
// create holder object with the the reply
- log.info("in onReply with correlationId= {}", correlationId);
+ log.debug("onReply with correlationId: {}", correlationId);
ReplyHolder holder = new ReplyHolder(exchange, callback, originalCorrelationId, correlationId, properties, reply);
// process the reply
replyManager.processReply(holder);
@@ -61,7 +61,7 @@ public class TemporaryQueueReplyHandler implements ReplyHandler {
public void onTimeout(String correlationId) {
// create holder object without the reply which means a timeout occurred
- log.info("in onTimeout with correlationId= {}", correlationId);
+ log.debug("onTimeout with correlationId: {}", correlationId);
ReplyHolder holder = new ReplyHolder(exchange, callback, originalCorrelationId, correlationId, timeout);
// process timeout
replyManager.processReply(holder);
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyManager.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyManager.java
index 6f03017..842d43e 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyManager.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/reply/TemporaryQueueReplyManager.java
@@ -76,10 +76,10 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport {
@Override
protected Connection createListenerContainer() throws Exception {
- log.debug("Creating connection");
+ log.trace("Creating connection");
Connection conn = endpoint.connect(executorService);
- log.debug("Creating channel");
+ log.trace("Creating channel");
Channel channel = conn.createChannel();
// setup the basicQos
if (endpoint.isPrefetchEnabled()) {
@@ -89,7 +89,7 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport {
//Let the server pick a random name for us
DeclareOk result = channel.queueDeclare();
- log.info("Using temporary queue name: {}", result.getQueue());
+ log.debug("Using temporary queue name: {}", result.getQueue());
setReplyTo(result.getQueue());
//TODO check for the RabbitMQConstants.EXCHANGE_NAME header
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java
new file mode 100644
index 0000000..209f940
--- /dev/null
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java
@@ -0,0 +1,94 @@
+/**
+ * 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.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
+
+public class RabbitMQIntBasicTest extends AbstractRabbitMQIntTest {
+
+ // Startup RabbitMQ via Docker
+ // docker run -d -it -p 5672:5672 -e RABBITMQ_DEFAULT_USER=cameltest -e RABBITMQ_DEFAULT_PASS=cameltest --hostname my-rabbit --name some-rabbit rabbitmq:3
+
+ @EndpointInject(uri = "rabbitmq:localhost:5672/foo?username=cameltest&password=cameltest")
+ private Endpoint foo;
+
+ @EndpointInject(uri = "rabbitmq:localhost:5672/bar?username=cameltest&password=cameltest")
+ private Endpoint bar;
+
+ @EndpointInject(uri = "mock:result")
+ private MockEndpoint mock;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from(foo)
+ .log("FOO received: ${body}")
+ .to(bar);
+
+ from(bar)
+ .log("BAR received: ${body}")
+ .to(mock)
+ .transform().simple("Bye ${body}");
+ }
+ };
+ }
+
+ @Test
+ public void sentBasicInOnly() throws Exception {
+ mock.expectedBodiesReceived("World");
+
+ log.info("Sending to FOO");
+ template.sendBody(foo, "World");
+ log.info("Sending to FOO done");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ public void sentBasicInOut() throws Exception {
+ mock.expectedBodiesReceived("World");
+
+ log.info("Sending to FOO");
+ String out = template.requestBody(foo, "World", String.class);
+ assertEquals("Bye World", out);
+ log.info("Sending to FOO done");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ public void sentBasicInOutTwo() throws Exception {
+ mock.expectedBodiesReceived("World", "Camel");
+
+ log.info("Sending to FOO");
+ String out = template.requestBody(foo, "World", String.class);
+ assertEquals("Bye World", out);
+ out = template.requestBody(foo, "Camel", String.class);
+ assertEquals("Bye Camel", out);
+ log.info("Sending to FOO done");
+
+ mock.assertIsSatisfied();
+ }
+
+}
+
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 01/03: CAMEL-12329: camel-rabbitmq. Use a new header to
override sending to another destination name in the producer to avoid issue
with from->to sending to itself in from. This is now aligned with how eg
the JMS component does it with a special OVERRIDE header for overrding the
destination.
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 972e12433931eeb4cbaf3621519dfaa9010bc0ed
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Mar 7 09:40:10 2018 +0100
CAMEL-12329: camel-rabbitmq. Use a new header to override sending to another destination name in the producer to avoid issue with from->to sending to itself in from. This is now aligned with how eg the JMS component does it with a special OVERRIDE header for overrding the destination.
---
.../src/main/docs/rabbitmq-component.adoc | 24 ++++++++++++++++++----
.../component/rabbitmq/RabbitMQConstants.java | 3 +++
.../rabbitmq/RabbitMQMessagePublisher.java | 11 +++++-----
.../camel/component/rabbitmq/RabbitMQProducer.java | 16 ++++++++++++---
4 files changed, 42 insertions(+), 12 deletions(-)
diff --git a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
index d781a04..2727679 100644
--- a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
+++ b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
@@ -173,6 +173,8 @@ that will be used when producing a message
|`rabbitmq.DELIVERY_TAG` |The rabbitmq delivery tag of the received message
+|`rabbitmq.REDELIVERY_TAG` |Whether the message is a redelivered
+
|`rabbitmq.REQUEUE` |*Camel 2.14.2:* This is used by the consumer to control rejection of the
message. When the consumer is complete processing the exchange, and if
the exchange failed, then the consumer is going to reject the message
@@ -191,7 +193,9 @@ camel exchange then they will be set on the RabbitMQ message.
|`rabbitmq.ROUTING_KEY` |The routing key that will be used when sending the message
-|`rabbitmq.EXCHANGE_NAME` |The exchange the message was received from, or sent to
+|`rabbitmq.EXCHANGE_NAME` |The exchange the message was received from
+
+|`rabbitmq.EXCHANGE_OVERRIDE_NAME` | *Camel 2.21:* Used for force sending the message to this exchange instead of the endpoint configured name on the producer
|`rabbitmq.CONTENT_TYPE` |The contentType to set on the RabbitMQ message
@@ -276,7 +280,7 @@ public Map<String, Object> bindArgsBuilder() {
}
----
-==== Routing between exchanges
+==== Issue when routing between exchanges (in Camel 2.20.x or older)
If you for example want to route messages from one Rabbit exchange to another as shown
in the example below with foo -> bar:
@@ -295,7 +299,7 @@ and instead send the message to `foo`.
To avoid this you need to either:
-- remove the header
+- Remove the header:
[source,java]
----
@@ -304,7 +308,7 @@ from("rabbitmq://localhost/foo")
.to("rabbitmq://localhost/bar")
----
-- or turn on `bridgeEndpoint` mode on the producer.
+- Or turn on `bridgeEndpoint` mode on the producer:
[source,java]
----
@@ -312,3 +316,15 @@ from("rabbitmq://localhost/foo")
.to("rabbitmq://localhost/bar?bridgeEndpoint=true")
----
+From Camel 2.21 onwards this has been improved so you can easily route between exchanges.
+The header `rabbitmq.EXCHANGE_NAME` is not longer used by the producer to override the destination exchange.
+Instead a new header `rabbitmq.EXCHANGE_OVERRIDE_NAME` can be used to send to a different exchange.
+For example to send to cheese exchange you can do
+
+[source,java]
+----
+from("rabbitmq://localhost/foo")
+ .setHeader("rabbitmq.EXCHANGE_OVERRIDE_NAME", constant("cheese"))
+ .to("rabbitmq://localhost/bar")
+----
+
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 8f2ef49..d796a58 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
@@ -17,8 +17,10 @@
package org.apache.camel.component.rabbitmq;
public final class RabbitMQConstants {
+
// TODO need to change the constant which is start with camel
public static final String ROUTING_KEY = "rabbitmq.ROUTING_KEY";
+ public static final String EXCHANGE_OVERRIDE_NAME = "rabbitmq.EXCHANGE_OVERRIDE_NAME";
public static final String EXCHANGE_NAME = "rabbitmq.EXCHANGE_NAME";
public static final String CONTENT_TYPE = "rabbitmq.CONTENT_TYPE";
public static final String PRIORITY = "rabbitmq.PRIORITY";
@@ -47,6 +49,7 @@ public final class RabbitMQConstants {
public static final String RABBITMQ_QUEUE_MESSAGE_TTL_KEY = "x-message-ttl";
public static final String RABBITMQ_QUEUE_TTL_KEY = "x-expires";
+
private RabbitMQConstants() {
//Constants class
}
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..7e59eec 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
@@ -49,7 +49,7 @@ public class RabbitMQMessagePublisher {
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);
+ LOG.warn("Delivery failed for exchange: {} and routing key: {}; replyCode: {}; replyText: {}", exchange, routingKey, replyCode, replyText);
basicReturnReceived = true;
}
};
@@ -67,11 +67,12 @@ public class RabbitMQMessagePublisher {
// Remove the SERIALIZE_HEADER in case it was previously set
if (message.getHeaders() != null && message.getHeaders().containsKey(RabbitMQEndpoint.SERIALIZE_HEADER)) {
- LOG.debug("Removing the {} header", RabbitMQEndpoint.SERIALIZE_HEADER);
+ LOG.trace("Removing header: {}", RabbitMQEndpoint.SERIALIZE_HEADER);
message.getHeaders().remove(RabbitMQEndpoint.SERIALIZE_HEADER);
}
if (routingKey != null && routingKey.startsWith(RabbitMQConstants.RABBITMQ_DIRECT_REPLY_ROUTING_KEY)) {
message.setHeader(RabbitMQConstants.EXCHANGE_NAME, RabbitMQConstants.RABBITMQ_DIRECT_REPLY_EXCHANGE); // use default exchange for reply-to messages
+ message.setHeader(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME, RabbitMQConstants.RABBITMQ_DIRECT_REPLY_EXCHANGE); // use default exchange for reply-to messages
}
return message;
@@ -95,7 +96,7 @@ public class RabbitMQMessagePublisher {
properties = endpoint.getMessageConverter().buildProperties(camelExchange).build();
body = null;
} else {
- LOG.warn("Could not convert {} to byte[]", message.getBody());
+ LOG.warn("Cannot convert {} to byte[]", message.getBody());
throw new RuntimeCamelException(e);
}
}
@@ -109,7 +110,7 @@ public class RabbitMQMessagePublisher {
Boolean mandatory = camelExchange.getIn().getHeader(RabbitMQConstants.MANDATORY, endpoint.isMandatory(), Boolean.class);
Boolean immediate = camelExchange.getIn().getHeader(RabbitMQConstants.IMMEDIATE, endpoint.isImmediate(), Boolean.class);
- LOG.debug("Sending message to exchange: {} with CorrelationId = {}", rabbitExchange, properties.getCorrelationId());
+ LOG.debug("Sending message to exchange: {} with CorrelationId: {}", rabbitExchange, properties.getCorrelationId());
if (isPublisherAcknowledgements()) {
channel.confirmSelect();
@@ -153,7 +154,7 @@ public class RabbitMQMessagePublisher {
o.writeObject(msg.getBody());
return b.toByteArray();
} catch (NotSerializableException nse) {
- LOG.warn("Can not send object " + msg.getBody().getClass() + " via RabbitMQ because it contains non-serializable objects.");
+ LOG.warn("Cannot send object {} via RabbitMQ because it contains non-serializable objects.", msg.getBody().getClass());
throw new RuntimeCamelException(nse);
}
}
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 759039d..a1cc075 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
@@ -220,10 +220,13 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
in.setHeader(RabbitMQConstants.REPLY_TO, replyManager.getReplyTo());
- String exchangeName = in.getHeader(RabbitMQConstants.EXCHANGE_NAME, String.class);
- // If it is BridgeEndpoint we should ignore the message header of EXCHANGE_NAME
+ // remove the OVERRIDE header so it does not propagate
+ String exchangeName = (String) exchange.getIn().removeHeader(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME);
+ // If it is BridgeEndpoint we should ignore the message header of EXCHANGE_OVERRIDE_NAME
if (exchangeName == null || getEndpoint().isBridgeEndpoint()) {
exchangeName = getEndpoint().getExchangeName();
+ } else {
+ log.debug("Overriding header: {} detected sending message to exchange: {}", RabbitMQConstants.EXCHANGE_OVERRIDE_NAME, exchangeName);
}
String key = in.getHeader(RabbitMQConstants.ROUTING_KEY, String.class);
@@ -249,7 +252,14 @@ public class RabbitMQProducer extends DefaultAsyncProducer {
}
private boolean processInOnly(Exchange exchange, AsyncCallback callback) throws Exception {
- String exchangeName = getEndpoint().getExchangeName(exchange.getIn());
+ // remove the OVERRIDE header so it does not propagate
+ String exchangeName = (String) exchange.getIn().removeHeader(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME);
+ // If it is BridgeEndpoint we should ignore the message header of EXCHANGE_OVERRIDE_NAME
+ if (exchangeName == null || getEndpoint().isBridgeEndpoint()) {
+ exchangeName = getEndpoint().getExchangeName();
+ } else {
+ log.debug("Overriding header: {} detected sending message to exchange: {}", RabbitMQConstants.EXCHANGE_OVERRIDE_NAME, exchangeName);
+ }
String key = exchange.getIn().getHeader(RabbitMQConstants.ROUTING_KEY, String.class);
// we just need to make sure RoutingKey option take effect if it is not BridgeEndpoint
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 03/03: CAMEL-12329: Fixed rabbitmq integration tests
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit fde3332ef6660e5007aecef638abe8b447a1c915
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Mar 7 10:34:11 2018 +0100
CAMEL-12329: Fixed rabbitmq integration tests
---
components/camel-rabbitmq/readme.txt | 16 ++++++++++++----
...bbitMQIntBasicTest.java => RabbitMQBasicIntTest.java} | 5 ++---
.../rabbitmq/RabbitMQConsumerIntTestReplyTo.java | 2 +-
.../rabbitmq/RabbitMQRequeueHandledExceptionIntTest.java | 4 ++--
.../camel/component/rabbitmq/RabbitMQRequeueIntTest.java | 4 ++--
.../RabbitMQRequeueUnhandledExceptionIntTest.java | 4 ++--
components/camel-rabbitmq/src/test/resources/README.md | 5 -----
7 files changed, 21 insertions(+), 19 deletions(-)
diff --git a/components/camel-rabbitmq/readme.txt b/components/camel-rabbitmq/readme.txt
index b95b015..ef005d5 100644
--- a/components/camel-rabbitmq/readme.txt
+++ b/components/camel-rabbitmq/readme.txt
@@ -2,12 +2,20 @@ Integration testing
===================
The camel-rabbitmq component has both unit tests and integration tests.
-The integration tests requires a running RabbitMQ broker. This broker can be run via Docker
- docker run -d -p 5672:5672 -e RABBITMQ_DEFAULT_USER=cameltest -e RABBITMQ_DEFAULT_PASS=cameltest --hostname my-rabbit --name some-rabbit rabbitmq:3
+The integration tests requires a running RabbitMQ broker.
-And then the integration tests can be run via Maven
+The broker can be run via Docker:
- mvn test -P itest
+ docker run -it -p 5672:5672 -e RABBITMQ_DEFAULT_USER=cameltest -e RABBITMQ_DEFAULT_PASS=cameltest --hostname my-rabbit --name some-rabbit rabbitmq:3
+
+Or to install RabbitMQ as standalone and then configure it:
+
+ rabbitmq-server
+ rabbitmqctl add_user cameltest cameltest
+ rabbitmqctl set_permissions -p / cameltest ".*" ".*" ".*"
+The integration tests can be run via Maven:
+
+ mvn test -P itest
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQBasicIntTest.java
similarity index 91%
rename from components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java
rename to components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQBasicIntTest.java
index 209f940..4d4ad74 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQIntBasicTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQBasicIntTest.java
@@ -22,10 +22,9 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-public class RabbitMQIntBasicTest extends AbstractRabbitMQIntTest {
+public class RabbitMQBasicIntTest extends AbstractRabbitMQIntTest {
- // Startup RabbitMQ via Docker
- // docker run -d -it -p 5672:5672 -e RABBITMQ_DEFAULT_USER=cameltest -e RABBITMQ_DEFAULT_PASS=cameltest --hostname my-rabbit --name some-rabbit rabbitmq:3
+ // Startup RabbitMQ via Docker (see readme.txt in camel-rabbitmq folder)
@EndpointInject(uri = "rabbitmq:localhost:5672/foo?username=cameltest&password=cameltest")
private Endpoint foo;
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQConsumerIntTestReplyTo.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQConsumerIntTestReplyTo.java
index 07c2d6e..1f3bd86 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQConsumerIntTestReplyTo.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQConsumerIntTestReplyTo.java
@@ -46,7 +46,7 @@ public class RabbitMQConsumerIntTestReplyTo extends AbstractRabbitMQIntTest {
protected Channel channel;
- @EndpointInject(uri = "rabbitmq:localhost:5672/" + EXCHANGE + "?routingKey=" + ROUTING_KEY)
+ @EndpointInject(uri = "rabbitmq:localhost:5672/" + EXCHANGE + "?username=cameltest&password=cameltest&routingKey=" + ROUTING_KEY)
private Endpoint from;
private Connection connection;
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
index a6e1d73..cea64be 100644
--- 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
@@ -34,8 +34,8 @@ public class RabbitMQRequeueHandledExceptionIntTest extends CamelTestSupport {
@Produce(uri = "direct:rabbitMQ")
protected ProducerTemplate directProducer;
- @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
- + "autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+ @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?username=cameltest&password=cameltest"
+ + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
private Endpoint rabbitMQEndpoint;
@EndpointInject(uri = "mock:producing")
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 cbaf3d0..1de23f0 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
@@ -33,8 +33,8 @@ public class RabbitMQRequeueIntTest extends AbstractRabbitMQIntTest {
@Produce(uri = "direct:rabbitMQ")
protected ProducerTemplate directProducer;
- @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
- + "autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+ @EndpointInject(uri = "rabbitmq:localhost:5672/ex4??username=cameltest&password=cameltest"
+ + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
private Endpoint rabbitMQEndpoint;
@EndpointInject(uri = "mock:producing")
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
index 2e18a4e..c0fea76 100644
--- 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
@@ -33,8 +33,8 @@ public class RabbitMQRequeueUnhandledExceptionIntTest extends AbstractRabbitMQIn
@Produce(uri = "direct:rabbitMQ")
protected ProducerTemplate directProducer;
- @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?"
- + "autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+ @EndpointInject(uri = "rabbitmq:localhost:5672/ex4?username=cameltest&password=cameltest"
+ + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
private Endpoint rabbitMQEndpoint;
@EndpointInject(uri = "mock:producing")
diff --git a/components/camel-rabbitmq/src/test/resources/README.md b/components/camel-rabbitmq/src/test/resources/README.md
deleted file mode 100644
index 89f3d1b..0000000
--- a/components/camel-rabbitmq/src/test/resources/README.md
+++ /dev/null
@@ -1,5 +0,0 @@
-How to configure RabbitMQ server for integration tests
------------------------
-rabbitmq-server
-rabbitmqctl add_user cameltest cameltest
-rabbitmqctl set_permissions -p / cameltest ".*" ".*" ".*"
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.