camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r982655 - in /camel/trunk/camel-core/src: main/java/org/apache/camel/processor/ main/java/org/apache/camel/util/ test/java/org/apache/camel/processor/ test/resources/
Date Thu, 05 Aug 2010 15:10:27 GMT
Author: davsclaus
Date: Thu Aug  5 15:10:27 2010
New Revision: 982655

URL: http://svn.apache.org/viewvc?rev=982655&view=rev
Log:
CAMEL-3029: Routing engine should stop continue routing Exchange if the thread was interrupted
by the JDK (such as future task cancelled etc.)

Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelayProcessorSupport.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/Pipeline.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
    camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutTest.java
    camel/trunk/camel-core/src/test/resources/log4j.properties

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java?rev=982655&r1=982654&r2=982655&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
Thu Aug  5 15:10:27 2010
@@ -117,7 +117,11 @@ public abstract class DelayProcessorSupp
                         exchange.setException(new RejectedExecutionException());
                     } else {
                         // let caller run by processing
-                        delay(delay, exchange);
+                        try {
+                            delay(delay, exchange);
+                        } catch (InterruptedException ie) {
+                            exchange.setException(ie);
+                        }
                         // then continue routing
                         return super.process(exchange, callback);
                     }
@@ -157,7 +161,7 @@ public abstract class DelayProcessorSupp
      * @param delay the delay time in millis
      * @param exchange the exchange being processed
      */
