camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject [4/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - RC consumer
Date Thu, 16 Mar 2017 12:21:03 GMT
CAMEL-11020: Camel Kubernetes consumers do not close watchers - RC consumer


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

Branch: refs/heads/master
Commit: cad4831054258bcfeaaaec5a6593ae33c0fa0e91
Parents: d44034a
Author: Andrea Cosentino <ancosen@gmail.com>
Authored: Thu Mar 16 12:14:43 2017 +0100
Committer: Andrea Cosentino <ancosen@gmail.com>
Committed: Thu Mar 16 13:20:47 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesNodesConsumer.java       |  4 +--
 ...ubernetesReplicationControllersConsumer.java | 26 +++++++++++++++++---
 2 files changed, 24 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/cad48310/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
index 557c07b..4aa18c2 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
@@ -72,12 +72,12 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null)
{
                 if (nodesWatcher != null) {
-                	nodesWatcher.getWatch().close();
+                    nodesWatcher.getWatch().close();
                 }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
                 if (nodesWatcher != null) {
-                	nodesWatcher.getWatch().close();
+                    nodesWatcher.getWatch().close();
                 }
                 executor.shutdownNow();
             }

http://git-wip-us.apache.org/repos/asf/camel/blob/cad48310/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
index e4f7700..9cb71da 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableReplicationController;
 import io.fabric8.kubernetes.api.model.ReplicationController;
 import io.fabric8.kubernetes.api.model.ReplicationControllerList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.RollableScallableResource;
@@ -42,6 +43,7 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
 
     private final Processor processor;
     private ExecutorService executor;
+    private ReplicationControllersConsumerTask rcWatcher;
 
     public KubernetesReplicationControllersConsumer(KubernetesEndpoint endpoint, Processor
processor) {
         super(endpoint, processor);
@@ -58,8 +60,8 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
         super.doStart();
 
         executor = getEndpoint().createExecutor();
-
-        executor.submit(new ReplicationControllersConsumerTask());       
+        rcWatcher = new ReplicationControllersConsumerTask();
+        executor.submit(rcWatcher);       
     }
 
     @Override
@@ -69,8 +71,14 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
         LOG.debug("Stopping Kubernetes Replication Controllers Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null)
{
+                if (rcWatcher != null) {
+                    rcWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (rcWatcher != null) {
+                    rcWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -78,7 +86,9 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
     }
     
     class ReplicationControllersConsumerTask implements Runnable {
-         
+        
+        private Watch watch;
+        
         @Override
         public void run() {
             MixedOperation<ReplicationController, ReplicationControllerList, DoneableReplicationController,

@@ -93,7 +103,7 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName()))
{
                 w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
             }
-            w.watch(new Watcher<ReplicationController>() {
+            watch = w.watch(new Watcher<ReplicationController>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -120,5 +130,13 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer
{
 
             });
         }
+        
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
+        } 
     }
 }


Mime
View raw message