camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/2] git commit: CAMEL-6390: Suspended seda consumers should break out if no tasks on queue. And also break out if forced to shutdown.
Date Thu, 30 May 2013 09:12:12 GMT
CAMEL-6390: Suspended seda consumers should break out if no tasks on queue. And also break
out if forced to shutdown.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/2806e816
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/2806e816
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/2806e816

Branch: refs/heads/camel-2.11.x
Commit: 2806e816d05c994f6f5c9d18124d920cf3694bcf
Parents: 5438802
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Thu May 30 11:11:33 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Thu May 30 11:11:55 2013 +0200

----------------------------------------------------------------------
 .../apache/camel/component/seda/SedaConsumer.java  |   28 +++++++++++----
 1 files changed, 21 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/2806e816/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java b/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
index 1b7f42b..bd5b2c3 100644
--- a/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
@@ -59,6 +59,7 @@ public class SedaConsumer extends ServiceSupport implements Consumer, Runnable,
     private final AtomicInteger taskCount = new AtomicInteger();
     private volatile CountDownLatch latch;
     private volatile boolean shutdownPending;
+    private volatile boolean forceShutdown;
     private SedaEndpoint endpoint;
     private AsyncProcessor processor;
     private ExecutorService executor;
@@ -110,6 +111,7 @@ public class SedaConsumer extends ServiceSupport implements Consumer,
Runnable,
     public void prepareShutdown(boolean forced) {
         // signal we want to shutdown
         shutdownPending = true;
+        forceShutdown = forced;
 
         if (latch != null) {
             LOG.debug("Preparing to shutdown, waiting for {} consumer threads to complete.",
latch.getCount());
@@ -125,6 +127,11 @@ public class SedaConsumer extends ServiceSupport implements Consumer,
Runnable,
 
     @Override
     public boolean isRunAllowed() {
+        // if we force shutdown then do not allow running anymore
+        if (forceShutdown) {
+            return false;
+        }
+
         if (isSuspending() || isSuspended()) {
             // allow to run even if we are suspended as we want to
             // keep the thread task running
@@ -161,14 +168,20 @@ public class SedaConsumer extends ServiceSupport implements Consumer,
Runnable,
 
             // do not poll if we are suspended
             if (isSuspending() || isSuspended()) {
-                LOG.trace("Consumer is suspended so skip polling");
-                try {
-                    // sleep at most 1 sec
-                    Thread.sleep(Math.min(pollTimeout, 1000));
-                } catch (InterruptedException e) {
-                    LOG.debug("Sleep interrupted, are we stopping? {}", isStopping() || isStopped());
+                if (shutdownPending && queue.isEmpty()) {
+                    LOG.trace("Consumer is suspended and shutdown is pending, so this consumer
thread is breaking out because the task queue is empty.");
+                    // we want to shutdown so break out if there queue is empty
+                    break;
+                } else {
+                    LOG.trace("Consumer is suspended so skip polling");
+                    try {
+                        // sleep at most 1 sec
+                        Thread.sleep(Math.min(pollTimeout, 1000));
+                    } catch (InterruptedException e) {
+                        LOG.debug("Sleep interrupted, are we stopping? {}", isStopping()
|| isStopped());
+                    }
+                    continue;
                 }
-                continue;
             }
 
             Exchange exchange = null;
@@ -285,6 +298,7 @@ public class SedaConsumer extends ServiceSupport implements Consumer,
Runnable,
     protected void doStart() throws Exception {
         latch = new CountDownLatch(endpoint.getConcurrentConsumers());
         shutdownPending = false;
+        forceShutdown = false;
 
         setupTasks();
         endpoint.onStarted(this);


Mime
View raw message