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:20:17 UTC
[1/4] camel git commit: CAMEL-11020: Camel Kubernetes consumers do
not close watchers - Pods Consumer
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 <an...@gmail.com>
Authored: Thu Mar 16 11:46:49 2017 +0100
Committer: Andrea Cosentino <an...@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;
}
}
}
[3/4] camel git commit: CAMEL-11020: Camel Kubernetes consumers do
not close watchers - RC consumer
Posted by ac...@apache.org.
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/e0a2a9eb
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e0a2a9eb
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e0a2a9eb
Branch: refs/heads/camel-2.18.x
Commit: e0a2a9eb97dc4355c59afd5ecd3f2424e0a34ec1
Parents: 1f92f81
Author: Andrea Cosentino <an...@gmail.com>
Authored: Thu Mar 16 12:14:43 2017 +0100
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Thu Mar 16 13:11:35 2017 +0100
----------------------------------------------------------------------
...ubernetesReplicationControllersConsumer.java | 26 +++++++++++++++++---
1 file changed, 22 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e0a2a9eb/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 76c8347..0081bc6 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
@@ -25,6 +25,7 @@ import io.fabric8.kubernetes.api.model.PodList;
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.ClientMixedOperation;
import io.fabric8.kubernetes.client.dsl.ClientPodResource;
@@ -46,6 +47,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);
@@ -62,8 +64,8 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
super.doStart();
executor = getEndpoint().createExecutor();
-
- executor.submit(new ReplicationControllersConsumerTask());
+ rcWatcher = new ReplicationControllersConsumerTask();
+ executor.submit(rcWatcher);
}
@Override
@@ -73,8 +75,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();
}
}
@@ -82,7 +90,9 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
}
class ReplicationControllersConsumerTask implements Runnable {
-
+
+ private Watch watch;
+
@Override
public void run() {
ClientMixedOperation<ReplicationController, ReplicationControllerList, DoneableReplicationController,
@@ -97,7 +107,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,
@@ -124,5 +134,13 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
});
}
+
+ public Watch getWatch() {
+ return watch;
+ }
+
+ public void setWatch(Watch watch) {
+ this.watch = watch;
+ }
}
}
[4/4] camel git commit: CAMEL-11020: Camel Kubernetes consumers do
not close watchers - Service consumer
Posted by ac...@apache.org.
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/49adc55b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/49adc55b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/49adc55b
Branch: refs/heads/camel-2.18.x
Commit: 49adc55b7c5cbc99f32f67095e5c7e99fe5cff23
Parents: e0a2a9e
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:12:00 2017 +0100
----------------------------------------------------------------------
.../consumer/KubernetesServicesConsumer.java | 23 ++++++++++++++++++--
1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/49adc55b/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 2d658bb..e117bc0 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.ClientMixedOperation;
import io.fabric8.kubernetes.client.dsl.ClientResource;
@@ -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() {
ClientMixedOperation<Service, ServiceList, DoneableService, ClientResource<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;
+ }
}
}
[2/4] camel git commit: CAMEL-11020: Camel Kubernetes consumers do
not close watchers - Namespace Consumer
Posted by ac...@apache.org.
CAMEL-11020: Camel Kubernetes consumers do not close watchers - Namespace Consumer
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1f92f814
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1f92f814
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1f92f814
Branch: refs/heads/camel-2.18.x
Commit: 1f92f814f85f2daa9b2f811b741e280a2d16c5c9
Parents: c2add9f
Author: Andrea Cosentino <an...@gmail.com>
Authored: Thu Mar 16 11:50:56 2017 +0100
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Thu Mar 16 13:09:07 2017 +0100
----------------------------------------------------------------------
.../consumer/KubernetesNamespacesConsumer.java | 23 ++++++++++++++++++--
1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/1f92f814/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
index 9e09105..51d5abe 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableNamespace;
import io.fabric8.kubernetes.api.model.Namespace;
import io.fabric8.kubernetes.api.model.NamespaceList;
import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
import io.fabric8.kubernetes.client.Watcher;
import io.fabric8.kubernetes.client.dsl.ClientNonNamespaceOperation;
import io.fabric8.kubernetes.client.dsl.ClientResource;
@@ -42,6 +43,7 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
private final Processor processor;
private ExecutorService executor;
+ private NamespacesConsumerTask nsWatcher;
public KubernetesNamespacesConsumer(KubernetesEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -58,7 +60,8 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
super.doStart();
executor = getEndpoint().createExecutor();
- executor.submit(new NamespacesConsumerTask());
+ nsWatcher = new NamespacesConsumerTask();
+ executor.submit(nsWatcher);
}
@Override
@@ -68,8 +71,14 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
LOG.debug("Stopping Kubernetes Namespace Consumer");
if (executor != null) {
if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+ if (nsWatcher != null) {
+ nsWatcher.getWatch().close();
+ }
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
} else {
+ if (nsWatcher != null) {
+ nsWatcher.getWatch().close();
+ }
executor.shutdownNow();
}
}
@@ -77,6 +86,8 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
}
class NamespacesConsumerTask implements Runnable {
+
+ private Watch watch;
@Override
public void run() {
@@ -84,7 +95,7 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getNamespace())) {
w.withName(getEndpoint().getKubernetesConfiguration().getNamespace());
}
- w.watch(new Watcher<Namespace>() {
+ watch = w.watch(new Watcher<Namespace>() {
@Override
public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -108,6 +119,14 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
}
}
});
+ }
+
+ public Watch getWatch() {
+ return watch;
+ }
+
+ public void setWatch(Watch watch) {
+ this.watch = watch;
}
}
}