You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2017/03/16 12:21:04 UTC
[5/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do
not close watchers - Service consumer
CAMEL-11020: Camel Kubernetes consumers do not close watchers - Service consumer
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f9e05f74
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f9e05f74
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f9e05f74
Branch: refs/heads/master
Commit: f9e05f740587b00964e99aaab9c2f800f0c19dca
Parents: cad4831
Author: Andrea Cosentino <an...@gmail.com>
Authored: Thu Mar 16 13:06:58 2017 +0100
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Thu Mar 16 13:20:47 2017 +0100
----------------------------------------------------------------------
.../consumer/KubernetesServicesConsumer.java | 23 ++++++++++++++++++--
1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f9e05f74/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
index 1b87ab8..5353f76 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableService;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.ServiceList;
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.Resource;
@@ -42,6 +43,7 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
private final Processor processor;
private ExecutorService executor;
+ private ServicesConsumerTask servicesWatcher;
public KubernetesServicesConsumer(KubernetesEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -58,7 +60,8 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
super.doStart();
executor = getEndpoint().createExecutor();
- executor.submit(new ServicesConsumerTask());
+ servicesWatcher = new ServicesConsumerTask();
+ executor.submit(servicesWatcher);
}
@@ -68,8 +71,14 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
LOG.debug("Stopping Kubernetes Services Consumer");
if (executor != null) {
if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+ if (servicesWatcher != null) {
+ servicesWatcher.getWatch().close();
+ }
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
} else {
+ if (servicesWatcher != null) {
+ servicesWatcher.getWatch().close();
+ }
executor.shutdownNow();
}
}
@@ -78,6 +87,8 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
class ServicesConsumerTask implements Runnable {
+ private Watch watch;
+
@Override
public void run() {
MixedOperation<Service, ServiceList, DoneableService, Resource<Service, DoneableService>> w = getEndpoint().getKubernetesClient().services();
@@ -91,7 +102,7 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName())) {
w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
}
- w.watch(new Watcher<Service>() {
+ watch = w.watch(new Watcher<Service>() {
@Override
public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -118,5 +129,13 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
});
}
+
+ public Watch getWatch() {
+ return watch;
+ }
+
+ public void setWatch(Watch watch) {
+ this.watch = watch;
+ }
}
}