svn commit: r1355781 - in /camel/branches/camel-2.8.x: ./ camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java camel-core/src/test/java/org/apache/camel/issues/ThreadsDoTryCatchInterceptSendToAllEndpointIssueTest.java

[da...@apache.org]

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;
                 }
             }