-    protected void delay(long delay, Exchange exchange) {
+    protected void delay(long delay, Exchange exchange) throws InterruptedException {
         // only run is we are started
         if (!isRunAllowed()) {
             return;
@@ -169,19 +173,20 @@ public abstract class DelayProcessorSupp
             try {
                 sleep(delay);
             } catch (InterruptedException e) {
-                handleSleepInterruptedException(e);
+                handleSleepInterruptedException(e, exchange);
             }
         }
     }
 
     /**
-     * Called when a sleep is interrupted; allows derived classes to handle this
-     * case differently
+     * Called when a sleep is interrupted; allows derived classes to handle this case differently
      */
-    protected void handleSleepInterruptedException(InterruptedException e) {
+    protected void handleSleepInterruptedException(InterruptedException e, Exchange exchange)
throws InterruptedException {
         if (log.isDebugEnabled()) {
             log.debug("Sleep interrupted, are we stopping? " + (isStopping() || isStopped()));
         }
+        Thread.currentThread().interrupt();
+        throw e;
     }
 
     protected long currentSystemTime() {

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=982655&r1=982654&r2=982655&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
Thu Aug  5 15:10:27 2010
@@ -261,6 +261,7 @@ public class MulticastProcessor extends 
 
         // 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
+        // TODO: use a stopwatch to keep track of timeout left
         boolean timedOut = false;
         for (int i = 0; i < total.intValue(); i++) {
             Future<Exchange> future;

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=982655&r1=982654&r2=982655&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 Thu Aug
 5 15:10:27 2010
@@ -236,7 +236,7 @@ public class Pipeline extends MulticastP
             boolean doStop = exchange.getContext().getTypeConverter().convertTo(Boolean.class,
stop);
             if (doStop) {
                 if (LOG.isDebugEnabled()) {
-                    LOG.debug("Exchange is marked to stop routing: " + exchange);
+                    LOG.debug("ExchangeId: " + exchange.getExchangeId() + " is marked to
stop routing: " + exchange);
                 }
                 answer = false;
             }
@@ -246,7 +246,7 @@ public class Pipeline extends MulticastP
         }
 
         if (LOG.isTraceEnabled()) {
-            LOG.trace("Continue routing: " + answer);
+            LOG.trace("ExchangeId: " + exchange.getExchangeId() + " should continue routing:
" + answer);
         }
         return answer;
     }

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java?rev=982655&r1=982654&r2=982655&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
Thu Aug  5 15:10:27 2010
@@ -270,6 +270,9 @@ public abstract class RedeliveryErrorHan
                         } catch (InterruptedException e) {
                             // we was interrupted so break out
                             exchange.setException(e);
+                            // mark the exchange to stop continue routing when interrupted
+                            // as we do not want to continue routing (for example a task
has been cancelled)
+                            exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);
                             callback.done(data.sync);
                             return data.sync;
                         }
@@ -396,12 +399,16 @@ public abstract class RedeliveryErrorHan
      * Strategy to determine if the exchange is done so we can continue
      */
     protected boolean isDone(Exchange exchange) {
+        boolean answer = isCancelledOrInterrupted(exchange);
+
         // only done if the exchange hasn't failed
         // and it has not been handled by the failure processor
         // or we are exhausted
-        boolean answer = exchange.getException() == null
+        if (!answer) {
+            answer = exchange.getException() == null
                 || ExchangeHelper.isFailureHandled(exchange)
                 || ExchangeHelper.isRedeliveryExhausted(exchange);
+        }
 
         if (log.isTraceEnabled()) {
             log.trace("Is exchangeId: " + exchange.getExchangeId() + " done? " + answer);
@@ -410,6 +417,25 @@ public abstract class RedeliveryErrorHan
     }
 
     /**
+     * Strategy to determine if the exchange was cancelled or interrupted
+     */
+    protected boolean isCancelledOrInterrupted(Exchange exchange) {
+        boolean answer = false;
+
+        if (ExchangeHelper.isInterrupted(exchange)) {
+            // mark the exchange to stop continue routing when interrupted
+            // as we do not want to continue routing (for example a task has been cancelled)
+            exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);
+            answer = true;
+        }
+
+        if (log.isTraceEnabled()) {
+            log.trace("Is exchangeId: " + exchange.getExchangeId() + " interrupted? " + answer);
+        }
+        return answer;
+    }
+
+    /**
      * Returns the output processor
      */
     public Processor getOutput() {

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java?rev=982655&r1=982654&r2=982655&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java Thu Aug
 5 15:10:27 2010
@@ -433,6 +433,10 @@ public final class ExchangeHelper {
         return exchange.getProperty(Exchange.REDELIVERY_EXHAUSTED, false, Boolean.class);
     }
 
+    public static boolean isInterrupted(Exchange exchange) {
+        return exchange.getException(InterruptedException.class) != null;
+    }
+
     /**
      * Extracts the body from the given exchange.
      * <p/>

Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutTest.java?rev=982655&r1=982654&r2=982655&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutTest.java
(original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutTest.java
Thu Aug  5 15:10:27 2010
@@ -32,8 +32,16 @@ public class MulticastParallelTimeoutTes
         // A will timeout so we only get B and C
         mock.expectedBodiesReceived("BC");
 
+        getMockEndpoint("mock:A").expectedMessageCount(0);
+        getMockEndpoint("mock:B").expectedMessageCount(1);
+        getMockEndpoint("mock:C").expectedMessageCount(1);
+
         template.sendBody("direct:start", "Hello");
 
+        // wait at least longer than the delay in A so we can ensure its being cancelled
+        // and wont continue routing
+        Thread.sleep(4000);
+
         assertMockEndpointsSatisfied();
     }
 
@@ -60,11 +68,11 @@ public class MulticastParallelTimeoutTes
                     .end()
                     .to("mock:result");
 
-                from("direct:a").delay(3000).setBody(constant("A"));
+                from("direct:a").delay(3000).to("mock:A").setBody(constant("A"));
 
-                from("direct:b").setBody(constant("B"));
+                from("direct:b").to("mock:B").setBody(constant("B"));
 
-                from("direct:c").delay(500).setBody(constant("C"));
+                from("direct:c").delay(500).to("mock:C").setBody(constant("C"));
                 // END SNIPPET: e1
             }
         };

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=982655&r1=982654&r2=982655&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/resources/log4j.properties (original)
+++ camel/trunk/camel-core/src/test/resources/log4j.properties Thu Aug  5 15:10:27 2010
@@ -41,7 +41,10 @@ log4j.logger.org.apache.camel.management
 log4j.logger.org.apache.camel.impl.DefaultPackageScanClassResolver=WARN
 #log4j.logger.org.apache.camel.impl=TRACE
 #log4j.logger.org.apache.camel.util.FileUtil=TRACE
+#log4j.logger.org.apache.camel.util.AsyncProcessorHelper=TRACE
 #log4j.logger.org.apache.camel.impl.converter.DefaultTypeConverter=TRACE
+#log4j.logger.org.apache.camel.processor.DefaultErrorHandler=TRACE
+#log4j.logger.org.apache.camel.processor.Pipeline=TRACE
 
 # CONSOLE appender not used by default
 log4j.appender.out=org.apache.log4j.ConsoleAppender



Mime
View raw message