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 2011/01/06 16:30:54 UTC
svn commit: r1055913 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/processor/ test/java/org/apache/camel/processor/
test/java/org/apache/camel/processor/async/
Author: davsclaus
Date: Thu Jan 6 15:30:53 2011
New Revision: 1055913
URL: http://svn.apache.org/viewvc?rev=1055913&view=rev
Log:
CAMEL-3498: Fixed splitter aggregating on-the-fly on the input exchange, where as it should have used a defensive copy, to avoid any side effects during splitting, such as an exception being set to be propagated back to the input message when splitting is done.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingErrorHandlingTest.java
- copied, changed from r1055800, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingUoWIssueTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionErrorHandlingTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitFineGrainedErrorHandlingTest.java
- copied, changed from r1055800, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoAggregationStrategyTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoStopOnExceptionTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelNoStopOnExceptionTest.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java?rev=1055913&r1=1055912&r2=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java Thu Jan 6 15:30:53 2011
@@ -109,6 +109,9 @@ public class Splitter extends MulticastP
private Iterable<ProcessorExchangePair> createProcessorExchangePairsIterable(final Exchange exchange, final Object value) {
final Iterator iterator = ObjectHelper.createIterator(value);
return new Iterable() {
+ // create a copy which we use as master to copy during splitting
+ // this avoids any side effect reflected upon the incoming exchange
+ private final Exchange copy = ExchangeHelper.createCopy(exchange, true);
public Iterator iterator() {
return new Iterator() {
@@ -137,7 +140,8 @@ public class Splitter extends MulticastP
public Object next() {
Object part = iterator.next();
- Exchange newExchange = ExchangeHelper.createCopy(exchange, true);
+ // create a copy as the new exchange to be routed in the splitter from the copy
+ Exchange newExchange = ExchangeHelper.createCopy(copy, true);
if (part instanceof Message) {
newExchange.setIn((Message)part);
} else {
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoAggregationStrategyTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoAggregationStrategyTest.java?rev=1055913&r1=1055912&r2=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoAggregationStrategyTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoAggregationStrategyTest.java Thu Jan 6 15:30:53 2011
@@ -31,25 +31,25 @@ public class SplitterNoAggregationStrate
public void testSplitNoAggregationStrategy() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:split");
- mock.expectedBodiesReceived("Hello World", "Bye World");
+ mock.expectedBodiesReceived("Hello World", "Bye World", "Hi World");
MockEndpoint result = getMockEndpoint("mock:result");
- result.expectedBodiesReceived("Hello World,Bye World");
+ result.expectedBodiesReceived("Hello World,Bye World,Hi World");
- template.sendBody("direct:start", "Hello World,Bye World");
+ template.sendBody("direct:start", "Hello World,Bye World,Hi World");
assertMockEndpointsSatisfied();
}
public void testSplitNoAggregationStrategyException() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:split");
- mock.expectedBodiesReceived("Hello World", "Bye World");
+ mock.expectedBodiesReceived("Hello World", "Bye World","Hi World");
MockEndpoint result = getMockEndpoint("mock:result");
result.expectedMessageCount(0);
try {
- template.sendBody("direct:start", "Hello World,Kaboom,Bye World");
+ template.sendBody("direct:start", "Hello World,Kaboom,Bye World,Hi World");
fail("Should thrown an exception");
} catch (CamelExecutionException e) {
assertEquals("Forced", e.getCause().getMessage());
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoStopOnExceptionTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoStopOnExceptionTest.java?rev=1055913&r1=1055912&r2=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoStopOnExceptionTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterNoStopOnExceptionTest.java Thu Jan 6 15:30:53 2011
@@ -30,20 +30,20 @@ public class SplitterNoStopOnExceptionTe
public void testSplitNoStopOnExceptionOk() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:split");
- mock.expectedBodiesReceived("Hello World", "Bye World");
+ mock.expectedBodiesReceived("Hello World", "Bye World", "Hi World");
- template.sendBody("direct:start", "Hello World,Bye World");
+ template.sendBody("direct:start", "Hello World,Bye World,Hi World");
assertMockEndpointsSatisfied();
}
public void testSplitNoStopOnExceptionStop() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:split");
- // we do not stop so we split to completion and thus we receive 2 messages
- mock.expectedBodiesReceived("Hello World", "Bye World");
+ // we do not stop so we split to completion and thus we receive 3 messages
+ mock.expectedBodiesReceived("Hello World", "Bye World", "Hi World");
try {
- template.sendBody("direct:start", "Hello World,Kaboom,Bye World");
+ template.sendBody("direct:start", "Hello World,Kaboom,Bye World,Hi World");
fail("Should thrown an exception");
} catch (CamelExecutionException e) {
assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelNoStopOnExceptionTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelNoStopOnExceptionTest.java?rev=1055913&r1=1055912&r2=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelNoStopOnExceptionTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelNoStopOnExceptionTest.java Thu Jan 6 15:30:53 2011
@@ -33,9 +33,9 @@ public class SplitterParallelNoStopOnExc
public void testSplitParallelNoStopOnExceptionOk() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:split");
- mock.expectedBodiesReceivedInAnyOrder("Hello World", "Bye World");
+ mock.expectedBodiesReceivedInAnyOrder("Hello World", "Bye World", "Hi World");
- template.sendBody("direct:start", "Hello World,Bye World");
+ template.sendBody("direct:start", "Hello World,Bye World,Hi World");
assertMockEndpointsSatisfied();
}
@@ -45,10 +45,10 @@ public class SplitterParallelNoStopOnExc
mock.expectedMinimumMessageCount(0);
// we do NOT stop so we receive all messages except the one that goes kaboom
mock.allMessages().body().isNotEqualTo("Kaboom");
- mock.expectedBodiesReceivedInAnyOrder("Hello World", "Goodday World", "Bye World");
+ mock.expectedBodiesReceivedInAnyOrder("Hello World", "Goodday World", "Bye World", "Hi World");
try {
- template.sendBody("direct:start", "Hello World,Goodday World,Kaboom,Bye World");
+ template.sendBody("direct:start", "Hello World,Goodday World,Kaboom,Bye World,Hi World");
fail("Should thrown an exception");
} catch (CamelExecutionException e) {
IllegalArgumentException cause = assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingErrorHandlingTest.java (from r1055800, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingUoWIssueTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingErrorHandlingTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingErrorHandlingTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingUoWIssueTest.java&r1=1055800&r2=1055913&rev=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingUoWIssueTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingErrorHandlingTest.java Thu Jan 6 15:30:53 2011
@@ -16,6 +16,7 @@
*/
package org.apache.camel.processor;
+import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -24,29 +25,33 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class SplitterStreamingUoWIssueTest extends ContextTestSupport {
+public class SplitterStreamingErrorHandlingTest extends ContextTestSupport {
- @Override
- protected void setUp() throws Exception {
- deleteDirectory("target/splitter");
- super.setUp();
- }
-
- public void testSplitterStreamingUoWIssue() throws Exception {
- getMockEndpoint("mock:foo").expectedBodiesReceived("A", "B", "C", "D", "E");
+ public void testSplitterStreamingNoError() throws Exception {
+ getMockEndpoint("mock:a").expectedBodiesReceived("A", "B", "C", "D", "E");
+ getMockEndpoint("mock:b").expectedBodiesReceived("A", "B", "C", "D", "E");
getMockEndpoint("mock:result").expectedBodiesReceived("A,B,C,D,E");
- template.sendBodyAndHeader("file:target/splitter", "A,B,C,D,E", Exchange.FILE_NAME, "splitme.txt");
+ template.sendBody("direct:start", "A,B,C,D,E");
assertMockEndpointsSatisfied();
}
- public void testSplitterTwoFilesStreamingUoWIssue() throws Exception {
- getMockEndpoint("mock:foo").expectedBodiesReceived("A", "B", "C", "D", "E", "F", "G", "H", "I");
- getMockEndpoint("mock:result").expectedBodiesReceived("A,B,C,D,E", "F,G,H,I");
-
- template.sendBodyAndHeader("file:target/splitter", "A,B,C,D,E", Exchange.FILE_NAME, "a.txt");
- template.sendBodyAndHeader("file:target/splitter", "F,G,H,I", Exchange.FILE_NAME, "b.txt");
+ public void testSplitterStreamingWithError() throws Exception {
+ getMockEndpoint("mock:a").expectedBodiesReceived("A", "B", "Kaboom", "D", "E");
+ getMockEndpoint("mock:b").expectedBodiesReceived("A", "B", "D", "E");
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ // we do not stop on exception and thus the splitted message which failed
+ // would be silently ignored so we can continue routing
+ // you can always use a custom aggregation strategy to deal with errors your-self
+ try {
+ template.sendBody("direct:start", "A,B,Kaboom,D,E");
+ fail("Should have thrown an exception");
+ } catch (CamelExecutionException e) {
+ assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
+ assertEquals("Cannot do this", e.getCause().getMessage());
+ }
assertMockEndpointsSatisfied();
}
@@ -56,16 +61,20 @@ public class SplitterStreamingUoWIssueTe
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("file:target/splitter?delete=true&sortBy=file:name")
+ from("direct:start")
.split(body().tokenize(",")).streaming()
- .to("seda:queue")
+ .to("mock:a")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ String body = exchange.getIn().getBody(String.class);
+ if ("Kaboom".equals(body)) {
+ throw new IllegalArgumentException("Cannot do this");
+ }
+ }
+ })
+ .to("mock:b")
.end()
- .log("End of file ${file:name}")
.to("mock:result");
-
- from("seda:queue")
- .log("Token: ${body}")
- .to("mock:foo");
}
};
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionErrorHandlingTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionErrorHandlingTest.java?rev=1055913&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionErrorHandlingTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionErrorHandlingTest.java Thu Jan 6 15:30:53 2011
@@ -0,0 +1,78 @@
+/**
+ * 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.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version $Revision$
+ */
+public class SplitterStreamingStopOnExceptionErrorHandlingTest extends ContextTestSupport {
+
+ public void testSplitterStreamingNoError() throws Exception {
+ getMockEndpoint("mock:a").expectedBodiesReceived("A", "B", "C", "D", "E");
+ getMockEndpoint("mock:b").expectedBodiesReceived("A", "B", "C", "D", "E");
+ getMockEndpoint("mock:result").expectedBodiesReceived("A,B,C,D,E");
+
+ template.sendBody("direct:start", "A,B,C,D,E");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testSplitterStreamingWithError() throws Exception {
+ getMockEndpoint("mock:a").expectedBodiesReceived("A", "B", "Kaboom");
+ getMockEndpoint("mock:b").expectedBodiesReceived("A", "B");
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ try {
+ template.sendBody("direct:start", "A,B,Kaboom,D,E");
+ fail("Should have thrown an exception");
+ } catch (CamelExecutionException e) {
+ assertIsInstanceOf(IllegalArgumentException.class, e.getCause().getCause());
+ assertEquals("Cannot do this", e.getCause().getCause().getMessage());
+ }
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start")
+ .split(body().tokenize(",")).streaming().stopOnException()
+ .to("mock:a")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ String body = exchange.getIn().getBody(String.class);
+ if ("Kaboom".equals(body)) {
+ throw new IllegalArgumentException("Cannot do this");
+ }
+ }
+ })
+ .to("mock:b")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+}
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitFineGrainedErrorHandlingTest.java (from r1055800, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitFineGrainedErrorHandlingTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitFineGrainedErrorHandlingTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java&r1=1055800&r2=1055913&rev=1055913&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitFineGrainedErrorHandlingTest.java Thu Jan 6 15:30:53 2011
@@ -24,20 +24,21 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointSplitTest extends ContextTestSupport {
+public class AsyncEndpointSplitFineGrainedErrorHandlingTest extends ContextTestSupport {
- private static String beforeThreadName;
- private static String afterThreadName;
+ private static int counter;
public void testAsyncEndpoint() throws Exception {
- getMockEndpoint("mock:before").expectedBodiesReceived("A", "B");
- getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel", "Bye Camel");
- getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:before").expectedBodiesReceived("A", "B", "C");
+ getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel", "Bye Camel", "Bye Camel");
+ getMockEndpoint("mock:result").expectedBodiesReceived("A,B,C");
- String reply = template.requestBody("direct:start", "A,B", String.class);
- assertEquals("A,B", reply);
+ template.sendBody("direct:start", "A,B,C");
- assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
+ assertMockEndpointsSatisfied();
+
+ // 3 split messages + 1 redelivery attempt
+ assertEquals(3 + 1, counter);
}
@Override
@@ -47,22 +48,19 @@ public class AsyncEndpointSplitTest exte
public void configure() throws Exception {
context.addComponent("async", new MyAsyncComponent());
+ onException(Exception.class).maximumRedeliveries(2).redeliveryDelay(0);
+
from("direct:start")
.split(body())
.to("mock:before")
- .to("log:before")
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- beforeThreadName = Thread.currentThread().getName();
- }
- })
.to("async:Bye Camel")
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
- afterThreadName = Thread.currentThread().getName();
+ if (counter++ == 1) {
+ throw new IllegalArgumentException("Cannot do this");
+ }
}
})
- .to("log:after")
.to("mock:after")
.end()
.to("mock:result");