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 2015/01/11 16:24:29 UTC
[1/2] camel git commit: CAMEL-8227: Using exchangePattern=xxx should
override pattern in send/recipient list when sending to the destination.
Repository: camel
Updated Branches:
refs/heads/master 21f69939d -> 88c7fe413
CAMEL-8227: Using exchangePattern=xxx should override pattern in send/recipient list when sending to the destination.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/54d7fc59
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/54d7fc59
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/54d7fc59
Branch: refs/heads/master
Commit: 54d7fc594380cdbf8630249aab8aacbb8c225449
Parents: 21f6993
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Jan 11 11:29:29 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Sun Jan 11 11:29:29 2015 +0100
----------------------------------------------------------------------
.../camel/processor/RecipientListProcessor.java | 15 ++--
.../apache/camel/processor/SendProcessor.java | 22 +++--
.../org/apache/camel/util/EndpointHelper.java | 22 ++++-
.../RecipientListExchangePatternOptionTest.java | 37 ++++++++
.../SendExchangePatternOptionTest.java | 63 ++++++++++++++
.../jms/issues/JmsInOnlyParameterTest.java | 88 ++++++++++++++++++++
6 files changed, 230 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
index 1087da6..db6af86 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
@@ -17,12 +17,11 @@
package org.apache.camel.processor;
import java.io.UnsupportedEncodingException;
-import java.net.URI;
+import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ExecutorService;
import org.apache.camel.CamelContext;
@@ -34,6 +33,7 @@ import org.apache.camel.Producer;
import org.apache.camel.impl.ProducerCache;
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.apache.camel.spi.RouteContext;
+import org.apache.camel.util.EndpointHelper;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.MessageHelper;
import org.apache.camel.util.ServiceHelper;
@@ -191,7 +191,7 @@ public class RecipientListProcessor extends MulticastProcessor {
ExchangePattern pattern;
try {
endpoint = resolveEndpoint(exchange, recipient);
- pattern = resolveExchangePattern(exchange, recipient);
+ pattern = resolveExchangePattern(recipient);
producer = producerCache.acquireProducer(endpoint);
} catch (Exception e) {
if (isIgnoreInvalidEndpoints()) {
@@ -254,18 +254,13 @@ public class RecipientListProcessor extends MulticastProcessor {
return ExchangeHelper.resolveEndpoint(exchange, recipient);
}
- protected ExchangePattern resolveExchangePattern(Exchange exchange, Object recipient) throws UnsupportedEncodingException, URISyntaxException {
+ protected ExchangePattern resolveExchangePattern(Object recipient) throws UnsupportedEncodingException, URISyntaxException, MalformedURLException {
// trim strings as end users might have added spaces between separators
if (recipient instanceof String) {
String s = ((String) recipient).trim();
// see if exchangePattern is a parameter in the url
s = URISupport.normalizeUri(s);
- URI url = new URI(s);
- Map<String, Object> parameters = URISupport.parseParameters(url);
- String pattern = (String) parameters.get("exchangePattern");
- if (pattern != null) {
- return ExchangePattern.asEnum(pattern);
- }
+ return EndpointHelper.resolveExchangePatternFromUrl(s);
}
return null;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
index d8e8803..efd6a60 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.processor;
+import java.net.URISyntaxException;
import java.util.HashMap;
import org.apache.camel.AsyncCallback;
@@ -33,6 +34,7 @@ import org.apache.camel.impl.ProducerCache;
import org.apache.camel.support.ServiceSupport;
import org.apache.camel.util.AsyncProcessorConverterHelper;
import org.apache.camel.util.AsyncProcessorHelper;
+import org.apache.camel.util.EndpointHelper;
import org.apache.camel.util.EventHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ServiceHelper;
@@ -53,6 +55,7 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra
protected ProducerCache producerCache;
protected AsyncProcessor producer;
protected Endpoint destination;
+ protected ExchangePattern destinationExchangePattern;
protected final boolean unhandleException;
public SendProcessor(Endpoint destination) {
@@ -69,6 +72,12 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra
this.camelContext = destination.getCamelContext();
this.pattern = pattern;
this.unhandleException = unhandleException;
+ try {
+ this.destinationExchangePattern = null;
+ this.destinationExchangePattern = EndpointHelper.resolveExchangePatternFromUrl(destination.getEndpointUri());
+ } catch (URISyntaxException e) {
+ throw ObjectHelper.wrapRuntimeCamelException(e);
+ }
ObjectHelper.notNull(this.camelContext, "camelContext");
}
@@ -133,11 +142,9 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra
}
});
} catch (Throwable throwable) {
- if (exchange != null) {
- exchange.setException(throwable);
- checkException(exchange);
- }
-
+ exchange.setException(throwable);
+ checkException(exchange);
+ callback.done(sync);
}
return sync;
@@ -180,7 +187,10 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra
}
protected Exchange configureExchange(Exchange exchange, ExchangePattern pattern) {
- if (pattern != null) {
+ // destination exchange pattern overrides pattern
+ if (destinationExchangePattern != null) {
+ exchange.setPattern(destinationExchangePattern);
+ } else if (pattern != null) {
exchange.setPattern(pattern);
}
// set property which endpoint we send to
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/camel-core/src/main/java/org/apache/camel/util/EndpointHelper.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/util/EndpointHelper.java b/camel-core/src/main/java/org/apache/camel/util/EndpointHelper.java
index 9df3ca2..18181b3 100644
--- a/camel-core/src/main/java/org/apache/camel/util/EndpointHelper.java
+++ b/camel-core/src/main/java/org/apache/camel/util/EndpointHelper.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.util;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -29,6 +31,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.DelegateEndpoint;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
@@ -486,5 +489,22 @@ public final class EndpointHelper {
sb.append("\n</messages>");
return sb.toString();
}
-
+
+ /**
+ * Attempts to resolve if the url has an <tt>exchangePattern</tt> option configured
+ *
+ * @param url the url
+ * @return the exchange pattern, or <tt>null</tt> if the url has no <tt>exchangePattern</tt> configured.
+ * @throws URISyntaxException is thrown if uri is invalid
+ */
+ public static ExchangePattern resolveExchangePatternFromUrl(String url) throws URISyntaxException {
+ URI uri = new URI(url);
+ Map<String, Object> parameters = URISupport.parseParameters(uri);
+ String pattern = (String) parameters.get("exchangePattern");
+ if (pattern != null) {
+ return ExchangePattern.asEnum(pattern);
+ }
+ return null;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/camel-core/src/test/java/org/apache/camel/processor/RecipientListExchangePatternOptionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/RecipientListExchangePatternOptionTest.java b/camel-core/src/test/java/org/apache/camel/processor/RecipientListExchangePatternOptionTest.java
new file mode 100644
index 0000000..dc7f7c5
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/RecipientListExchangePatternOptionTest.java
@@ -0,0 +1,37 @@
+/**
+ * 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.processor;
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class RecipientListExchangePatternOptionTest extends SendExchangePatternOptionTest {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start")
+ .recipientList(constant("stub:foo?exchangePattern=InOnly"))
+ .to("mock:result");
+
+ from("stub:foo")
+ .to("mock:stub");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/camel-core/src/test/java/org/apache/camel/processor/SendExchangePatternOptionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SendExchangePatternOptionTest.java b/camel-core/src/test/java/org/apache/camel/processor/SendExchangePatternOptionTest.java
new file mode 100644
index 0000000..63db2a8
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SendExchangePatternOptionTest.java
@@ -0,0 +1,63 @@
+/**
+ * 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.processor;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+
+public class SendExchangePatternOptionTest extends ContextTestSupport {
+
+ public void testExchangePatternOptionInOnly() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ getMockEndpoint("mock:result").message(0).exchangePattern().isEqualTo(ExchangePattern.InOnly);
+
+ getMockEndpoint("mock:stub").expectedMessageCount(1);
+ getMockEndpoint("mock:stub").message(0).exchangePattern().isEqualTo(ExchangePattern.InOnly);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testExchangePatternOptionInOut() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ getMockEndpoint("mock:result").message(0).exchangePattern().isEqualTo(ExchangePattern.InOut);
+
+ getMockEndpoint("mock:stub").expectedMessageCount(1);
+ getMockEndpoint("mock:stub").message(0).exchangePattern().isEqualTo(ExchangePattern.InOnly);
+
+ template.requestBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start")
+ .to("stub:foo?exchangePattern=InOnly")
+ .to("mock:result");
+
+ from("stub:foo")
+ .to("mock:stub");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/54d7fc59/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOnlyParameterTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOnlyParameterTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOnlyParameterTest.java
new file mode 100644
index 0000000..ba474e3
--- /dev/null
+++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOnlyParameterTest.java
@@ -0,0 +1,88 @@
+/**
+ * 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.jms.issues;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jms.CamelJmsTestHelper;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * @version
+ */
+public class JmsInOnlyParameterTest extends CamelTestSupport {
+
+ @Test
+ public void testInOnly() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:result").message(0).header("JMSCorrelationID").isNull();
+ getMockEndpoint("mock:result").message(0).exchangePattern().isEqualTo(ExchangePattern.InOut);
+
+ getMockEndpoint("mock:in").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:in").message(0).header("JMSCorrelationID").isNull();
+ getMockEndpoint("mock:in").message(0).exchangePattern().isEqualTo(ExchangePattern.InOnly);
+
+ String out = template.requestBody("direct:start", "Hello World", String.class);
+ assertEquals("Bye World", out);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testInOnlyWithJMSCorrelationID() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:result").message(0).header("JMSCorrelationID").isEqualTo("foobar");
+ getMockEndpoint("mock:result").message(0).exchangePattern().isEqualTo(ExchangePattern.InOut);
+
+ getMockEndpoint("mock:in").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:in").message(0).header("JMSCorrelationID").isEqualTo("foobar");
+ getMockEndpoint("mock:in").message(0).exchangePattern().isEqualTo(ExchangePattern.InOnly);
+
+ String out = template.requestBodyAndHeader("direct:start", "Hello World", "JMSCorrelationID", "foobar", String.class);
+ assertEquals("Bye World", out);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent("activemq", jmsComponentAutoAcknowledge(connectionFactory));
+ return camelContext;
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+ from("direct:start")
+ .to("activemq:queue:in?exchangePattern=InOnly")
+ .transform().constant("Bye World")
+ .to("mock:result");
+
+ from("activemq:queue:in")
+ .to("mock:in");
+ }
+ };
+ }
+
+}
\ No newline at end of file
[2/2] camel git commit: CAMEL-8204 Added the putIfAbsent method to
avoid overriding the old key value as Claus suggested
Posted by da...@apache.org.
CAMEL-8204 Added the putIfAbsent method to avoid overriding the old key value as Claus suggested
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/88c7fe41
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/88c7fe41
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/88c7fe41
Branch: refs/heads/master
Commit: 88c7fe413ad5cbdfdad6b64a7878a8c0aeca67d8
Parents: 54d7fc5
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Jan 11 14:04:24 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Sun Jan 11 14:04:24 2015 +0100
----------------------------------------------------------------------
.../jms/reply/CorrelationTimeoutMap.java | 25 ++++++++++++++++++++
.../jms/reply/ReplyManagerSupport.java | 6 ++---
2 files changed, 28 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/88c7fe41/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
index d72c8d0..e0e6731 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
@@ -91,6 +91,31 @@ public class CorrelationTimeoutMap extends DefaultTimeoutMap<String, ReplyHandle
}
@Override
+ public ReplyHandler putIfAbsent(String key, ReplyHandler value, long timeoutMillis) {
+ try {
+ if (listener != null) {
+ listener.onPut(key);
+ }
+ } catch (Throwable e) {
+ // ignore
+ }
+
+ ReplyHandler result;
+ if (timeoutMillis <= 0) {
+ // no timeout (must use Integer.MAX_VALUE)
+ result = super.putIfAbsent(key, value, Integer.MAX_VALUE);
+ } else {
+ result = super.putIfAbsent(key, value, timeoutMillis);
+ }
+ if (result == null) {
+ log.trace("Added correlationID: {} to timeout after: {} millis", key, timeoutMillis);
+ } else {
+ log.trace("Duplicate correlationID: {} detected", key);
+ }
+ return result;
+ }
+
+ @Override
public ReplyHandler remove(String key) {
try {
if (listener != null) {
http://git-wip-us.apache.org/repos/asf/camel/blob/88c7fe41/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
index 30e5919..83a7729 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
@@ -142,13 +142,13 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
if (timeout) {
// timeout occurred do a WARN log so its easier to spot in the logs
if (log.isWarnEnabled()) {
- log.warn("Timeout occurred after {} millis waiting for reply message with correlationID [{}]."
+ log.warn("Timeout occurred after {} millis waiting for reply message with correlationID [{}] on destination {}."
+ " Setting ExchangeTimedOutException on {} and continue routing.",
- new Object[]{holder.getRequestTimeout(), holder.getCorrelationId(), ExchangeHelper.logIds(exchange)});
+ new Object[]{holder.getRequestTimeout(), holder.getCorrelationId(), replyTo, ExchangeHelper.logIds(exchange)});
}
// no response, so lets set a timed out exception
- String msg = "reply message with correlationID: " + holder.getCorrelationId() + " not received";
+ String msg = "reply message with correlationID: " + holder.getCorrelationId() + " not received on destination: " + replyTo;
exchange.setException(new ExchangeTimedOutException(exchange, holder.getRequestTimeout(), msg));
} else {
JmsMessage response = new JmsMessage(message, endpoint.getBinding());