camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject [1/4] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Pods Consumer
Date Thu, 16 Mar 2017 12:20:17 GMT
Repository: camel
Updated Branches:
  refs/heads/camel-2.18.x b0e35cb47 -> 49adc55b7


CAMEL-11020: Camel Kubernetes consumers do not close watchers - Pods Consumer


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

Branch: refs/heads/camel-2.18.x
Commit: c2add9fd951f6acbf49a30f0e1909e6f116861df
Parents: b0e35cb
Author: Andrea Cosentino <ancosen@gmail.com>
Authored: Thu Mar 16 11:46:49 2017 +0100
Committer: Andrea Cosentino <ancosen@gmail.com>
Committed: Thu Mar 16 13:08:58 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesPodsConsumer.java        | 25 +++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/c2add9fd/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java
index 24947b0..9abf29b 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneablePod;
 import io.fabric8.kubernetes.api.model.Pod;
 import io.fabric8.kubernetes.api.model.PodList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.ClientMixedOperation;
 import io.fabric8.kubernetes.client.dsl.ClientNonNamespaceOperation;
@@ -43,6 +44,7 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
 
     private final Processor processor;
     private ExecutorService executor;
+    private PodsConsumerTask podsWatcher;
 
     public KubernetesPodsConsumer(KubernetesEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -58,8 +60,9 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
     protected void doStart() throws Exception {
         super.doStart();
         executor = getEndpoint().createExecutor();
-
-        executor.submit(new PodsConsumerTask());
+        
+        podsWatcher = new PodsConsumerTask();
+        executor.submit(podsWatcher);
     }
 
     @Override
@@ -69,8 +72,14 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
         LOG.debug("Stopping Kubernetes Pods Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null)
{
+                if (podsWatcher != null) {
+                    podsWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (podsWatcher != null) {
+                    podsWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -78,6 +87,8 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
     }
 
     class PodsConsumerTask implements Runnable {
+
+        private Watch watch;
         
         @Override
         public void run() {
@@ -92,7 +103,7 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName()))
{
                 w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
             }
-            w.watch(new Watcher<Pod>() {
+            watch = w.watch(new Watcher<Pod>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -117,6 +128,14 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
 
                 }
             });
+        }
+
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
         } 
     }
 }


Mime
View raw message