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 2012/10/07 08:34:50 UTC
svn commit: r1395228 - in /camel/branches/camel-2.10.x: ./
camel-core/src/main/java/org/apache/camel/impl/
camel-core/src/test/java/org/apache/camel/impl/
Author: davsclaus
Date: Sun Oct 7 06:34:50 2012
New Revision: 1395228
URL: http://svn.apache.org/viewvc?rev=1395228&view=rev
Log:
CAMEL-5683: ProducerCache/ConsumerCache should use regular LRUCache to ensure prouducer/consumer is always stopped when evicted/stopping cache itself. As a soft cache may GC the entries without stopping any may cause internal resources running, which should have been stopped otherwise.
Modified:
camel/branches/camel-2.10.x/ (props changed)
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1395226
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified: camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java?rev=1395228&r1=1395227&r2=1395228&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ConsumerCache.java Sun Oct 7 06:34:50 2012
@@ -27,7 +27,6 @@ import org.apache.camel.PollingConsumer;
import org.apache.camel.support.ServiceSupport;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.LRUCache;
-import org.apache.camel.util.LRUSoftCache;
import org.apache.camel.util.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,14 +59,17 @@ public class ConsumerCache extends Servi
/**
* Creates the {@link LRUCache} to be used.
* <p/>
- * This implementation returns a {@link org.apache.camel.util.LRUSoftCache} instance.
+ * This implementation returns a {@link LRUCache} instance.
* @param cacheSize the cache size
* @return the cache
*/
protected static LRUCache<String, PollingConsumer> createLRUCache(int cacheSize) {
- // We use a soft reference cache to allow the JVM to re-claim memory if it runs low on memory.
- return new LRUSoftCache<String, PollingConsumer>(cacheSize);
+ // Use a regular cache as we want to ensure that the lifecycle of the consumers
+ // being cache is properly handled, such as they are stopped when being evicted
+ // or when this cache is stopped. This is needed as some consumers requires to
+ // be stopped so they can shutdown internal resources that otherwise may cause leaks
+ return new LRUCache<String, PollingConsumer>(cacheSize);
}
public synchronized PollingConsumer getConsumer(Endpoint endpoint) {
Modified: camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java?rev=1395228&r1=1395227&r2=1395228&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java Sun Oct 7 06:34:50 2012
@@ -37,7 +37,6 @@ import org.apache.camel.util.AsyncProces
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.EventHelper;
import org.apache.camel.util.LRUCache;
-import org.apache.camel.util.LRUSoftCache;
import org.apache.camel.util.ServiceHelper;
import org.apache.camel.util.StopWatch;
import org.slf4j.Logger;
@@ -78,14 +77,17 @@ public class ProducerCache extends Servi
/**
* Creates the {@link LRUCache} to be used.
* <p/>
- * This implementation returns a {@link LRUSoftCache} instance.
+ * This implementation returns a {@link LRUCache} instance.
* @param cacheSize the cache size
* @return the cache
*/
protected static LRUCache<String, Producer> createLRUCache(int cacheSize) {
- // We use a soft reference cache to allow the JVM to re-claim memory if it runs low on memory.
- return new LRUSoftCache<String, Producer>(cacheSize);
+ // Use a regular cache as we want to ensure that the lifecycle of the producers
+ // being cache is properly handled, such as they are stopped when being evicted
+ // or when this cache is stopped. This is needed as some producers requires to
+ // be stopped so they can shutdown internal resources that otherwise may cause leaks
+ return new LRUCache<String, Producer>(cacheSize);
}
public CamelContext getCamelContext() {
Modified: camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java?rev=1395228&r1=1395227&r2=1395228&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java Sun Oct 7 06:34:50 2012
@@ -16,8 +16,13 @@
*/
package org.apache.camel.impl;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Consumer;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
import org.apache.camel.Producer;
/**
@@ -25,6 +30,8 @@ import org.apache.camel.Producer;
*/
public class DefaultProducerCacheTest extends ContextTestSupport {
+ private static final AtomicInteger counter = new AtomicInteger();
+
public void testCacheProducerAcquireAndRelease() throws Exception {
ProducerCache cache = new ProducerCache(this, context);
cache.start();
@@ -42,4 +49,73 @@ public class DefaultProducerCacheTest ex
cache.stop();
}
+ public void testCacheStopExpired() throws Exception {
+ ProducerCache cache = new ProducerCache(this, context, 5);
+ cache.start();
+
+ assertEquals("Size should be 0", 0, cache.size());
+
+ for (int i = 0; i < 8; i++) {
+ Endpoint e = new MyEndpoint(i);
+ Producer p = cache.acquireProducer(e);
+ cache.releaseProducer(e, p);
+ }
+
+ assertEquals("Size should be 5", 5, cache.size());
+
+ // should have stopped the 3 evicted
+ assertEquals(3, counter.get());
+
+ cache.stop();
+
+ // should have stopped all 8
+ assertEquals(8, counter.get());
+ }
+
+ private class MyEndpoint extends DefaultEndpoint {
+
+ private int number;
+
+ private MyEndpoint(int number) {
+ this.number = number;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new MyProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return null;
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ @Override
+ public String getEndpointUri() {
+ return "my://" + number;
+ }
+ }
+
+ private class MyProducer extends DefaultProducer {
+
+ public MyProducer(Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ // noop
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ counter.incrementAndGet();
+ }
+ }
+
}