Author: davsclaus
Date: Sat Jun 30 17:28:57 2012
New Revision: 1355781
URL: http://svn.apache.org/viewvc?rev=1355781&view=rev
Log:
CAMEL-5406: Fixed mock endpoints interceptor to support async routing engine,
to ensure callback is properly called if async kicks in.
Added:
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/issues/ThreadsDoTryCatchInterceptSendToAllEndpointIssueTest.java
- copied unchanged from r1355779,
camel/branches/camel-2.9.x/camel-core/src/test/java/org/apache/camel/issues/ThreadsDoTryCatchInterceptSendToAllEndpointIssueTest.java
Modified:
camel/branches/camel-2.8.x/ (props changed)
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
Propchange: camel/branches/camel-2.8.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1355774
Merged /camel/branches/camel-2.9.x:r1355779
Propchange: camel/branches/camel-2.8.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
URL:
http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java?rev=1355781&r1=1355780&r2=1355781&view=diff
==============================================================================
---
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
(original)
+++
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
Sat Jun 30 17:28:57 2012
@@ -18,6 +18,8 @@ package org.apache.camel.impl;
import java.util.Map;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
@@ -26,6 +28,8 @@ import org.apache.camel.ExchangePattern;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
+import org.apache.camel.impl.converter.AsyncProcessorTypeConverter;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,7 +95,7 @@ public class InterceptSendToEndpoint imp
public Producer createProducer() throws Exception {
producer = delegate.createProducer();
- return new Producer() {
+ return new DefaultAsyncProducer(delegate) {
public Endpoint getEndpoint() {
return producer.getEndpoint();
@@ -109,7 +113,8 @@ public class InterceptSendToEndpoint imp
return producer.createExchange(exchange);
}
- public void process(Exchange exchange) throws Exception {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
// process the detour so we do the detour routing
if (LOG.isDebugEnabled()) {
LOG.debug("Sending to endpoint: {} is intercepted and
detoured to: {} for exchange: {}", new Object[]{getEndpointUri(), detour,
exchange});
@@ -117,16 +122,20 @@ public class InterceptSendToEndpoint imp
// add header with the real endpoint uri
exchange.getIn().setHeader(Exchange.INTERCEPTED_ENDPOINT,
delegate.getEndpointUri());
+ // detour the exchange using synchronous processing
try {
detour.process(exchange);
} catch (Exception e) {
exchange.setException(e);
+ callback.done(true);
+ return true;
}
// Decide whether to continue or not; similar logic to the
Pipeline
// check for error if so we should break out
if (!continueProcessing(exchange, "skip sending to original
intended destination: " + getEndpointUri(), LOG)) {
- return;
+ callback.done(true);
+ return true;
}
if (!skip) {
@@ -136,10 +145,15 @@ public class InterceptSendToEndpoint imp
exchange.setOut(null);
}
- // route to original destination
- producer.process(exchange);
+ // route to original destination leveraging the
asynchronous routing engine
+ AsyncProcessor async =
AsyncProcessorTypeConverter.convert(producer);
+ return AsyncProcessorHelper.process(async, exchange,
callback);
} else {
- LOG.debug("Stop() means skip sending exchange to original
intended destination: {} for exchange: {}", getEndpointUri(), exchange);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Stop() means skip sending exchange to
original intended destination: {} for exchange: {}", getEndpoint(), exchange);
+ }
+ callback.done(true);
+ return true;
}
}
