You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/12/05 06:42:25 UTC
[4/6] camel git commit: CAMEL-8124 Using ServicePool in ConsumerCache
CAMEL-8124 Using ServicePool in ConsumerCache
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/27e13d91
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/27e13d91
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/27e13d91
Branch: refs/heads/master
Commit: 27e13d916cc918d04b06621777da9aedcba144b4
Parents: ed1e629
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Dec 5 10:23:28 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Dec 5 11:02:05 2014 +0800
----------------------------------------------------------------------
.../java/org/apache/camel/CamelContext.java | 14 ++
.../component/seda/SedaPollingConsumer.java | 8 +-
.../org/apache/camel/impl/ConsumerCache.java | 127 +++++++++++++++----
.../apache/camel/impl/DefaultCamelContext.java | 15 ++-
.../impl/DefaultPollingConsumerServicePool.java | 39 ++++++
.../camel/impl/EventDrivenPollingConsumer.java | 9 +-
.../camel/impl/ProcessorPollingConsumer.java | 9 +-
.../impl/SharedPollingConsumerServicePool.java | 49 +++++++
.../component/file/FileConsumeTemplateTest.java | 6 +-
.../file/FileConsumerTemplateTest.java | 5 +-
.../component/http/HttpPollingConsumer.java | 3 +-
.../component/http4/HttpPollingConsumer.java | 3 +-
.../apache/camel/component/jms/JmsEndpoint.java | 3 +-
.../camel/component/jms/JmsPollingConsumer.java | 10 +-
14 files changed, 264 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/CamelContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/CamelContext.java b/camel-core/src/main/java/org/apache/camel/CamelContext.java
index 8fc03ad..9906047 100644
--- a/camel-core/src/main/java/org/apache/camel/CamelContext.java
+++ b/camel-core/src/main/java/org/apache/camel/CamelContext.java
@@ -1070,8 +1070,22 @@ public interface CamelContext extends SuspendableService, RuntimeConfiguration {
* @return the service pool
*/
ServicePool<Endpoint, Producer> getProducerServicePool();
+
+ /**
+ * Sets a pluggable service pool to use for {@link PollingConsumer} pooling.
+ *
+ * @param servicePool the pool
+ */
+ void setPollingConsumerServicePool(ServicePool<Endpoint, PollingConsumer> servicePool);
/**
+ * Gets the service pool for {@link Producer} pooling.
+ *
+ * @return the service pool
+ */
+ ServicePool<Endpoint, PollingConsumer> getPollingConsumerServicePool();
+
+ /**
* Uses a custom node id factory when generating auto assigned ids to the nodes in the route definitions
*
* @param factory custom factory to use
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java b/camel-core/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
index 06bf3e9..7243906 100644
--- a/camel-core/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
@@ -20,10 +20,11 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.IsSingleton;
import org.apache.camel.impl.PollingConsumerSupport;
import org.apache.camel.util.ObjectHelper;
-public class SedaPollingConsumer extends PollingConsumerSupport {
+public class SedaPollingConsumer extends PollingConsumerSupport implements IsSingleton {
public SedaPollingConsumer(Endpoint endpoint) {
super(endpoint);
@@ -68,4 +69,9 @@ public class SedaPollingConsumer extends PollingConsumerSupport {
protected void doStop() throws Exception {
// noop
}
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/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 548ec1d..6f60f46 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
@@ -24,6 +24,9 @@ import org.apache.camel.Exchange;
import org.apache.camel.FailedToCreateConsumerException;
import org.apache.camel.IsSingleton;
import org.apache.camel.PollingConsumer;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.ServicePoolAware;
+import org.apache.camel.spi.ServicePool;
import org.apache.camel.support.ServiceSupport;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.LRUCache;
@@ -39,6 +42,7 @@ import org.slf4j.LoggerFactory;
public class ConsumerCache extends ServiceSupport {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerCache.class);
private final CamelContext camelContext;
+ private final ServicePool<Endpoint, PollingConsumer> pool;
private final Map<String, PollingConsumer> consumers;
private final Object source;
@@ -49,11 +53,17 @@ public class ConsumerCache extends ServiceSupport {
public ConsumerCache(Object source, CamelContext camelContext, int cacheSize) {
this(source, camelContext, createLRUCache(cacheSize));
}
-
+
public ConsumerCache(Object source, CamelContext camelContext, Map<String, PollingConsumer> cache) {
+ this(source, camelContext, cache, camelContext.getPollingConsumerServicePool());
+ }
+
+
+ public ConsumerCache(Object source, CamelContext camelContext, Map<String, PollingConsumer> cache, ServicePool<Endpoint, PollingConsumer> pool) {
this.camelContext = camelContext;
this.consumers = cache;
this.source = source;
+ this.pool = pool;
}
/**
@@ -71,10 +81,59 @@ public class ConsumerCache extends ServiceSupport {
// be stopped so they can shutdown internal resources that otherwise may cause leaks
return new LRUCache<String, PollingConsumer>(cacheSize);
}
+
+ /**
+ * Acquires a pooled PollingConsumer which you <b>must</b> release back again after usage using the
+ * {@link #releasePollingConsumer(org.apache.camel.Endpoint, org.apache.camel.PollingConsumer)} method.
+ *
+ * @param endpoint the endpoint
+ * @return the PollingConsumer
+ */
+ public PollingConsumer acquirePollingConsumer(Endpoint endpoint) {
+ return doGetPollingConsumer(endpoint, true);
+ }
- public synchronized PollingConsumer getConsumer(Endpoint endpoint) {
+ /**
+ * Releases an acquired producer back after usage.
+ *
+ * @param endpoint the endpoint
+ * @param pollingConsumer the pollingConsumer to release
+ */
+ public void releasePollingConsumer(Endpoint endpoint, PollingConsumer pollingConsumer) {
+ if (pollingConsumer instanceof ServicePoolAware) {
+ // release back to the pool
+ pool.release(endpoint, pollingConsumer);
+ } else {
+ boolean singleton = false;
+ if (pollingConsumer instanceof IsSingleton) {
+ singleton = ((IsSingleton) pollingConsumer).isSingleton();
+ }
+ if (!singleton) {
+ try {
+ // stop and shutdown non-singleton producers as we should not leak resources
+ ServiceHelper.stopAndShutdownService(pollingConsumer);
+ } catch (Exception ex) {
+ if (ex instanceof RuntimeCamelException) {
+ throw (RuntimeCamelException)ex;
+ } else {
+ throw new RuntimeCamelException(ex);
+ }
+ }
+ }
+ }
+ }
+
+ public PollingConsumer getConsumer(Endpoint endpoint) {
+ return doGetPollingConsumer(endpoint, true);
+ }
+
+ protected synchronized PollingConsumer doGetPollingConsumer(Endpoint endpoint, boolean pooled) {
String key = endpoint.getEndpointUri();
PollingConsumer answer = consumers.get(key);
+ if (pooled && answer == null) {
+ pool.acquire(endpoint);
+ }
+
if (answer == null) {
try {
answer = endpoint.createPollingConsumer();
@@ -82,41 +141,63 @@ public class ConsumerCache extends ServiceSupport {
} catch (Exception e) {
throw new FailedToCreateConsumerException(endpoint, e);
}
-
- boolean singleton = true;
- if (answer instanceof IsSingleton) {
- singleton = ((IsSingleton) answer).isSingleton();
- }
-
- if (singleton) {
- LOG.debug("Adding to consumer cache with key: {} for consumer: {}", endpoint, answer);
- consumers.put(key, answer);
+ if (pooled && answer instanceof ServicePoolAware) {
+ LOG.debug("Adding to producer service pool with key: {} for producer: {}", endpoint, answer);
+ answer = pool.addAndAcquire(endpoint, answer);
} else {
- LOG.debug("Consumer for endpoint: {} is not singleton and thus not added to consumer cache", key);
+ boolean singleton = false;
+ if (answer instanceof IsSingleton) {
+ singleton = ((IsSingleton) answer).isSingleton();
+ }
+ if (singleton) {
+ LOG.debug("Adding to consumer cache with key: {} for consumer: {}", endpoint, answer);
+ consumers.put(key, answer);
+ } else {
+ LOG.debug("Consumer for endpoint: {} is not singleton and thus not added to consumer cache", key);
+ }
}
}
return answer;
+
}
-
+
public Exchange receive(Endpoint endpoint) {
LOG.debug("<<<< {}", endpoint);
-
- PollingConsumer consumer = getConsumer(endpoint);
- return consumer.receive();
+ PollingConsumer consumer = null;
+ try {
+ consumer = acquirePollingConsumer(endpoint);
+ return consumer.receive();
+ } finally {
+ if (consumer != null) {
+ releasePollingConsumer(endpoint, consumer);
+ }
+ }
}
public Exchange receive(Endpoint endpoint, long timeout) {
LOG.debug("<<<< {}", endpoint);
-
- PollingConsumer consumer = getConsumer(endpoint);
- return consumer.receive(timeout);
+ PollingConsumer consumer = null;
+ try {
+ consumer = acquirePollingConsumer(endpoint);
+ return consumer.receive(timeout);
+ } finally {
+ if (consumer != null) {
+ releasePollingConsumer(endpoint, consumer);
+ }
+ }
}
public Exchange receiveNoWait(Endpoint endpoint) {
LOG.debug("<<<< {}", endpoint);
-
- PollingConsumer consumer = getConsumer(endpoint);
- return consumer.receiveNoWait();
+ PollingConsumer consumer = null;
+ try {
+ consumer = doGetPollingConsumer(endpoint, true);
+ return consumer.receiveNoWait();
+ } finally {
+ if (consumer != null) {
+ releasePollingConsumer(endpoint, consumer);
+ }
+ }
}
public CamelContext getCamelContext() {
@@ -237,6 +318,8 @@ public class ConsumerCache extends ServiceSupport {
// when stopping we intend to shutdown
ServiceHelper.stopAndShutdownServices(consumers.values());
consumers.clear();
+ // we need to stop the pool service here
+ ServiceHelper.stopService(pool);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
index 93dfbfc..f1043cb 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.impl;
+import static org.apache.camel.util.StringQuoteHelper.doubleQuote;
+
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
@@ -37,6 +39,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+
import javax.naming.Context;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.Unmarshaller;
@@ -53,6 +56,7 @@ import org.apache.camel.IsSingleton;
import org.apache.camel.MultipleConsumersSupport;
import org.apache.camel.NoFactoryAvailableException;
import org.apache.camel.NoSuchEndpointException;
+import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.ProducerTemplate;
@@ -148,8 +152,6 @@ import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.camel.util.StringQuoteHelper.doubleQuote;
-
/**
* Represents the context used to configure routes and the policies to use.
*
@@ -223,6 +225,7 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
// we use a capacity of 100 per endpoint, so for the same endpoint we have at most 100 producers in the pool
// so if we have 6 endpoints in the pool, we can have 6 x 100 producers in total
private ServicePool<Endpoint, Producer> producerServicePool = new SharedProducerServicePool(100);
+ private ServicePool<Endpoint, PollingConsumer> pollingConsumerServicePool = new SharedPollingConsumerServicePool(100);
private NodeIdFactory nodeIdFactory = new DefaultNodeIdFactory();
private ProcessorFactory processorFactory;
private InterceptStrategy defaultTracer;
@@ -1817,6 +1820,14 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
public ServicePool<Endpoint, Producer> getProducerServicePool() {
return producerServicePool;
}
+
+ public ServicePool<Endpoint, PollingConsumer> getPollingConsumerServicePool() {
+ return pollingConsumerServicePool;
+ }
+
+ public void setPollingConsumerServicePool(ServicePool<Endpoint, PollingConsumer> pollingConsumerServicePool) {
+ this.pollingConsumerServicePool = pollingConsumerServicePool;
+ }
public UnitOfWorkFactory getUnitOfWorkFactory() {
return unitOfWorkFactory;
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerServicePool.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerServicePool.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerServicePool.java
new file mode 100644
index 0000000..76f389b
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerServicePool.java
@@ -0,0 +1,39 @@
+/**
+ * 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.Endpoint;
+import org.apache.camel.PollingConsumer;
+
+/**
+ * A service pool for {@link PollingConsumer}.
+ * <p/>
+ * For instance camel-jms leverages this to allow a pool of pollingConsumers so we
+ * can support concurrent pollingConsumer in a thread safe manner.
+ *
+ */
+
+public class DefaultPollingConsumerServicePool extends DefaultServicePool<Endpoint, PollingConsumer> {
+
+ public DefaultPollingConsumerServicePool() {
+ }
+
+ public DefaultPollingConsumerServicePool(int capacity) {
+ super(capacity);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java b/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
index 22efbec..9d944ad 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.IsSingleton;
import org.apache.camel.PollingConsumerPollingStrategy;
import org.apache.camel.Processor;
import org.apache.camel.spi.ExceptionHandler;
@@ -39,7 +40,7 @@ import org.slf4j.LoggerFactory;
*
* @version
*/
-public class EventDrivenPollingConsumer extends PollingConsumerSupport implements Processor {
+public class EventDrivenPollingConsumer extends PollingConsumerSupport implements Processor, IsSingleton {
private static final Logger LOG = LoggerFactory.getLogger(EventDrivenPollingConsumer.class);
private final BlockingQueue<Exchange> queue;
private ExceptionHandler interruptedExceptionHandler;
@@ -203,4 +204,10 @@ public class EventDrivenPollingConsumer extends PollingConsumerSupport implement
ServiceHelper.stopAndShutdownService(consumer);
queue.clear();
}
+
+ @Override
+ // As the consumer could take the messages at once, so we cannot release the consumer
+ public boolean isSingleton() {
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/impl/ProcessorPollingConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ProcessorPollingConsumer.java b/camel-core/src/main/java/org/apache/camel/impl/ProcessorPollingConsumer.java
index 4994834..d3ec52b 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ProcessorPollingConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ProcessorPollingConsumer.java
@@ -20,6 +20,7 @@ import java.util.concurrent.RejectedExecutionException;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.IsSingleton;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeExchangeException;
import org.apache.camel.util.ServiceHelper;
@@ -31,7 +32,7 @@ import org.apache.camel.util.ServiceHelper;
*
* @version
*/
-public class ProcessorPollingConsumer extends PollingConsumerSupport {
+public class ProcessorPollingConsumer extends PollingConsumerSupport implements IsSingleton {
private final Processor processor;
public ProcessorPollingConsumer(Endpoint endpoint, Processor processor) {
@@ -69,4 +70,10 @@ public class ProcessorPollingConsumer extends PollingConsumerSupport {
public Exchange receive(long timeout) {
return receive();
}
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/main/java/org/apache/camel/impl/SharedPollingConsumerServicePool.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/SharedPollingConsumerServicePool.java b/camel-core/src/main/java/org/apache/camel/impl/SharedPollingConsumerServicePool.java
new file mode 100644
index 0000000..7478a14
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/impl/SharedPollingConsumerServicePool.java
@@ -0,0 +1,49 @@
+/**
+ * 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.ShutdownableService;
+import org.apache.camel.StaticService;
+
+/**
+ * A shared {@link org.apache.camel.impl.DefaultPollingConsumerServicePool} which is used by
+ * {@link org.apache.camel.CamelContext} by default.
+ *
+ * @version
+ */
+public class SharedPollingConsumerServicePool extends DefaultPollingConsumerServicePool implements ShutdownableService, StaticService {
+
+ public SharedPollingConsumerServicePool() {
+ }
+
+ public SharedPollingConsumerServicePool(int capacity) {
+ super(capacity);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // only let CamelContext stop it since its shared and should
+ // only be stopped when CamelContext stops
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ // now we are shutting down then stop it, which properly stops the pool
+ super.doStop();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java b/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
index fb62e3f..eec0687 100644
--- a/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
+++ b/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.file;
+import java.io.File;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
@@ -28,11 +29,12 @@ public class FileConsumeTemplateTest extends ContextTestSupport {
protected void setUp() throws Exception {
deleteDirectory("target/template");
super.setUp();
- template.sendBodyAndHeader("file://target/template", "Hello World", Exchange.FILE_NAME, "hello.txt");
- template.sendBodyAndHeader("file://target/template/", "Bye World", Exchange.FILE_NAME, "bye.txt");
}
public void testConsumeFileWithTemplate() throws Exception {
+ template.sendBodyAndHeader("file://target/template", "Hello World", Exchange.FILE_NAME, "hello.txt");
+ template.sendBodyAndHeader("file://target/template", "Bye World", Exchange.FILE_NAME, "bye.txt");
+
Exchange out = consumer.receive("file://target/template?sortBy=file:name");
assertNotNull(out);
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateTest.java b/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateTest.java
index 6101125..2382fe3 100644
--- a/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateTest.java
+++ b/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateTest.java
@@ -39,12 +39,15 @@ public class FileConsumerTemplateTest extends ContextTestSupport {
public void testFileConsumerTemplate() throws Exception {
template.sendBodyAndHeader("file:target/consumer", "Hello World", Exchange.FILE_NAME, "hello.txt");
+ // file should exist
+ File file = new File("target/consumer/hello.txt");
+ assertTrue("File should exist " + file, file.exists());
+
String body = consumer.receiveBody("file:target/consumer?delete=true", 5000, String.class);
assertEquals("Hello World", body);
// file should be deleted
- File file = new File("target/consumer/hello.txt");
assertFalse("File should be deleted " + file, file.exists());
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/components/camel-http/src/main/java/org/apache/camel/component/http/HttpPollingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-http/src/main/java/org/apache/camel/component/http/HttpPollingConsumer.java b/components/camel-http/src/main/java/org/apache/camel/component/http/HttpPollingConsumer.java
index 7180099..63603ea 100644
--- a/components/camel-http/src/main/java/org/apache/camel/component/http/HttpPollingConsumer.java
+++ b/components/camel-http/src/main/java/org/apache/camel/component/http/HttpPollingConsumer.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.ServicePoolAware;
import org.apache.camel.component.http.helper.HttpHelper;
import org.apache.camel.impl.PollingConsumerSupport;
import org.apache.camel.spi.HeaderFilterStrategy;
@@ -34,7 +35,7 @@ import org.apache.commons.httpclient.methods.GetMethod;
*
* @version
*/
-public class HttpPollingConsumer extends PollingConsumerSupport {
+public class HttpPollingConsumer extends PollingConsumerSupport implements ServicePoolAware {
private final HttpEndpoint endpoint;
private HttpClient httpClient;
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/components/camel-http4/src/main/java/org/apache/camel/component/http4/HttpPollingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-http4/src/main/java/org/apache/camel/component/http4/HttpPollingConsumer.java b/components/camel-http4/src/main/java/org/apache/camel/component/http4/HttpPollingConsumer.java
index 3cfb8ec..758c1c9 100644
--- a/components/camel-http4/src/main/java/org/apache/camel/component/http4/HttpPollingConsumer.java
+++ b/components/camel-http4/src/main/java/org/apache/camel/component/http4/HttpPollingConsumer.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.ServicePoolAware;
import org.apache.camel.component.http4.helper.HttpHelper;
import org.apache.camel.impl.PollingConsumerSupport;
import org.apache.camel.spi.HeaderFilterStrategy;
@@ -39,7 +40,7 @@ import org.apache.http.util.EntityUtils;
*
* @version
*/
-public class HttpPollingConsumer extends PollingConsumerSupport {
+public class HttpPollingConsumer extends PollingConsumerSupport implements ServicePoolAware {
private final HttpEndpoint endpoint;
private HttpClient httpClient;
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
index 17d7ae8..705042d 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
@@ -261,8 +261,7 @@ public class JmsEndpoint extends DefaultEndpoint implements HeaderFilterStrategy
@Override
public PollingConsumer createPollingConsumer() throws Exception {
- JmsOperations template = createInOnlyTemplate();
- JmsPollingConsumer answer = new JmsPollingConsumer(this, template);
+ JmsPollingConsumer answer = new JmsPollingConsumer(this);
configurePollingConsumer(answer);
return answer;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/27e13d91/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
index f304319..90bfb79 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsPollingConsumer.java
@@ -19,15 +19,16 @@ package org.apache.camel.component.jms;
import javax.jms.Message;
import org.apache.camel.Exchange;
+import org.apache.camel.ServicePoolAware;
import org.apache.camel.impl.PollingConsumerSupport;
import org.apache.camel.util.ObjectHelper;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.JmsTemplate;
/**
- * @version
+ * As the template is created per JmsPollingConsumer, we need to put JmsPollingConsumer into this endpoint
*/
-public class JmsPollingConsumer extends PollingConsumerSupport {
+public class JmsPollingConsumer extends PollingConsumerSupport implements ServicePoolAware {
private JmsOperations template;
private JmsEndpoint jmsEndpoint;
@@ -36,6 +37,10 @@ public class JmsPollingConsumer extends PollingConsumerSupport {
this.jmsEndpoint = endpoint;
this.template = template;
}
+
+ public JmsPollingConsumer(JmsEndpoint endpoint) {
+ this(endpoint, endpoint.createInOnlyTemplate());
+ }
@Override
public JmsEndpoint getEndpoint() {
@@ -69,6 +74,7 @@ public class JmsPollingConsumer extends PollingConsumerSupport {
}
protected void doStop() throws Exception {
+
}
protected void setReceiveTimeout(long timeout) {