Author: davsclaus
Date: Sat Jun 30 17:24:25 2012
New Revision: 1355779
URL: http://svn.apache.org/viewvc?rev=1355779&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.9.x/camel-core/src/test/java/org/apache/camel/issues/ThreadsDoTryCatchInterceptSendToAllEndpointIssueTest.java
- copied unchanged from r1355774,
camel/trunk/camel-core/src/test/java/org/apache/camel/issues/ThreadsDoTryCatchInterceptSendToAllEndpointIssueTest.java
Modified:
camel/branches/camel-2.9.x/ (props changed)
camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1355774
Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
URL:
http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java?rev=1355779&r1=1355778&r2=1355779&view=diff
==============================================================================
---
camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
(original)
+++
camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java
Sat Jun 30 17:24:25 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;
@@ -27,6 +29,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.util.AsyncProcessorConverterHelper;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -95,7 +99,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();
@@ -113,7 +117,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});
@@ -121,16 +126,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) {
@@ -140,10 +149,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 =
AsyncProcessorConverterHelper.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;
}
}
