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/07/05 11:52:25 UTC
svn commit: r960507 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/ main/java/org/apache/camel/processor/
test/java/org/apache/camel/processor/async/
Author: davsclaus
Date: Mon Jul 5 09:52:25 2010
New Revision: 960507
URL: http://svn.apache.org/viewvc?rev=960507&view=rev
Log:
CAMEL-2849: multicast, recipinent list DSL is compatible with async routing engine.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel2Test.java
- copied, changed from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java (contents, props changed)
- copied, changed from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel4Test.java
- copied, changed from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList4Test.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallelTest.java
- copied, changed from r960448, 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/AsyncProducerCallback.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/RecipientList.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/AsyncProducerCallback.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/AsyncProducerCallback.java?rev=960507&r1=960506&r2=960507&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/AsyncProducerCallback.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/AsyncProducerCallback.java Mon Jul 5 09:52:25 2010
@@ -34,7 +34,7 @@ public interface AsyncProducerCallback {
* @param exchange the exchange, can be <tt>null</tt> if so then create a new exchange from the producer
* @param exchangePattern the exchange pattern, can be <tt>null</tt>
* @param callback the async callback
- * @return the response
+ * @return (doneSync) <tt>true</tt> to continue execute synchronously, <tt>false</tt> to continue being executed asynchronously
*/
boolean doInAsyncProducer(Producer producer, AsyncProcessor asyncProducer, Exchange exchange,
ExchangePattern exchangePattern, AsyncCallback callback);
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=960507&r1=960506&r2=960507&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 Mon Jul 5 09:52:25 2010
@@ -199,7 +199,7 @@ public class MulticastProcessor extends
}
protected void doProcessParallel(final Exchange original, final AtomicExchange result, final Iterable<ProcessorExchangePair> pairs,
- boolean streaming, final AsyncCallback callback) throws InterruptedException, ExecutionException {
+ final boolean streaming, final AsyncCallback callback) throws InterruptedException, ExecutionException {
final CompletionService<Exchange> completion;
final AtomicBoolean running = new AtomicBoolean(true);
@@ -226,7 +226,11 @@ public class MulticastProcessor extends
return subExchange;
}
- doProcess(original, result, pairs, it, pair, callback, total);
+ try {
+ doProcessParallel(pair);
+ } catch (Exception e) {
+ subExchange.setException(e);
+ }
// should we stop in case of an exception occurred during processing?
if (stopOnException && subExchange.getException() != null) {
@@ -245,6 +249,8 @@ public class MulticastProcessor extends
total.incrementAndGet();
}
+ // its to hard to do parallel async routing so we let the caller thread be synchronously
+ // and have it pickup the replies and do the aggregation
for (int i = 0; i < total.intValue(); i++) {
Future<Exchange> future = completion.take();
Exchange subExchange = future.get();
@@ -265,7 +271,7 @@ public class MulticastProcessor extends
Exchange subExchange = pair.getExchange();
updateNewExchange(subExchange, total.get(), pairs, it);
- boolean sync = doProcess(original, result, pairs, it, pair, callback, total);
+ boolean sync = doProcessSequential(original, result, pairs, it, pair, callback, total);
if (!sync) {
if (LOG.isTraceEnabled()) {
LOG.trace("Processing exchangeId: " + pair.getExchange().getExchangeId() + " is continued being processed asynchronously");
@@ -299,9 +305,9 @@ public class MulticastProcessor extends
return true;
}
- private boolean doProcess(final Exchange original, final AtomicExchange result,
- final Iterable<ProcessorExchangePair> pairs, final Iterator<ProcessorExchangePair> it,
- final ProcessorExchangePair pair, final AsyncCallback callback, final AtomicInteger total) {
+ private boolean doProcessSequential(final Exchange original, final AtomicExchange result,
+ final Iterable<ProcessorExchangePair> pairs, final Iterator<ProcessorExchangePair> it,
+ final ProcessorExchangePair pair, final AsyncCallback callback, final AtomicInteger total) {
boolean sync = true;
final Exchange exchange = pair.getExchange();
@@ -366,7 +372,7 @@ public class MulticastProcessor extends
ProcessorExchangePair pair = it.next();
subExchange = pair.getExchange();
updateNewExchange(subExchange, total.get(), pairs, it);
- boolean sync = doProcess(original, result, pairs, it, pair, callback, total);
+ boolean sync = doProcessSequential(original, result, pairs, it, pair, callback, total);
if (!sync) {
if (LOG.isTraceEnabled()) {
@@ -417,6 +423,45 @@ public class MulticastProcessor extends
return sync;
}
+ private void doProcessParallel(final ProcessorExchangePair pair) throws Exception {
+ final Exchange exchange = pair.getExchange();
+ Processor processor = pair.getProcessor();
+ Producer producer = pair.getProducer();
+
+ TracedRouteNodes traced = exchange.getUnitOfWork() != null ? exchange.getUnitOfWork().getTracedRouteNodes() : null;
+
+ // compute time taken if sending to another endpoint
+ StopWatch watch = null;
+ if (producer != null) {
+ watch = new StopWatch();
+ }
+
+ try {
+ // prepare tracing starting from a new block
+ if (traced != null) {
+ traced.pushBlock();
+ }
+
+ // let the prepared process it, remember to begin the exchange pair
+ // we invoke it synchronously as parallel async routing is too hard
+ AsyncProcessor async = AsyncProcessorTypeConverter.convert(processor);
+ pair.begin();
+ AsyncProcessorHelper.process(async, exchange);
+ } finally {
+ pair.done();
+ // pop the block so by next round we have the same staring point and thus the tracing looks accurate
+ if (traced != null) {
+ traced.popBlock();
+ }
+ if (producer != null) {
+ long timeTaken = watch.stop();
+ Endpoint endpoint = producer.getEndpoint();
+ // emit event that the exchange was sent to the endpoint
+ EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint, timeTaken);
+ }
+ }
+ }
+
/**
* Common work which must be done when we are done multicasting.
* <p/>
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java?rev=960507&r1=960506&r2=960507&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java Mon Jul 5 09:52:25 2010
@@ -33,8 +33,6 @@ import org.apache.camel.util.AsyncProces
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ServiceHelper;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
import static org.apache.camel.util.ObjectHelper.notNull;
@@ -47,7 +45,6 @@ import static org.apache.camel.util.Obje
* @version $Revision$
*/
public class RecipientList extends ServiceSupport implements AsyncProcessor {
- private static final transient Log LOG = LogFactory.getLog(RecipientList.class);
private final CamelContext camelContext;
private ProducerCache producerCache;
private Expression expression;
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel2Test.java (from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel2Test.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel2Test.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java&r1=960448&r2=960507&rev=960507&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel2Test.java Mon Jul 5 09:52:25 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientList2Test extends ContextTestSupport {
+public class AsyncEndpointRecipientListParallel2Test extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -57,7 +57,7 @@ public class AsyncEndpointRecipientList2
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("direct:foo"));
+ .recipientList(constant("direct:foo")).parallelProcessing();
from("direct:foo")
.to("async:Bye Camel")
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java (from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java&r1=960448&r2=960507&rev=960507&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java Mon Jul 5 09:52:25 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientList3Test extends ContextTestSupport {
+public class AsyncEndpointRecipientListParallel3Test extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -57,7 +57,7 @@ public class AsyncEndpointRecipientList3
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("async:Hi Camel,direct:foo"));
+ .recipientList(constant("async:Hi Camel,direct:foo")).parallelProcessing();
from("direct:foo")
.process(new Processor() {
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel3Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel4Test.java (from r960448, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList4Test.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel4Test.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel4Test.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList4Test.java&r1=960448&r2=960507&rev=960507&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList4Test.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel4Test.java Mon Jul 5 09:52:25 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientList4Test extends ContextTestSupport {
+public class AsyncEndpointRecipientListParallel4Test extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -57,7 +57,7 @@ public class AsyncEndpointRecipientList4
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("async:Hi Camel,async:Hi World,direct:foo"));
+ .recipientList(constant("async:Hi Camel,async:Hi World,direct:foo")).parallelProcessing();
from("direct:foo")
.process(new Processor() {
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java?rev=960507&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java Mon Jul 5 09:52:25 2010
@@ -0,0 +1,79 @@
+/**
+ * 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 AsyncEndpointRecipientListParallel5Test 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:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+
+ // to hard to do parallel async routing so the caller thread is synchronized
+ assertTrue("Should use same 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")
+ .to("mock:before")
+ .to("log:before")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ beforeThreadName = Thread.currentThread().getName();
+ }
+ })
+ .recipientList(constant("async:Hi Camel,async:Hi World,direct:foo")).parallelProcessing()
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ afterThreadName = Thread.currentThread().getName();
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .to("mock:result");
+
+
+ from("direct:foo")
+ .transform(constant("Bye Camel"))
+ .to("mock:foo");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallel5Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallelTest.java (from r960448, 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/AsyncEndpointRecipientListParallelTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListParallelTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java&r1=960448&r2=960507&rev=960507&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/AsyncEndpointRecipientListParallelTest.java Mon Jul 5 09:52:25 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointRecipientListTest extends ContextTestSupport {
+public class AsyncEndpointRecipientListParallelTest extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -37,7 +37,8 @@ public class AsyncEndpointRecipientListT
String reply = template.requestBody("direct:start", "Hello Camel", String.class);
assertEquals("Bye Camel", reply);
- assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
+ // to hard to do parallel async routing so the caller thread is synchronized
+ assertTrue("Should use same threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
}
@Override
@@ -55,7 +56,7 @@ public class AsyncEndpointRecipientListT
beforeThreadName = Thread.currentThread().getName();
}
})
- .recipientList(constant("async:Bye Camel"))
+ .recipientList(constant("async:Bye Camel")).parallelProcessing()
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
afterThreadName = Thread.currentThread().getName();