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 2017/09/07 12:57:45 UTC
[3/3] camel git commit: CAMEL-11690: Fixed routing slip and dynamic
router EIP to not trigger multiple async callback done on the other callback.
CAMEL-11690: Fixed routing slip and dynamic router EIP to not trigger multiple async callback done on the other callback.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/431747a1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/431747a1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/431747a1
Branch: refs/heads/camel-2.18.x
Commit: 431747a1b9e582d89f389a4bd2b1e7aa5e700969
Parents: 01b02b4
Author: Claus Ibsen <da...@apache.org>
Authored: Thu Sep 7 14:40:50 2017 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Thu Sep 7 14:57:33 2017 +0200
----------------------------------------------------------------------
.../apache/camel/processor/DynamicRouter.java | 2 +-
.../org/apache/camel/processor/RoutingSlip.java | 130 ++++++++++--------
.../DynamicRouterWithInterceptorTest.java | 131 +++++++++++++++++++
.../RecipientListWithInterceptorTest.java | 118 +++++++++++++++++
.../issues/RoutingSlipWithInterceptorTest.java | 117 +++++++++++++++++
.../camel/issues/SplitWithInterceptorTest.java | 109 +++++++++++++++
.../DynamicRouterEventNotifierTest.java | 114 ++++++++++++++++
.../RecipientListEventNotifierTest.java | 98 ++++++++++++++
8 files changed, 763 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/main/java/org/apache/camel/processor/DynamicRouter.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/DynamicRouter.java b/camel-core/src/main/java/org/apache/camel/processor/DynamicRouter.java
index 565c64e..c37cec6 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/DynamicRouter.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/DynamicRouter.java
@@ -69,7 +69,7 @@ public class DynamicRouter extends RoutingSlip {
return false;
}
current = ObjectHelper.createIterator(routingSlip, uriDelimiter);
- return current != null && current.hasNext();
+ return current.hasNext();
}
public Object next(Exchange exchange) {
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java b/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java
index c081d46..88aae16 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java
@@ -203,14 +203,14 @@ public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Trace
};
}
- private boolean doRoutingSlipWithExpression(final Exchange exchange, final Expression expression, final AsyncCallback callback) {
+ private boolean doRoutingSlipWithExpression(final Exchange exchange, final Expression expression, final AsyncCallback originalCallback) {
Exchange current = exchange;
RoutingSlipIterator iter;
try {
iter = createRoutingSlipIterator(exchange, expression);
} catch (Exception e) {
exchange.setException(e);
- callback.done(true);
+ originalCallback.done(true);
return true;
}
@@ -234,7 +234,7 @@ public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Trace
}
//process and prepare the routing slip
- boolean sync = processExchange(endpoint, current, exchange, callback, iter);
+ boolean sync = processExchange(endpoint, current, exchange, originalCallback, iter);
current = prepareExchangeForRoutingSlip(current, endpoint);
if (!sync) {
@@ -272,7 +272,9 @@ public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Trace
// copy results back to the original exchange
ExchangeHelper.copyResults(exchange, current);
- callback.done(true);
+ // okay we are completely done with the routing slip
+ // so we need to signal done on the original callback so it can continue
+ originalCallback.done(true);
return true;
}
@@ -338,11 +340,23 @@ public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Trace
}
protected boolean processExchange(final Endpoint endpoint, final Exchange exchange, final Exchange original,
- final AsyncCallback callback, final RoutingSlipIterator iter) {
+ final AsyncCallback originalCallback, final RoutingSlipIterator iter) {
// this does the actual processing so log at trace level
log.trace("Processing exchangeId: {} >>> {}", exchange.getExchangeId(), exchange);
+ // routing slip callback which are used when
+ // - routing slip was routed asynchronously
+ // - and we are completely done with the routing slip
+ // so we need to signal done on the original callback so it can continue
+ AsyncCallback callback = new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ if (!doneSync) {
+ originalCallback.done(false);
+ }
+ }
+ };
boolean sync = producerCache.doInAsyncProducer(endpoint, exchange, null, callback, new AsyncProducerCallback() {
public boolean doInAsyncProducer(Producer producer, AsyncProcessor asyncProducer, final Exchange exchange,
ExchangePattern exchangePattern, final AsyncCallback callback) {
@@ -359,73 +373,79 @@ public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Trace
public void done(boolean doneSync) {
// we only have to handle async completion of the routing slip
if (doneSync) {
- callback.done(doneSync);
+ callback.done(true);
return;
}
- // continue processing the routing slip asynchronously
- Exchange current = prepareExchangeForRoutingSlip(exchange, endpoint);
-
- while (iter.hasNext(current)) {
-
- // we ignore some kind of exceptions and allow us to continue
- if (isIgnoreInvalidEndpoints()) {
- FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
- if (e != null) {
- if (log.isDebugEnabled()) {
- log.debug("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
+ try {
+ // continue processing the routing slip asynchronously
+ Exchange current = prepareExchangeForRoutingSlip(exchange, endpoint);
+
+ while (iter.hasNext(current)) {
+
+ // we ignore some kind of exceptions and allow us to continue
+ if (isIgnoreInvalidEndpoints()) {
+ FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
+ if (e != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
+ }
+ current.setException(null);
}
- current.setException(null);
}
- }
- // Decide whether to continue with the recipients or not; similar logic to the Pipeline
- // check for error if so we should break out
- if (!continueProcessing(current, "so breaking out of the routing slip", log)) {
- break;
- }
+ // Decide whether to continue with the recipients or not; similar logic to the Pipeline
+ // check for error if so we should break out
+ if (!continueProcessing(current, "so breaking out of the routing slip", log)) {
+ break;
+ }
- Endpoint endpoint;
- try {
- endpoint = resolveEndpoint(iter, exchange);
- // if no endpoint was resolved then try the next
- if (endpoint == null) {
- continue;
+ Endpoint endpoint;
+ try {
+ endpoint = resolveEndpoint(iter, exchange);
+ // if no endpoint was resolved then try the next
+ if (endpoint == null) {
+ continue;
+ }
+ } catch (Exception e) {
+ // error resolving endpoint so we should break out
+ exchange.setException(e);
+ break;
}
- } catch (Exception e) {
- // error resolving endpoint so we should break out
- exchange.setException(e);
- break;
- }
- // prepare and process the routing slip
- boolean sync = processExchange(endpoint, current, original, callback, iter);
- current = prepareExchangeForRoutingSlip(current, endpoint);
+ // prepare and process the routing slip
+ boolean sync = processExchange(endpoint, current, original, callback, iter);
+ current = prepareExchangeForRoutingSlip(current, endpoint);
- if (!sync) {
- log.trace("Processing exchangeId: {} is continued being processed asynchronously", original.getExchangeId());
- return;
+ if (!sync) {
+ log.trace("Processing exchangeId: {} is continued being processed asynchronously", original.getExchangeId());
+ return;
+ }
}
- }
- // logging nextExchange as it contains the exchange that might have altered the payload and since
- // we are logging the completion if will be confusing if we log the original instead
- // we could also consider logging the original and the nextExchange then we have *before* and *after* snapshots
- log.trace("Processing complete for exchangeId: {} >>> {}", original.getExchangeId(), current);
+ // logging nextExchange as it contains the exchange that might have altered the payload and since
+ // we are logging the completion if will be confusing if we log the original instead
+ // we could also consider logging the original and the nextExchange then we have *before* and *after* snapshots
+ log.trace("Processing complete for exchangeId: {} >>> {}", original.getExchangeId(), current);
- // copy results back to the original exchange
- ExchangeHelper.copyResults(original, current);
+ // copy results back to the original exchange
+ ExchangeHelper.copyResults(original, current);
- if (target instanceof DeadLetterChannel) {
- Processor deadLetter = ((DeadLetterChannel) target).getDeadLetter();
- try {
- ServiceHelper.stopService(deadLetter);
- } catch (Exception e) {
- log.warn("Error stopping DeadLetterChannel error handler on routing slip. This exception is ignored.", e);
+ if (target instanceof DeadLetterChannel) {
+ Processor deadLetter = ((DeadLetterChannel) target).getDeadLetter();
+ try {
+ ServiceHelper.stopService(deadLetter);
+ } catch (Exception e) {
+ log.warn("Error stopping DeadLetterChannel error handler on routing slip. This exception is ignored.", e);
+ }
}
+ } catch (Throwable e) {
+ exchange.setException(e);
}
- callback.done(false);
+ // okay we are completely done with the routing slip
+ // so we need to signal done on the original callback so it can continue
+ originalCallback.done(false);
}
});
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/issues/DynamicRouterWithInterceptorTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/DynamicRouterWithInterceptorTest.java b/camel-core/src/test/java/org/apache/camel/issues/DynamicRouterWithInterceptorTest.java
new file mode 100644
index 0000000..78adbeb
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/DynamicRouterWithInterceptorTest.java
@@ -0,0 +1,131 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Body;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Header;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.DynamicRouterDefinition;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RecipientListDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DynamicRouterWithInterceptorTest extends ContextTestSupport {
+
+ private final MyInterceptStrategy interceptStrategy = new MyInterceptStrategy();
+
+ public static class MyInterceptStrategy implements InterceptStrategy {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MyInterceptStrategy.class);
+ private static int doneCount;
+
+ public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
+ final Processor target, final Processor nextTarget) throws Exception {
+ if (definition instanceof DynamicRouterDefinition<?>) {
+ final DelegateAsyncProcessor delegateAsyncProcessor = new DelegateAsyncProcessor() {
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ LOGGER.info("I'm doing someting");
+ return super.process(exchange, new AsyncCallback() {
+ public void done(final boolean doneSync) {
+ LOGGER.info("I'm done");
+ doneCount++;
+ callback.done(doneSync);
+ }
+ });
+ }
+ };
+ delegateAsyncProcessor.setProcessor(target);
+ return delegateAsyncProcessor;
+ }
+ return new DelegateAsyncProcessor(target);
+ }
+
+ public void reset() {
+ doneCount = 0;
+ }
+ }
+
+ public void testDynamicRouterOne() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ public void testDynamicRouterTwo() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addInterceptStrategy(interceptStrategy);
+
+ from("direct:start")
+ .dynamicRouter(method(DynamicRouterWithInterceptorTest.class, "slip"))
+ .to("mock:result");
+
+ from("direct:foo")
+ .to("log:foo")
+ .to("mock:foo");
+
+ from("direct:bar")
+ .to("log:bar")
+ .to("mock:bar");
+ }
+ };
+ }
+
+ public String slip(@Body String body, @Header(Exchange.SLIP_ENDPOINT) String previous) {
+ if (previous == null) {
+ return "direct:foo";
+ } else if ("Bye World".equals(body) && "direct://foo".equals(previous)) {
+ return "direct:bar";
+ }
+
+ // no more so return null
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/issues/RecipientListWithInterceptorTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/RecipientListWithInterceptorTest.java b/camel-core/src/test/java/org/apache/camel/issues/RecipientListWithInterceptorTest.java
new file mode 100644
index 0000000..e41d1a8
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/RecipientListWithInterceptorTest.java
@@ -0,0 +1,118 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RecipientListDefinition;
+import org.apache.camel.model.RoutingSlipDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RecipientListWithInterceptorTest extends ContextTestSupport {
+
+ private final MyInterceptStrategy interceptStrategy = new MyInterceptStrategy();
+
+ public static class MyInterceptStrategy implements InterceptStrategy {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MyInterceptStrategy.class);
+ private static int doneCount;
+
+ public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
+ final Processor target, final Processor nextTarget) throws Exception {
+ if (definition instanceof RecipientListDefinition<?>) {
+ final DelegateAsyncProcessor delegateAsyncProcessor = new DelegateAsyncProcessor() {
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ LOGGER.info("I'm doing someting");
+ return super.process(exchange, new AsyncCallback() {
+ public void done(final boolean doneSync) {
+ LOGGER.info("I'm done");
+ doneCount++;
+ callback.done(doneSync);
+ }
+ });
+ }
+ };
+ delegateAsyncProcessor.setProcessor(target);
+ return delegateAsyncProcessor;
+ }
+ return new DelegateAsyncProcessor(target);
+ }
+
+ public void reset() {
+ doneCount = 0;
+ }
+ }
+
+ public void testRecipientListOne() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "slip", "direct:foo");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ public void testRecipientListTwo() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "slip", "direct:foo,direct:bar");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addInterceptStrategy(interceptStrategy);
+
+ from("direct:start")
+ .recipientList(header("slip"))
+ .to("mock:result");
+
+ from("direct:foo")
+ .to("log:foo")
+ .to("mock:foo");
+
+ from("direct:bar")
+ .to("log:bar")
+ .to("mock:bar");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/issues/RoutingSlipWithInterceptorTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/RoutingSlipWithInterceptorTest.java b/camel-core/src/test/java/org/apache/camel/issues/RoutingSlipWithInterceptorTest.java
new file mode 100644
index 0000000..7c3cbf4
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/RoutingSlipWithInterceptorTest.java
@@ -0,0 +1,117 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RoutingSlipDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RoutingSlipWithInterceptorTest extends ContextTestSupport {
+
+ private final MyInterceptStrategy interceptStrategy = new MyInterceptStrategy();
+
+ public static class MyInterceptStrategy implements InterceptStrategy {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MyInterceptStrategy.class);
+ private static int doneCount;
+
+ public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
+ final Processor target, final Processor nextTarget) throws Exception {
+ if (definition instanceof RoutingSlipDefinition<?>) {
+ final DelegateAsyncProcessor delegateAsyncProcessor = new DelegateAsyncProcessor() {
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ LOGGER.info("I'm doing someting");
+ return super.process(exchange, new AsyncCallback() {
+ public void done(final boolean doneSync) {
+ LOGGER.info("I'm done");
+ doneCount++;
+ callback.done(doneSync);
+ }
+ });
+ }
+ };
+ delegateAsyncProcessor.setProcessor(target);
+ return delegateAsyncProcessor;
+ }
+ return new DelegateAsyncProcessor(target);
+ }
+
+ public void reset() {
+ doneCount = 0;
+ }
+ }
+
+ public void testRoutingSlipOne() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "slip", "direct:foo");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ public void testRoutingSlipTwo() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "slip", "direct:foo,direct:bar");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addInterceptStrategy(interceptStrategy);
+
+ from("direct:start")
+ .routingSlip(header("slip"))
+ .to("mock:result");
+
+ from("direct:foo")
+ .to("log:foo")
+ .to("mock:foo");
+
+ from("direct:bar")
+ .to("log:bar")
+ .to("mock:bar");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/issues/SplitWithInterceptorTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/SplitWithInterceptorTest.java b/camel-core/src/test/java/org/apache/camel/issues/SplitWithInterceptorTest.java
new file mode 100644
index 0000000..95fd8c0
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/SplitWithInterceptorTest.java
@@ -0,0 +1,109 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RoutingSlipDefinition;
+import org.apache.camel.model.SplitDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SplitWithInterceptorTest extends ContextTestSupport {
+
+ private final MyInterceptStrategy interceptStrategy = new MyInterceptStrategy();
+
+ public static class MyInterceptStrategy implements InterceptStrategy {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MyInterceptStrategy.class);
+ private static int doneCount;
+
+ public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
+ final Processor target, final Processor nextTarget) throws Exception {
+ if (definition instanceof SplitDefinition) {
+ final DelegateAsyncProcessor delegateAsyncProcessor = new DelegateAsyncProcessor() {
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ LOGGER.info("I'm doing someting");
+ return super.process(exchange, new AsyncCallback() {
+ public void done(final boolean doneSync) {
+ LOGGER.info("I'm done");
+ doneCount++;
+ callback.done(doneSync);
+ }
+ });
+ }
+ };
+ delegateAsyncProcessor.setProcessor(target);
+ return delegateAsyncProcessor;
+ }
+ return new DelegateAsyncProcessor(target);
+ }
+
+ public void reset() {
+ doneCount = 0;
+ }
+ }
+
+ public void testSplitOne() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:line").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ public void testSplitTwo() throws Exception {
+ interceptStrategy.reset();
+
+ getMockEndpoint("mock:line").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World,Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Done method shall be called only once", 1, MyInterceptStrategy.doneCount);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addInterceptStrategy(interceptStrategy);
+
+ from("direct:start")
+ .split(body().tokenize(","))
+ .to("log:line")
+ .to("mock:line")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java b/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
new file mode 100644
index 0000000..f09214c
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
@@ -0,0 +1,114 @@
+/**
+ * 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.routingslip;
+
+import java.util.EventObject;
+
+import org.apache.camel.Body;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Header;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.management.event.ExchangeSendingEvent;
+import org.apache.camel.management.event.ExchangeSentEvent;
+import org.apache.camel.support.EventNotifierSupport;
+
+public class DynamicRouterEventNotifierTest extends ContextTestSupport {
+
+ private MyEventNotifier notifier = new MyEventNotifier();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getManagementStrategy().addEventNotifier(notifier);
+ return context;
+ }
+
+ public void testDynamicRouterEventNotifier() throws Exception {
+ getMockEndpoint("mock:x").expectedMessageCount(1);
+ getMockEndpoint("mock:y").expectedMessageCount(1);
+ getMockEndpoint("mock:z").expectedMessageCount(1);
+ getMockEndpoint("mock:end").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Should have 5 sending events", 5, notifier.getSending());
+ assertEquals("Should have 5 sent events", 5, notifier.getSent());
+ }
+
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start").dynamicRouter(method(DynamicRouterEventNotifierTest.class, "slip")).to("mock:end");
+ }
+ };
+ }
+
+ public String slip(@Body String body, @Header(Exchange.SLIP_ENDPOINT) String previous) {
+ if (previous == null) {
+ return "mock:x";
+ } else if ("mock://x".equals(previous)) {
+ return "mock:y";
+ } else if ("mock://y".equals(previous)) {
+ return "mock:z";
+ }
+
+ // no more so return null
+ return null;
+ }
+
+ private final class MyEventNotifier extends EventNotifierSupport {
+
+ private int sending;
+ private int sent;
+
+ @Override
+ public void notify(EventObject event) throws Exception {
+ if (event instanceof ExchangeSendingEvent) {
+ sending++;
+ } else {
+ sent++;
+ }
+ }
+
+ @Override
+ public boolean isEnabled(EventObject event) {
+ return event instanceof ExchangeSendingEvent || event instanceof ExchangeSentEvent;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ // noop
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // noop
+ }
+
+ public int getSending() {
+ return sending;
+ }
+
+ public int getSent() {
+ return sent;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/431747a1/camel-core/src/test/java/org/apache/camel/processor/routingslip/RecipientListEventNotifierTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/routingslip/RecipientListEventNotifierTest.java b/camel-core/src/test/java/org/apache/camel/processor/routingslip/RecipientListEventNotifierTest.java
new file mode 100644
index 0000000..c1a0969
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/routingslip/RecipientListEventNotifierTest.java
@@ -0,0 +1,98 @@
+/**
+ * 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.routingslip;
+
+import java.util.EventObject;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.management.event.ExchangeSendingEvent;
+import org.apache.camel.management.event.ExchangeSentEvent;
+import org.apache.camel.support.EventNotifierSupport;
+
+public class RecipientListEventNotifierTest extends ContextTestSupport {
+
+ private MyEventNotifier notifier = new MyEventNotifier();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getManagementStrategy().addEventNotifier(notifier);
+ return context;
+ }
+
+ public void testRecipientListEventNotifier() throws Exception {
+ getMockEndpoint("mock:x").expectedMessageCount(1);
+ getMockEndpoint("mock:y").expectedMessageCount(1);
+ getMockEndpoint("mock:z").expectedMessageCount(1);
+ getMockEndpoint("mock:end").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "myHeader", "mock:x,mock:y,mock:z");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals("Should have 5 sending events", 5, notifier.getSending());
+ assertEquals("Should have 5 sent events", 5, notifier.getSent());
+ }
+
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start").recipientList(header("myHeader")).to("mock:end");
+ }
+ };
+ }
+
+ private final class MyEventNotifier extends EventNotifierSupport {
+
+ private int sending;
+ private int sent;
+
+ @Override
+ public void notify(EventObject event) throws Exception {
+ if (event instanceof ExchangeSendingEvent) {
+ sending++;
+ } else {
+ sent++;
+ }
+ }
+
+ @Override
+ public boolean isEnabled(EventObject event) {
+ return event instanceof ExchangeSendingEvent || event instanceof ExchangeSentEvent;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ // noop
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // noop
+ }
+
+ public int getSending() {
+ return sending;
+ }
+
+ public int getSent() {
+ return sent;
+ }
+ }
+}