Author: davsclaus
Date: Sat Mar 13 06:00:21 2010
New Revision: 922485
URL: http://svn.apache.org/viewvc?rev=922485&view=rev
Log:
CAMEL-1588: Having ExecutorServiceStrategy to shutdown thread pools.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetConsumer.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetConsumer.java
Sat Mar 13 06:00:21 2010
@@ -18,6 +18,7 @@ package org.apache.camel.component.datas
import java.util.concurrent.ExecutorService;
+import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
@@ -29,6 +30,7 @@ import org.apache.camel.processor.Throug
* @version $Revision$
*/
public class DataSetConsumer extends DefaultConsumer {
+ private final CamelContext camelContext;
private DataSetEndpoint endpoint;
private Processor reporter;
private ExecutorService executorService;
@@ -36,6 +38,7 @@ public class DataSetConsumer extends Def
public DataSetConsumer(DataSetEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.endpoint = endpoint;
+ this.camelContext = endpoint.getCamelContext();
}
@Override
@@ -50,7 +53,7 @@ public class DataSetConsumer extends Def
sendMessages(0, preloadSize);
- executorService =
endpoint.getCamelContext().getExecutorServiceStrategy()
+ executorService = camelContext.getExecutorServiceStrategy()
.newSingleThreadExecutor(this, endpoint.getEndpointUri());
executorService.execute(new Runnable() {
@@ -74,7 +77,7 @@ public class DataSetConsumer extends Def
super.doStop();
if (executorService != null) {
- executorService.shutdown();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
executorService = null;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
Sat Mar 13 06:00:21 2010
@@ -190,7 +190,7 @@ public class SedaConsumer extends Servic
protected void doStop() throws Exception {
endpoint.onStopped(this);
// must shutdown executor on stop to avoid overhead of having them
running
- executor.shutdownNow();
+
endpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(executor);
executor = null;
if (multicast != null) {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
Sat Mar 13 06:00:21 2010
@@ -1215,7 +1215,7 @@ public class DefaultCamelContext extends
// shutdown management as the last one
shutdownServices(getManagementStrategy());
- LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") shutdown");
+ LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") is shutdown");
}
private void stopServices(Object service) throws Exception {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
Sat Mar 13 06:00:21 2010
@@ -419,17 +419,6 @@ public class DefaultProducerTemplate ext
return answer;
}
- protected void doStart() throws Exception {
- producerCache.start();
- }
-
- protected void doStop() throws Exception {
- producerCache.stop();
- if (executor != null) {
- executor.shutdown();
- }
- }
-
protected Object extractResultBody(Exchange result) {
return extractResultBody(result, null);
}
@@ -677,11 +666,23 @@ public class DefaultProducerTemplate ext
return executor.submit(task);
}
+ protected void doStart() throws Exception {
+ producerCache.start();
+ }
+
+ protected void doStop() throws Exception {
+ producerCache.stop();
+ if (executor != null) {
+ context.getExecutorServiceStrategy().shutdownNow(executor);
+ executor = null;
+ }
+ }
+
@Override
public void start() throws Exception {
super.start();
ServiceHelper.startService(producerCache);
- if (executor == null || executor.isShutdown()) {
+ if (executor == null) {
executor =
context.getExecutorServiceStrategy().newCachedThreadPool(this,
"ProducerTemplate");
}
}
@@ -691,7 +692,7 @@ public class DefaultProducerTemplate ext
super.stop();
ServiceHelper.stopService(producerCache);
if (executor != null) {
- executor.shutdown();
+ context.getExecutorServiceStrategy().shutdownNow(executor);
}
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
Sat Mar 13 06:00:21 2010
@@ -367,7 +367,7 @@ public class MulticastProcessor extends
protected void doShutdown() throws Exception {
// only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
executorService = null;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
Sat Mar 13 06:00:21 2010
@@ -70,7 +70,7 @@ public class OnCompletionProcessor exten
protected void doShutdown() throws Exception {
// only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
// must null it so we can restart
executorService = null;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
Sat Mar 13 06:00:21 2010
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -43,6 +44,7 @@ import org.apache.camel.util.ExchangeHel
public class SendAsyncProcessor extends SendProcessor implements Runnable,
Navigate<Processor> {
private static final int DEFAULT_THREADPOOL_SIZE = 10;
+ protected final CamelContext camelContext;
protected final Processor target;
protected final BlockingQueue<Exchange> completedTasks = new
LinkedBlockingQueue<Exchange>();
protected ExecutorService executorService;
@@ -53,11 +55,13 @@ public class SendAsyncProcessor extends
public SendAsyncProcessor(Endpoint destination, Processor target) {
super(destination);
this.target = target;
+ this.camelContext = destination.getCamelContext();
}
public SendAsyncProcessor(Endpoint destination, ExchangePattern pattern,
Processor target) {
super(destination, pattern);
this.target = target;
+ this.camelContext = destination.getCamelContext();
}
@Override
@@ -277,11 +281,11 @@ public class SendAsyncProcessor extends
// must shutdown thread pools on stop as we are consumers
if (producerExecutorService != null) {
- producerExecutorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(producerExecutorService);
producerExecutorService = null;
}
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
executorService = null;
}
completedTasks.clear();
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
Sat Mar 13 06:00:21 2010
@@ -114,7 +114,7 @@ public class ThreadsProcessor extends De
super.doShutdown();
// only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
executorService = null;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
Sat Mar 13 06:00:21 2010
@@ -19,6 +19,7 @@ package org.apache.camel.processor;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
+import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -37,6 +38,7 @@ import org.apache.camel.util.ObjectHelpe
*/
public class WireTapProcessor extends SendProcessor {
+ private final CamelContext camelContext;
private ExecutorService executorService;
// expression or processor used for populating a new exchange to send
@@ -46,10 +48,12 @@ public class WireTapProcessor extends Se
public WireTapProcessor(Endpoint destination) {
super(destination);
+ this.camelContext = destination.getCamelContext();
}
public WireTapProcessor(Endpoint destination, ExchangePattern pattern) {
super(destination, pattern);
+ this.camelContext = destination.getCamelContext();
}
@Override
@@ -57,7 +61,7 @@ public class WireTapProcessor extends Se
super.doShutdown();
// only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
// must null it so we can restart
executorService = null;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=922485&r1=922484&r2=922485&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Sat Mar 13 06:00:21 2010
@@ -506,7 +506,7 @@ public class AggregateProcessor extends
protected void doShutdown() throws Exception {
// only shutdown thread pool when we are shutting down
if (executorService != null) {
- executorService.shutdownNow();
+
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
executorService = null;
}
}
