You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/04/12 11:00:09 UTC
[3/5] camel git commit: CAMEL-11138: ConsumerTemplate - If cache is
full then polling consumer should be stopped to not leak resources
CAMEL-11138: ConsumerTemplate - If cache is full then polling consumer should be stopped to not leak resources
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/bb68f965
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/bb68f965
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/bb68f965
Branch: refs/heads/camel-2.18.x
Commit: bb68f96597e82eacdfafc54ec8fe3fa849c6b191
Parents: 6886c0b
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Apr 12 12:58:08 2017 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Apr 12 12:58:59 2017 +0200
----------------------------------------------------------------------
.../org/apache/camel/impl/ConsumerCache.java | 11 +++-
.../impl/ConsumerCacheZeroCapacityTest.java | 59 ++++++++++++++++++++
2 files changed, 68 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/bb68f965/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java b/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
index 2e6fdda..20deb90 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
@@ -134,9 +134,16 @@ public class ConsumerCache extends ServiceSupport {
if (pollingConsumer instanceof IsSingleton) {
singleton = ((IsSingleton) pollingConsumer).isSingleton();
}
- if (!singleton) {
+ String key = endpoint.getEndpointUri();
+ boolean cached = consumers.containsKey(key);
+ if (!singleton || !cached) {
try {
- // stop and shutdown non-singleton producers as we should not leak resources
+ // stop and shutdown non-singleton/non-cached consumers as we should not leak resources
+ if (!singleton) {
+ LOG.debug("Released PollingConsumer: {} is stopped as consumer is not singleton", endpoint);
+ } else {
+ LOG.debug("Released PollingConsumer: {} is stopped as consumer cache is full", endpoint);
+ }
ServiceHelper.stopAndShutdownService(pollingConsumer);
} catch (Exception ex) {
if (ex instanceof RuntimeCamelException) {
http://git-wip-us.apache.org/repos/asf/camel/blob/bb68f965/camel-core/src/test/java/org/apache/camel/impl/ConsumerCacheZeroCapacityTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/impl/ConsumerCacheZeroCapacityTest.java b/camel-core/src/test/java/org/apache/camel/impl/ConsumerCacheZeroCapacityTest.java
new file mode 100644
index 0000000..3cbf2d4
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/impl/ConsumerCacheZeroCapacityTest.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.impl;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.PollingConsumer;
+
+/**
+ * @version
+ */
+public class ConsumerCacheZeroCapacityTest extends ContextTestSupport {
+
+ public void testConsumerCacheZeroCapacity() throws Exception {
+ ConsumerCache cache = new ConsumerCache(this, context, 0);
+ cache.start();
+
+ assertEquals("Size should be 0", 0, cache.size());
+
+ Endpoint endpoint = context.getEndpoint("file:target/foo?fileName=foo.txt");
+ PollingConsumer consumer = cache.acquirePollingConsumer(endpoint);
+ assertNotNull(consumer);
+ assertEquals("Started", ((org.apache.camel.support.ServiceSupport) consumer).getStatus().name());
+
+ // let it run a poll
+ consumer.receive(1000);
+
+ boolean found = Thread.getAllStackTraces().keySet().stream().anyMatch(t -> t.getName().contains("target/foo"));
+ assertTrue("Should find file consumer thread", found);
+
+ cache.releasePollingConsumer(endpoint, consumer);
+
+ // takes a little to stop
+ Thread.sleep(1000);
+
+ assertEquals("Stopped", ((org.apache.camel.support.ServiceSupport) consumer).getStatus().name());
+
+ // should not be a file consumer thread
+ found = Thread.getAllStackTraces().keySet().stream().anyMatch(t -> t.getName().contains("target/foo"));
+ assertFalse("Should not find file consumer thread", found);
+
+ cache.stop();
+ }
+
+}
\ No newline at end of file