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/21 16:30:13 UTC
svn commit: r956598 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/processor/ main/java/org/apache/camel/util/
test/java/org/apache/camel/processor/async/
test/java/org/apache/camel/processor/resequencer/
test/java/org/apache/camel/proces...
Author: davsclaus
Date: Mon Jun 21 14:30:12 2010
New Revision: 956598
URL: http://svn.apache.org/viewvc?rev=956598&view=rev
Log:
CAMEL-2835: Routing slip EIP now fully supports async routing engine.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java (contents, props changed)
- copied, changed from r956466, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java (with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java (with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.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/resequencer/MyFileNameExpression.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipIgnoreInvalidEndpointsTest.java
camel/trunk/camel-core/src/test/resources/log4j.properties
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java Mon Jun 21 14:30:12 2010
@@ -129,7 +129,6 @@ public class Pipeline extends MulticastP
callback.done(true);
return true;
-
}
private boolean process(final Exchange original, final Exchange exchange, final AsyncCallback callback,
@@ -189,11 +188,13 @@ public class Pipeline extends MulticastP
}
ExchangeHelper.copyResults(original, nextExchange);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing complete for exchangeId: " + original.getExchangeId() + " >>> " + original);
+ }
callback.done(false);
}
});
-
return sync;
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RoutingSlip.java Mon Jun 21 14:30:12 2010
@@ -18,6 +18,9 @@ package org.apache.camel.processor;
import java.util.Iterator;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.AsyncProducerCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -25,14 +28,13 @@ import org.apache.camel.ExchangePattern;
import org.apache.camel.Expression;
import org.apache.camel.FailedToCreateProducerException;
import org.apache.camel.Message;
-import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.ProducerCallback;
import org.apache.camel.builder.ExpressionBuilder;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.impl.ProducerCache;
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.model.RoutingSlipDefinition;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ServiceHelper;
@@ -46,7 +48,7 @@ import static org.apache.camel.util.Obje
* pattern where the list of actual endpoints to send a message exchange to are
* dependent on the value of a message header.
*/
-public class RoutingSlip extends ServiceSupport implements Processor, Traceable {
+public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Traceable {
private static final transient Log LOG = LogFactory.getLog(RoutingSlip.class);
private ProducerCache producerCache;
private boolean ignoreInvalidEndpoints;
@@ -114,58 +116,71 @@ public class RoutingSlip extends Service
}
public void process(Exchange exchange) throws Exception {
+ AsyncProcessorHelper.process(this, exchange);
+ }
+
+ public boolean process(Exchange exchange, AsyncCallback callback) {
if (!isStarted()) {
throw new IllegalStateException("RoutingSlip has not been started: " + this);
- }
-
+ }
+
Object routingSlip = expression.evaluate(exchange, Object.class);
- doRoutingSlip(exchange, routingSlip);
+ return doRoutingSlip(exchange, routingSlip, callback);
}
-
- public void doRoutingSlip(Exchange exchange, Object routingSlip) throws Exception {
+
+ public boolean doRoutingSlip(Exchange exchange, Object routingSlip) {
+ return doRoutingSlip(exchange, routingSlip, new AsyncCallback() {
+ public void done(boolean doneSync) {
+ // noop
+ }
+ });
+ }
+
+ public boolean doRoutingSlip(Exchange exchange, Object routingSlip, AsyncCallback callback) {
Iterator<Object> iter = ObjectHelper.createIterator(routingSlip, uriDelimiter);
Exchange current = exchange;
while (iter.hasNext()) {
- Object nextRecipient = iter.next();
Endpoint endpoint;
try {
- endpoint = resolveEndpoint(exchange, nextRecipient);
- } catch (Exception e) {
- if (isIgnoreInvalidEndpoints()) {
- LOG.info("Endpoint uri is invalid: " + nextRecipient + ". This exception will be ignored.", e);
+ endpoint = resolveEndpoint(iter, exchange);
+ // if no endpoint was resolved then try the next
+ if (endpoint == null) {
continue;
- } else {
- throw e;
}
+ } catch (Exception e) {
+ // error resolving endpoint so we should break out
+ exchange.setException(e);
+ return true;
}
- Exchange copy = new DefaultExchange(current);
- updateRoutingSlipHeader(current);
- copyOutToIn(copy, current);
-
- try {
- producerCache.doInProducer(endpoint, copy, null, new ProducerCallback<Object>() {
- public Object doInProducer(Producer producer, Exchange exchange, ExchangePattern exchangePattern) throws Exception {
- // set property which endpoint we send to
- exchange.setProperty(Exchange.TO_ENDPOINT, producer.getEndpoint().getEndpointUri());
- producer.process(exchange);
- return exchange;
- }
- });
- } catch (Exception e) {
- // Need to check the if the exception is thrown when camel try to create and start the producer
- if (e instanceof FailedToCreateProducerException && isIgnoreInvalidEndpoints()) {
- LOG.info("Endpoint uri is invalid: " + nextRecipient + ". This exception will be ignored.", e);
- continue;
- } else {
- // catch exception so we can decide if we want to continue or not
- copy.setException(e);
+ // prepare and process the routing slip
+ Exchange copy = prepareExchangeForRoutingSlip(current);
+ boolean sync = processExchange(endpoint, copy, exchange, callback, iter);
+ current = copy;
+
+ if (!sync) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchangeId: " + exchange.getExchangeId() + " is continued being processed asynchronously");
+ }
+ // the remainder of the pipeline will be completed async
+ // so we break out now, then the callback will be invoked which then continue routing from where we left here
+ return false;
+ }
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchangeId: " + exchange.getExchangeId() + " is continued being processed synchronously");
+ }
+
+ // we ignore some kind of exceptions and allow us to continue
+ if (isIgnoreInvalidEndpoints()) {
+ FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
+ if (e != null) {
+ LOG.info("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
+ current.setException(null);
}
- } finally {
- current = copy;
}
-
+
// Decide whether to continue with the recipients or not; similar logic to the Pipeline
boolean exceptionHandled = hasExceptionBeenHandledByErrorHandler(current);
if (current.isFailed() || current.isRollbackOnly() || exceptionHandled) {
@@ -191,16 +206,157 @@ public class RoutingSlip extends Service
break;
}
}
+
+ if (LOG.isTraceEnabled()) {
+ // 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: " + exchange.getExchangeId() + " >>> " + current);
+ }
+
+ // copy results back to the original exchange
ExchangeHelper.copyResults(exchange, current);
+
+ callback.done(true);
+ return true;
+ }
+
+ protected Endpoint resolveEndpoint(Iterator<Object> iter, Exchange exchange) throws Exception {
+ Object nextRecipient = iter.next();
+ Endpoint endpoint = null;
+ try {
+ endpoint = ExchangeHelper.resolveEndpoint(exchange, nextRecipient);
+ } catch (Exception e) {
+ if (isIgnoreInvalidEndpoints()) {
+ LOG.info("Endpoint uri is invalid: " + nextRecipient + ". This exception will be ignored.", e);
+ } else {
+ throw e;
+ }
+ }
+ return endpoint;
+ }
+
+ protected Exchange prepareExchangeForRoutingSlip(Exchange current) {
+ Exchange copy = new DefaultExchange(current);
+ // we must use the same id as this is a snapshot strategy where Camel copies a snapshot
+ // before processing the next step in the pipeline, so we have a snapshot of the exchange
+ // just before. This snapshot is used if Camel should do redeliveries (re try) using
+ // DeadLetterChannel. That is why it's important the id is the same, as it is the *same*
+ // exchange being routed.
+ copy.setExchangeId(current.getExchangeId());
+ updateRoutingSlipHeader(current);
+ copyOutToIn(copy, current);
+ return copy;
+ }
+
+ protected boolean processExchange(final Endpoint endpoint, final Exchange exchange, final Exchange original,
+ final AsyncCallback callback, final Iterator<Object> iter) {
+
+ if (LOG.isTraceEnabled()) {
+ // this does the actual processing so log at trace level
+ LOG.trace("Processing exchangeId: " + exchange.getExchangeId() + " >>> " + exchange);
+ }
+
+ 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) {
+ // set property which endpoint we send to
+ exchange.setProperty(Exchange.TO_ENDPOINT, producer.getEndpoint().getEndpointUri());
+ boolean sync = asyncProducer.process(exchange, new AsyncCallback() {
+ public void done(boolean doneSync) {
+ // we only have to handle async completion of the pipeline
+ if (doneSync) {
+ return;
+ }
+
+ // continue processing the routing slip asynchronously
+ Exchange current = exchange;
+
+ while (iter.hasNext()) {
+
+ // we ignore some kind of exceptions and allow us to continue
+ if (isIgnoreInvalidEndpoints()) {
+ FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
+ if (e != null) {
+ LOG.info("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
+ current.setException(null);
+ }
+ }
+
+ // Decide whether to continue with the recipients or not; similar logic to the Pipeline
+ boolean exceptionHandled = hasExceptionBeenHandledByErrorHandler(current);
+ if (current.isFailed() || current.isRollbackOnly() || exceptionHandled) {
+ // The Exchange.ERRORHANDLED_HANDLED property is only set if satisfactory handling was done
+ // by the error handler. It's still an exception, the exchange still failed.
+ if (LOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Message exchange has failed so breaking out of the routing slip: ").append(current);
+ if (current.isRollbackOnly()) {
+ sb.append(" Marked as rollback only.");
+ }
+ if (current.getException() != null) {
+ sb.append(" Exception: ").append(current.getException());
+ }
+ if (current.hasOut() && current.getOut().isFault()) {
+ sb.append(" Fault: ").append(current.getOut());
+ }
+ if (exceptionHandled) {
+ sb.append(" Handled by the error handler.");
+ }
+ LOG.debug(sb.toString());
+ }
+ break;
+ }
+
+ 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;
+ }
+
+ // prepare and process the routing slip
+ Exchange copy = prepareExchangeForRoutingSlip(current);
+ boolean sync = processExchange(endpoint, copy, original, callback, iter);
+ current = copy;
+
+ if (!sync) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchangeId: " + original.getExchangeId() + " is continued being processed asynchronously");
+ }
+ return;
+ }
+ }
+
+ if (LOG.isTraceEnabled()) {
+ // 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);
+ callback.done(false);
+ }
+ });
+
+ return sync;
+ }
+ });
+
+ return sync;
}
private static boolean hasExceptionBeenHandledByErrorHandler(Exchange nextExchange) {
return Boolean.TRUE.equals(nextExchange.getProperty(Exchange.ERRORHANDLER_HANDLED));
}
-
- protected Endpoint resolveEndpoint(Exchange exchange, Object recipient) {
- return ExchangeHelper.resolveEndpoint(exchange, recipient);
- }
protected void doStart() throws Exception {
if (producerCache == null) {
@@ -227,10 +383,9 @@ public class RoutingSlip extends Service
}
}
}
-
+
/**
- * Returns the outbound message if available. Otherwise return the inbound
- * message.
+ * Returns the outbound message if available. Otherwise return the inbound message.
*/
private Message getResultMessage(Exchange exchange) {
if (exchange.hasOut()) {
@@ -241,8 +396,6 @@ public class RoutingSlip extends Service
}
}
-
-
/**
* Copy the outbound data in 'source' to the inbound data in 'result'.
*/
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=956598&r1=956597&r2=956598&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 Mon Jun 21 14:30:12 2010
@@ -25,12 +25,16 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
/**
* Helper methods for {@link AsyncProcessor} objects.
*/
public final class AsyncProcessorHelper {
+ private static final transient Log LOG = LogFactory.getLog(AsyncProcessorHelper.class);
+
private AsyncProcessorHelper() {
// utility class
}
@@ -45,6 +49,9 @@ public final class AsyncProcessorHelper
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());
+ }
latch.countDown();
}
}
@@ -55,6 +62,9 @@ public final class AsyncProcessorHelper
}
});
if (!sync) {
+ if (LOG.isWarnEnabled()) {
+ LOG.warn("Waiting for asynchronous callback before continuing for exchangeId: " + exchange.getExchangeId());
+ }
latch.await();
}
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java Mon Jun 21 14:30:12 2010
@@ -0,0 +1,49 @@
+/**
+ * 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 AsyncEndpointPipelineTest extends ContextTestSupport {
+
+ public void testAsyncEndpoint() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start")
+ .pipeline("async:Bye Camel", "mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointPipelineTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java (from r956466, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java&r1=956466&r2=956598&rev=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java Mon Jun 21 14:30:12 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointTest extends ContextTestSupport {
+public class AsyncEndpointRecipientList2Test extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -57,6 +57,9 @@ public class AsyncEndpointTest extends C
beforeThreadName = Thread.currentThread().getName();
}
})
+ .recipientList(constant("direct:foo"));
+
+ from("direct:foo")
.to("async:Bye Camel")
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
@@ -70,4 +73,4 @@ public class AsyncEndpointTest extends C
};
}
-}
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList2Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: 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/AsyncEndpointRecipientList3Test.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java Mon Jun 21 14:30:12 2010
@@ -0,0 +1,77 @@
+/**
+ * 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 AsyncEndpointRecipientList3Test 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();
+
+ // use same threads as its recipient list, and direct is the last in the recipient list
+ 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,direct:foo"));
+
+ from("direct:foo")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ afterThreadName = Thread.currentThread().getName();
+ exchange.getOut().setBody("Bye Camel");
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientList3Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: 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/AsyncEndpointRecipientListTest.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java Mon Jun 21 14:30:12 2010
@@ -0,0 +1,72 @@
+/**
+ * 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 AsyncEndpointRecipientListTest 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);
+
+ // should use same threads (recipient list is not async supported yet)
+ 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:Bye Camel"))
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ afterThreadName = Thread.currentThread().getName();
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRecipientListTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java Mon Jun 21 14:30:12 2010
@@ -0,0 +1,45 @@
+/**
+ * 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 AsyncEndpointRoutingSlip2Test extends ContextTestSupport {
+
+ public void testAsyncEndpoint() throws Exception {
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye World", reply);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start")
+ .routingSlip(constant("async:Bye Camel,async:Bye World"));
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip2Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java Mon Jun 21 14:30:12 2010
@@ -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.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 AsyncEndpointRoutingSlip3Test 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();
+
+ 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")
+ .to("mock:before")
+ .to("log:before")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ beforeThreadName = Thread.currentThread().getName();
+ }
+ })
+ .routingSlip(constant("async:Hi World,direct:foo"));
+
+ from("direct:foo")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ afterThreadName = Thread.currentThread().getName();
+ String body = exchange.getIn().getBody(String.class);
+ assertEquals("Hi World", body);
+ exchange.getOut().setBody("Bye Camel");
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlip3Test.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java?rev=956598&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java Mon Jun 21 14:30:12 2010
@@ -0,0 +1,49 @@
+/**
+ * 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 AsyncEndpointRoutingSlipTest extends ContextTestSupport {
+
+ public void testAsyncEndpoint() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start")
+ .routingSlip(constant("async:Bye Camel,mock:result"));
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointRoutingSlipTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/MyFileNameExpression.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/MyFileNameExpression.java?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/MyFileNameExpression.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/MyFileNameExpression.java Mon Jun 21 14:30:12 2010
@@ -16,10 +16,9 @@
*/
package org.apache.camel.processor.resequencer;
-import java.util.List;
-
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+
//START SNIPPET: example
public class MyFileNameExpression implements Expression {
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java Mon Jun 21 14:30:12 2010
@@ -16,16 +16,9 @@
*/
package org.apache.camel.processor.resequencer;
-import java.util.List;
-
import org.apache.camel.ContextTestSupport;
-import org.apache.camel.Endpoint;
-import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.impl.EventDrivenConsumerRoute;
-import org.apache.camel.management.JmxSystemPropertyKeys;
-import org.apache.camel.processor.interceptor.StreamCaching;
/**
* @version $Revision$
@@ -43,7 +36,7 @@ public class ResequencerFileNameTest ext
return new RouteBuilder() {
public void configure() {
// START SNIPPET: example
- from("direct:start").resequence(new MyFileNameExpression()).stream().to("mock:result");
+ from("direct:start").resequence(new MyFileNameExpression()).stream().timeout(500).to("mock:result");
// END SNIPPET: example
}
};
@@ -51,11 +44,15 @@ public class ResequencerFileNameTest ext
public void testStreamResequence() throws Exception {
resultEndpoint.expectedBodiesReceived("20090612-D001", "20090612-D003", "20090612-D002", "20090615-D001");
+
template.requestBody("direct:start", "20090612-D003");
template.requestBody("direct:start", "20090612-D001");
+
Thread.sleep(2000);
+
template.requestBody("direct:start", "20090615-D001");
template.requestBody("direct:start", "20090612-D002");
+
resultEndpoint.assertIsSatisfied();
}
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipIgnoreInvalidEndpointsTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipIgnoreInvalidEndpointsTest.java?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipIgnoreInvalidEndpointsTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipIgnoreInvalidEndpointsTest.java Mon Jun 21 14:30:12 2010
@@ -23,7 +23,7 @@ import org.apache.camel.component.mock.M
public class RoutingSlipIgnoreInvalidEndpointsTest extends ContextTestSupport {
- public void testEndpontResolvedFailedWithIgnoreInvalidEndpoints() throws Exception {
+ public void testEndpointResolvedFailedWithIgnoreInvalidEndpoints() throws Exception {
MockEndpoint result = getMockEndpoint("mock:result");
result.expectedBodiesReceived("Hello World");
MockEndpoint end = getMockEndpoint("mock:end");
Modified: camel/trunk/camel-core/src/test/resources/log4j.properties
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/resources/log4j.properties?rev=956598&r1=956597&r2=956598&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/resources/log4j.properties (original)
+++ camel/trunk/camel-core/src/test/resources/log4j.properties Mon Jun 21 14:30:12 2010
@@ -28,6 +28,7 @@ log4j.logger.org.apache.activemq.spring=
#log4j.logger.org.apache.camel.component.mock=DEBUG
#log4j.logger.org.apache.camel.component.file=TRACE
#log4j.logger.org.apache.camel.processor.Pipeline=TRACE
+#log4j.logger.org.apache.camel.processor.RoutingSlip=TRACE
#log4j.logger.org.apache.camel.processor.loadbalancer=TRACE
log4j.logger.org.apache.camel.impl.converter=WARN
log4j.logger.org.apache.camel.management=WARN