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 2010/06/23 12:53:22 UTC
svn commit: r957153 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/ main/java/org/apache/camel/impl/
main/java/org/apache/camel/model/ main/java/org/apache/camel/processor/
main/java/org/apache/camel/processor/aggregate/ main/java/org/apa...
Author: davsclaus
Date: Wed Jun 23 10:53:21 2010
New Revision: 957153
URL: http://svn.apache.org/viewvc?rev=957153&view=rev
Log:
CAMEL-2338: Multicast, RecipientList and Splitter EIP supports async routing engine.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastTest.java
- copied, changed from r957101, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java (contents, props changed)
- copied, changed from r957101, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/Exchange.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/AsyncProcessorHelper.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/Exchange.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/Exchange.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/Exchange.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/Exchange.java Wed Jun 23 10:53:21 2010
@@ -37,6 +37,7 @@ public interface Exchange {
String AGGREGATED_SIZE = "CamelAggregatedSize";
String AGGREGATED_COMPLETED_BY = "CamelAggregatedCompletedBy";
String AGGREGATED_CORRELATION_KEY = "CamelAggregatedCorrelationKey";
+ String AGGREGATION_STRATEGY = "CamelAggregationStrategy";
String ASYNC_WAIT = "CamelAsyncWait";
String BATCH_INDEX = "CamelBatchIndex";
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java Wed Jun 23 10:53:21 2010
@@ -109,7 +109,7 @@ public final class DefaultExchange imple
}
public Object getProperty(String name) {
- if (properties != null) {
+ if (hasProperties()) {
return properties.get(name);
}
return null;
@@ -117,7 +117,7 @@ public final class DefaultExchange imple
public Object getProperty(String name, Object defaultValue) {
Object answer = null;
- if (properties != null) {
+ if (hasProperties()) {
answer = properties.get(name);
}
return answer != null ? answer : defaultValue;
@@ -160,6 +160,9 @@ public final class DefaultExchange imple
}
public Object removeProperty(String name) {
+ if (!hasProperties()) {
+ return null;
+ }
return getProperties().remove(name);
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java Wed Jun 23 10:53:21 2010
@@ -29,7 +29,6 @@ import org.apache.camel.builder.Expressi
import org.apache.camel.model.language.ExpressionDefinition;
import org.apache.camel.processor.Splitter;
import org.apache.camel.processor.aggregate.AggregationStrategy;
-import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
import org.apache.camel.spi.RouteContext;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.concurrent.ExecutorServiceHelper;
@@ -106,10 +105,6 @@ public class SplitDefinition extends Exp
if (strategy == null && strategyRef != null) {
strategy = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), strategyRef, AggregationStrategy.class);
}
- if (strategy == null) {
- // fallback to keep the original exchange strategy
- strategy = new UseOriginalAggregationStrategy(true);
- }
return strategy;
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java Wed Jun 23 10:53:21 2010
@@ -197,6 +197,9 @@ public class MulticastProcessor extends
exchange.setException(e);
}
+ // cleanup any per exchange aggregation strategy
+ exchange.removeProperty(Exchange.AGGREGATION_STRATEGY);
+
callback.done(true);
return true;
}
@@ -229,7 +232,7 @@ public class MulticastProcessor extends
return subExchange;
}
- doProcess(original, result, it, pair, callback);
+ doProcess(original, result, it, pair, callback, total);
// should we stop in case of an exception occurred during processing?
if (stopOnException && subExchange.getException() != null) {
@@ -251,9 +254,7 @@ public class MulticastProcessor extends
for (int i = 0; i < total.intValue(); i++) {
Future<Exchange> future = completion.take();
Exchange subExchange = future.get();
- if (aggregationStrategy != null) {
- doAggregate(result, subExchange);
- }
+ doAggregate(getAggregationStrategy(subExchange), result, subExchange);
}
if (LOG.isDebugEnabled()) {
@@ -262,15 +263,15 @@ public class MulticastProcessor extends
}
protected boolean doProcessSequential(Exchange original, AtomicExchange result, Iterable<ProcessorExchangePair> pairs, AsyncCallback callback) throws Exception {
- int total = 0;
+ AtomicInteger total = new AtomicInteger();
Iterator<ProcessorExchangePair> it = pairs.iterator();
while (it.hasNext()) {
ProcessorExchangePair pair = it.next();
Exchange subExchange = pair.getExchange();
- updateNewExchange(subExchange, total, pairs);
+ updateNewExchange(subExchange, total.get(), pairs);
- boolean sync = doProcess(original, result, it, pair, callback);
+ boolean sync = doProcess(original, result, it, pair, callback, total);
if (!sync) {
if (LOG.isTraceEnabled()) {
LOG.trace("Processing exchangeId: " + pair.getExchange().getExchangeId() + " is continued being processed asynchronously");
@@ -286,17 +287,15 @@ public class MulticastProcessor extends
// should we stop in case of an exception occurred during processing?
if (stopOnException && subExchange.getException() != null) {
- throw new CamelExchangeException("Sequential processing failed for number " + total, subExchange, subExchange.getException());
+ throw new CamelExchangeException("Sequential processing failed for number " + total.get(), subExchange, subExchange.getException());
}
if (LOG.isTraceEnabled()) {
LOG.trace("Sequential processing complete for number " + total + " exchange: " + subExchange);
}
- if (aggregationStrategy != null) {
- doAggregate(result, subExchange);
- }
- total++;
+ doAggregate(getAggregationStrategy(subExchange), result, subExchange);
+ total.incrementAndGet();
}
if (LOG.isDebugEnabled()) {
@@ -307,7 +306,7 @@ public class MulticastProcessor extends
}
private boolean doProcess(final Exchange original, final AtomicExchange result, final Iterator<ProcessorExchangePair> it,
- final ProcessorExchangePair pair, final AsyncCallback callback) {
+ final ProcessorExchangePair pair, final AsyncCallback callback, final AtomicInteger total) {
boolean sync = true;
final Exchange exchange = pair.getExchange();
@@ -328,11 +327,12 @@ public class MulticastProcessor extends
traced.pushBlock();
}
- // let the prepared process it
+ // let the prepared process it, remember to begin the exchange pair
AsyncProcessor async = AsyncProcessorTypeConverter.convert(processor);
pair.begin();
sync = async.process(exchange, new AsyncCallback() {
public void done(boolean doneSync) {
+ // we are done with the exchange pair
pair.done();
// we only have to handle async completion of the routing slip
@@ -340,55 +340,98 @@ public class MulticastProcessor extends
return;
}
- // TODO: total number
// continue processing the multicast asynchronously
Exchange subExchange = exchange;
- int total = 0;
+ // remember to test for stop on exception and aggregate before copying back results
+ if (stopOnException && subExchange.getException() != null) {
+ // wrap in exception to explain where it failed
+ subExchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, subExchange.getException()));
+ // multicast uses error handling on its output processors and they have tried to redeliver
+ // so we shall signal back to the other error handlers that we are exhausted and they should not
+ // also try to redeliver as we will then do that twice
+ exchange.setProperty(Exchange.REDELIVERY_EXHAUSTED, Boolean.TRUE);
+ // and copy the current result to original so it will contain this exception
+ // cleanup any per exchange aggregation strategy
+ original.removeProperty(Exchange.AGGREGATION_STRATEGY);
+ ExchangeHelper.copyResults(original, subExchange);
+ callback.done(false);
+ return;
+ }
+
+ try {
+ doAggregate(getAggregationStrategy(subExchange), result, subExchange);
+ } catch (Throwable e) {
+ // wrap in exception to explain where it failed
+ subExchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, e));
+ // multicast uses error handling on its output processors and they have tried to redeliver
+ // so we shall signal back to the other error handlers that we are exhausted and they should not
+ // also try to redeliver as we will then do that twice
+ original.setProperty(Exchange.REDELIVERY_EXHAUSTED, Boolean.TRUE);
+ // cleanup any per exchange aggregation strategy
+ original.removeProperty(Exchange.AGGREGATION_STRATEGY);
+ // and copy the current result to original so it will contain this exception
+ ExchangeHelper.copyResults(original, subExchange);
+ callback.done(false);
+ return;
+ }
+
+ total.incrementAndGet();
+
+ // maybe there are more processors to multicast
while (it.hasNext()) {
- if (stopOnException && exchange.getException() != null) {
- // wrap in exception to explain where it failed
- exchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, subExchange.getException()));
- callback.done(false);
+ // prepare and run the next
+ ProcessorExchangePair pair = it.next();
+ subExchange = pair.getExchange();
+ updateNewExchange(subExchange, total.get(), null);
+ boolean sync = doProcess(original, result, it, pair, callback, total);
+
+ if (!sync) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchangeId: " + original.getExchangeId() + " is continued being processed asynchronously");
+ }
return;
}
- if (aggregationStrategy != null) {
- doAggregate(result, subExchange);
+ if (stopOnException && subExchange.getException() != null) {
+ // wrap in exception to explain where it failed
+ subExchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, subExchange.getException()));
+ // multicast uses error handling on its output processors and they have tried to redeliver
+ // so we shall signal back to the other error handlers that we are exhausted and they should not
+ // also try to redeliver as we will then do that twice
+ original.setProperty(Exchange.REDELIVERY_EXHAUSTED, Boolean.TRUE);
+ // cleanup any per exchange aggregation strategy
+ original.removeProperty(Exchange.AGGREGATION_STRATEGY);
+ // and copy the current result to original so it will contain this exception
+ ExchangeHelper.copyResults(original, subExchange);
+ callback.done(false);
+ return;
}
- if (it.hasNext()) {
- // prepare and run the next
- ProcessorExchangePair pair = it.next();
- subExchange = pair.getExchange();
- updateNewExchange(subExchange, total, null);
- boolean sync = doProcess(original, result, it, pair, callback);
-
- if (!sync) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Processing exchangeId: " + original.getExchangeId() + " is continued being processed asynchronously");
- }
- return;
- }
-
- total++;
+ try {
+ doAggregate(getAggregationStrategy(subExchange), result, subExchange);
+ } catch (Throwable e) {
+ // wrap in exception to explain where it failed
+ subExchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, e));
+ // multicast uses error handling on its output processors and they have tried to redeliver
+ // so we shall signal back to the other error handlers that we are exhausted and they should not
+ // also try to redeliver as we will then do that twice
+ original.setProperty(Exchange.REDELIVERY_EXHAUSTED, Boolean.TRUE);
+ // cleanup any per exchange aggregation strategy
+ original.removeProperty(Exchange.AGGREGATION_STRATEGY);
+ // and copy the current result to original so it will contain this exception
+ ExchangeHelper.copyResults(original, subExchange);
+ callback.done(false);
+ return;
}
- }
-
- // remember to test for stop on exception and aggregate before copying back results
- if (stopOnException && exchange.getException() != null) {
- // wrap in exception to explain where it failed
- exchange.setException(new CamelExchangeException("Sequential processing failed for number " + total, subExchange, subExchange.getException()));
- callback.done(false);
- return;
- }
- if (aggregationStrategy != null) {
- doAggregate(result, subExchange);
+ total.incrementAndGet();
}
- // copy results back to the original exchange
+ // cleanup any per exchange aggregation strategy
+ original.removeProperty(Exchange.AGGREGATION_STRATEGY);
+ // multicasting complete so copy results back to the original exchange
if (result.get() != null) {
ExchangeHelper.copyResults(original, result.get());
}
@@ -414,15 +457,16 @@ public class MulticastProcessor extends
/**
* Aggregate the {@link Exchange} with the current result
*
+ * @param strategy the aggregation strategy to use
* @param result the current result
* @param exchange the exchange to be added to the result
*/
- protected synchronized void doAggregate(AtomicExchange result, Exchange exchange) {
- if (aggregationStrategy != null) {
+ protected synchronized void doAggregate(AggregationStrategy strategy, AtomicExchange result, Exchange exchange) {
+ if (strategy != null) {
// prepare the exchanges for aggregation
Exchange oldExchange = result.get();
ExchangeHelper.prepareAggregation(oldExchange, exchange);
- result.set(aggregationStrategy.aggregate(oldExchange, exchange));
+ result.set(strategy.aggregate(oldExchange, exchange));
}
}
@@ -496,6 +540,16 @@ public class MulticastProcessor extends
}
}
+ protected AggregationStrategy getAggregationStrategy(Exchange exhange) {
+ // prefer to use per Exchange aggregation strategy over a global strategy
+ AggregationStrategy answer = exhange.getProperty(Exchange.AGGREGATION_STRATEGY, AggregationStrategy.class);
+ if (answer == null) {
+ // fallback to global strategy
+ answer = getAggregationStrategy();
+ }
+ return answer;
+ }
+
/**
* Is the multicast processor working in streaming mode?
*
@@ -524,6 +578,9 @@ public class MulticastProcessor extends
return processors;
}
+ /**
+ * Use {@link #getAggregationStrategy(org.apache.camel.Exchange)} instead.
+ */
public AggregationStrategy getAggregationStrategy() {
return aggregationStrategy;
}
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=957153&r1=957152&r2=957153&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 Wed Jun 23 10:53:21 2010
@@ -75,23 +75,15 @@ public class Splitter extends MulticastP
public boolean process(Exchange exchange, final AsyncCallback callback) {
final AggregationStrategy strategy = getAggregationStrategy();
- // if original aggregation strategy then store exchange
- // on it as the original exchange
- UseOriginalAggregationStrategy original = null;
- if (strategy instanceof UseOriginalAggregationStrategy) {
- original = (UseOriginalAggregationStrategy) strategy;
- original.setOriginal(exchange);
+ // if no custom aggregation strategy is being used then fallback to keep the original
+ // and propagate exceptions which is done by a per exchange specific aggregation strategy
+ // to ensure it supports async routing
+ if (strategy == null) {
+ UseOriginalAggregationStrategy original = new UseOriginalAggregationStrategy(exchange, true);
+ exchange.setProperty(Exchange.AGGREGATION_STRATEGY, original);
}
- // TODO: we will lose the original in the async routing engine when it return false
- try {
- return super.process(exchange, callback);
- } finally {
- if (original != null) {
- // and remove the reference when we are done (due to thread local stuff)
- original.setOriginal(null);
- }
- }
+ return super.process(exchange, callback);
}
@Override
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java Wed Jun 23 10:53:21 2010
@@ -17,50 +17,35 @@
package org.apache.camel.processor.aggregate;
import org.apache.camel.Exchange;
+import org.apache.camel.util.ObjectHelper;
/**
* An {@link org.apache.camel.processor.aggregate.AggregationStrategy} which just uses the original exchange
- * which can be needed when you want to preserve the original Exchange. For example when splitting an Exchange
- * and then you may want to keep routing using the original Exchange.
- * <p/>
- * You must call {@link #setOriginal(org.apache.camel.Exchange)} before this aggregation strategy can be used,
- * as it needs to have a reference to the original exchange.
+ * which can be needed when you want to preserve the original Exchange. For example when splitting an {@link Exchange}
+ * and then you may want to keep routing using the original {@link Exchange}.
*
* @see org.apache.camel.processor.Splitter
* @version $Revision$
*/
public class UseOriginalAggregationStrategy implements AggregationStrategy {
- // must use a thread local to cater for concurrency
- private final ThreadLocal<Exchange> original = new ThreadLocal<Exchange>();
+ private final Exchange original;
private final boolean propagateException;
- public UseOriginalAggregationStrategy(boolean propagateException) {
+ public UseOriginalAggregationStrategy(Exchange original, boolean propagateException) {
+ ObjectHelper.notNull(original, "Original Exchange");
+ this.original = original;
this.propagateException = propagateException;
}
- public void setOriginal(Exchange exchange) {
- if (exchange == null) {
- // clear it
- this.original.remove();
- } else {
- this.original.set(exchange);
- }
- }
-
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- Exchange answer = original.get();
- if (answer == null) {
- throw new IllegalStateException("Original Exchange has not been set");
- }
-
if (propagateException) {
Exception exception = checkException(oldExchange, newExchange);
if (exception != null) {
- answer.setException(exception);
+ original.setException(exception);
}
}
- return answer;
+ return original;
}
protected Exception checkException(Exchange oldExchange, Exchange newExchange) {
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/util/AsyncProcessorHelper.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/AsyncProcessorHelper.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/util/AsyncProcessorHelper.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/util/AsyncProcessorHelper.java Wed Jun 23 10:53:21 2010
@@ -43,14 +43,18 @@ public final class AsyncProcessorHelper
* Calls the async version of the processor's process method and waits
* for it to complete before returning. This can be used by {@link AsyncProcessor}
* objects to implement their sync version of the process method.
+ *
+ * @param processor the processor
+ * @param exchange the exchange
+ * @throws Exception can be thrown if waiting is interrupted
*/
public static void process(final AsyncProcessor processor, final Exchange exchange) throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
boolean sync = processor.process(exchange, new AsyncCallback() {
public void done(boolean doneSync) {
if (!doneSync) {
- if (LOG.isWarnEnabled()) {
- LOG.warn("Asynchronous callback received for exchangeId: " + exchange.getExchangeId());
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Asynchronous callback received for exchangeId: " + exchange.getExchangeId());
}
latch.countDown();
}
@@ -62,10 +66,13 @@ public final class AsyncProcessorHelper
}
});
if (!sync) {
- if (LOG.isWarnEnabled()) {
- LOG.warn("Waiting for asynchronous callback before continuing for exchangeId: " + exchange.getExchangeId());
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Waiting for asynchronous callback before continuing for exchangeId: " + exchange.getExchangeId() + " -> " + exchange);
}
latch.await();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Asynchronous callback received, will continue routing exchangeId: " + exchange.getExchangeId() + " -> " + exchange);
+ }
}
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java?rev=957153&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java Wed Jun 23 10:53:21 2010
@@ -0,0 +1,87 @@
+/**
+ * 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.async;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version $Revision$
+ */
+public class AsyncEndpointMulticastFineGrainedErrorHandlingTest extends ContextTestSupport {
+
+ public void testAsyncEndpointOK() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ onException(Exception.class).maximumRedeliveries(2);
+
+ from("direct:start")
+ .to("mock:a")
+ .multicast().stopOnException()
+ .to("mock:foo", "async:Bye Camel", "mock:bar");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:a").expectedMessageCount(1);
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testAsyncEndpointERROR() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ onException(Exception.class).maximumRedeliveries(2);
+
+ from("direct:start")
+ .to("mock:a")
+ .multicast().stopOnException()
+ .to("mock:foo", "async:Bye Camel").throwException(new IllegalArgumentException("Damn")).to("mock:bar");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:a").expectedMessageCount(1);
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(0);
+
+ try {
+ template.sendBody("direct:start", "Hello World");
+ fail("Should throw exception");
+ } catch (Exception e) {
+ // expected
+ }
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastFineGrainedErrorHandlingTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastTest.java (from r957101, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java&r1=957101&r2=957153&rev=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointMulticastTest.java Wed Jun 23 10:53:21 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientListTest extends ContextTestSupport {
+public class AsyncEndpointMulticastTest extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -55,7 +55,9 @@ public class AsyncEndpointRecipientListT
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("async:Bye Camel"))
+ .multicast()
+ .to("async:Hi Moon", "direct:foo")
+ .end()
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
afterThreadName = Thread.currentThread().getName();
@@ -64,6 +66,8 @@ public class AsyncEndpointRecipientListT
.to("log:after")
.to("mock:after")
.to("mock:result");
+
+ from("direct:foo").setBody(constant("Bye Camel"));
}
};
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java?rev=957153&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java Wed Jun 23 10:53:21 2010
@@ -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.processor.async;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.JndiRegistry;
+
+/**
+ * @version $Revision$
+ */
+public class AsyncEndpointRecipientListFineGrainedErrorHandlingTest extends ContextTestSupport {
+
+ private static int counter;
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry jndi = super.createRegistry();
+ jndi.bind("fail", new MyFailBean());
+ return jndi;
+ }
+
+ public void testAsyncEndpointOK() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ onException(Exception.class).maximumRedeliveries(2);
+
+ from("direct:start")
+ .to("mock:a")
+ .recipientList(header("foo")).stopOnException();
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:a").expectedMessageCount(1);
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ getMockEndpoint("mock:baz").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:start", "Hello World", "foo", "mock:foo,async:Bye Camel,mock:bar,mock:baz");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testAsyncEndpointERROR() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ onException(Exception.class).maximumRedeliveries(2);
+
+ from("direct:start")
+ .to("mock:a")
+ .recipientList(header("foo")).stopOnException();
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:a").expectedMessageCount(1);
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ getMockEndpoint("mock:baz").expectedMessageCount(0);
+
+ try {
+ template.sendBodyAndHeader("direct:start", "Hello World", "foo", "mock:foo,mock:bar,bean:fail,mock:baz");
+ fail("Should throw exception");
+ } catch (Exception e) {
+ // expected
+ }
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(3, counter);
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ public static class MyFailBean {
+
+ public String doSomething(Exchange exchange) throws Exception {
+ counter++;
+ assertEquals("bean://fail", exchange.getProperty(Exchange.TO_ENDPOINT, String.class));
+ throw new IllegalArgumentException("Damn");
+ }
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListFineGrainedErrorHandlingTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: 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/AsyncEndpointSplitTest.java?rev=957153&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java Wed Jun 23 10:53:21 2010
@@ -0,0 +1,73 @@
+/**
+ * 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.async;
+
+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 AsyncEndpointSplitTest extends ContextTestSupport {
+
+ private static String beforeThreadName;
+ private static String afterThreadName;
+
+ 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");
+
+ String reply = template.requestBody("direct:start", "A,B", String.class);
+ assertEquals("A,B", reply);
+
+ assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ 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();
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java (from r957101, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java&r1=957101&r2=957153&rev=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java Wed Jun 23 10:53:21 2010
@@ -20,21 +20,22 @@ import org.apache.camel.ContextTestSuppo
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientListTest extends ContextTestSupport {
+public class AsyncEndpointSplitUseLatestAggregationStrategyTest extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
public void testAsyncEndpoint() throws Exception {
- getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
- getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:before").expectedBodiesReceived("A", "B");
+ getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel", "Bye Camel");
getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
- String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ String reply = template.requestBody("direct:start", "A,B", String.class);
assertEquals("Bye Camel", reply);
assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
@@ -48,6 +49,7 @@ public class AsyncEndpointRecipientListT
context.addComponent("async", new MyAsyncComponent());
from("direct:start")
+ .split(body(), new UseLatestAggregationStrategy())
.to("mock:before")
.to("log:before")
.process(new Processor() {
@@ -55,7 +57,7 @@ public class AsyncEndpointRecipientListT
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("async:Bye Camel"))
+ .to("async:Bye Camel")
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
afterThreadName = Thread.currentThread().getName();
@@ -63,7 +65,8 @@ public class AsyncEndpointRecipientListT
})
.to("log:after")
.to("mock:after")
- .to("mock:result");
+ .end()
+ .to("mock:result");
}
};
}
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointSplitUseLatestAggregationStrategyTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java?rev=957153&r1=957152&r2=957153&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java Wed Jun 23 10:53:21 2010
@@ -34,7 +34,7 @@ import org.apache.commons.logging.LogFac
public class MyAsyncProducer extends DefaultAsyncProducer {
private static final Log LOG = LogFactory.getLog(MyAsyncProducer.class);
- private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final ExecutorService executor = Executors.newCachedThreadPool();
private final AtomicInteger counter = new AtomicInteger();
public MyAsyncProducer(MyAsyncEndpoint endpoint) {