Author: davsclaus
Date: Sat Mar 13 10:54:45 2010
New Revision: 922524
URL: http://svn.apache.org/viewvc?rev=922524&view=rev
Log:
CAMEL-1588: EIP using thread pool is not created from model as EIP processor
mandates the thread pool. Goal is better knowledge which routes uses which
thread pools.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ToDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/BatchProcessor.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
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationCollection.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/DefaultAggregationCollection.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/PredicateAggregationCollection.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
Sat Mar 13 10:54:45 2010
@@ -90,6 +90,12 @@ public class MethodInfo {
recipientList.setExecutorService(executor);
}
+ if (annotation.parallelProcessoing() &&
recipientList.getExecutorService() == null) {
+ // we are running in parallel so create a cached thread pool
which grows/shrinks automatic
+ ExecutorService executor =
camelContext.getExecutorServiceStrategy().newCachedThreadPool(this,
"@RecipientList");
+ recipientList.setExecutorService(executor);
+ }
+
if (ObjectHelper.isNotEmpty(annotation.strategyRef())) {
AggregationStrategy strategy =
CamelContextHelper.mandatoryLookup(camelContext, annotation.strategyRef(),
AggregationStrategy.class);
recipientList.setAggregationStrategy(strategy);
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
Sat Mar 13 10:54:45 2010
@@ -149,18 +149,26 @@ public class AggregateDefinition extends
Expression correlation =
getExpression().createExpression(routeContext);
AggregationStrategy strategy = createAggregationStrategy(routeContext);
- AggregateProcessor answer = new
AggregateProcessor(routeContext.getCamelContext(), processor, correlation,
strategy);
-
+ // executor service is mandatory for the Aggregator
executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- answer.setExecutorService(executorService);
- if (isParallelProcessing() != null) {
- answer.setParallelProcessing(isParallelProcessing());
+ if (executorService == null) {
+ if (isParallelProcessing()) {
+ // we are running in parallel so create a cached thread pool
which grows/shrinks automatic
+ executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"Aggregator");
+ } else {
+ // use a single threaded if we are not running in parallel
+ executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newSingleThreadExecutor(this,
"Aggregator");
+ }
}
+ AggregateProcessor answer = new
AggregateProcessor(routeContext.getCamelContext(), processor, correlation,
strategy, executorService);
+
AggregationRepository<Object> repository =
createAggregationRepository(routeContext);
if (repository != null) {
answer.setAggregationRepository(repository);
}
+ // set other options
+ answer.setParallelProcessing(isParallelProcessing());
if (getCompletionPredicate() != null) {
Predicate predicate =
getCompletionPredicate().createPredicate(routeContext);
answer.setCompletionPredicate(predicate);
@@ -301,11 +309,11 @@ public class AggregateDefinition extends
this.executorService = executorService;
}
- public Boolean isParallelProcessing() {
- return parallelProcessing;
+ public boolean isParallelProcessing() {
+ return parallelProcessing != null && parallelProcessing;
}
- public void setParallelProcessing(Boolean parallelProcessing) {
+ public void setParallelProcessing(boolean parallelProcessing) {
this.parallelProcessing = parallelProcessing;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
Sat Mar 13 10:54:45 2010
@@ -147,7 +147,12 @@ public class MulticastDefinition extends
// default to use latest aggregation strategy
aggregationStrategy = new UseLatestAggregationStrategy();
}
+
executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
+ if (isParallelProcessing() && executorService == null) {
+ // we are running in parallel so create a cached thread pool which
grows/shrinks automatic
+ executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"Multicast");
+ }
return new MulticastProcessor(routeContext.getCamelContext(), list,
aggregationStrategy, isParallelProcessing(),
executorService, isStreaming(),
isStopOnException());
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
Sat Mar 13 10:54:45 2010
@@ -97,10 +97,12 @@ public class OnCompletionDefinition exte
when = onWhen.getExpression().createPredicate(routeContext);
}
- OnCompletionProcessor answer = new
OnCompletionProcessor(routeContext.getCamelContext(), childProcessor,
-
onCompleteOnly, onFailureOnly, when);
executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- answer.setExecutorService(executorService);
+ if (executorService == null) {
+ executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"OnCompletion");
+ }
+ OnCompletionProcessor answer = new
OnCompletionProcessor(routeContext.getCamelContext(), childProcessor,
executorService,
+
onCompleteOnly, onFailureOnly, when);
return answer;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
Sat Mar 13 10:54:45 2010
@@ -89,18 +89,15 @@ public class RecipientListDefinition ext
} else {
answer = new RecipientList(expression);
}
-
- if (parallelProcessing != null) {
- answer.setParallelProcessing(isParallelProcessing());
- }
+ answer.setAggregationStrategy(createAggregationStrategy(routeContext));
+ answer.setParallelProcessing(isParallelProcessing());
if (stopOnException != null) {
answer.setStopOnException(isStopOnException());
}
-
- answer.setAggregationStrategy(createAggregationStrategy(routeContext));
+
executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- if (executorService == null) {
- // fallback to create a new executor
+ if (isParallelProcessing() && executorService == null) {
+ // we are running in parallel so create a cached thread pool which
grows/shrinks automatic
executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"RecipientList");
}
answer.setExecutorService(executorService);
@@ -201,11 +198,11 @@ public class RecipientListDefinition ext
this.delimiter = delimiter;
}
- public Boolean isParallelProcessing() {
- return parallelProcessing;
+ public boolean isParallelProcessing() {
+ return parallelProcessing != null && parallelProcessing;
}
- public void setParallelProcessing(Boolean parallelProcessing) {
+ public void setParallelProcessing(boolean parallelProcessing) {
this.parallelProcessing = parallelProcessing;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
Sat Mar 13 10:54:45 2010
@@ -89,8 +89,8 @@ public class SplitDefinition extends Exp
aggregationStrategy = createAggregationStrategy(routeContext);
executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- if (executorService == null) {
- // fallback to create a new executor
+ if (isParallelProcessing() && executorService == null) {
+ // we are running in parallel so create a cached thread pool which
grows/shrinks automatic
executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"Split");
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ToDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ToDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ToDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ToDefinition.java
Sat Mar 13 10:54:45 2010
@@ -90,13 +90,6 @@ public class ToDefinition extends SendDe
// this code below is only for creating when async is enabled
// ----------------------------------------------------------
- executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- if (executorService == null && poolSize != null) {
- // crete a new based on the other options
- executorService =
routeContext.getCamelContext().getExecutorServiceStrategy()
- .newThreadPool(this, "ToAsync[" + getLabel() +
"]", poolSize, poolSize);
- }
-
// create the child processor which is the async route
Processor childProcessor = routeContext.createProcessor(this);
@@ -106,7 +99,11 @@ public class ToDefinition extends SendDe
// create async processor
Endpoint endpoint = resolveEndpoint(routeContext);
+ // TODO: rework to have configured executor service in
SendAsyncProcessor being handled in stop/start scenario
+
SendAsyncProcessor async = new SendAsyncProcessor(endpoint,
getPattern(), uow);
+
+ executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
if (executorService != null) {
async.setExecutorService(executorService);
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
Sat Mar 13 10:54:45 2010
@@ -66,7 +66,12 @@ public class WireTapDefinition extends S
@Override
public Processor createProcessor(RouteContext routeContext) throws
Exception {
Endpoint endpoint = resolveEndpoint(routeContext);
- WireTapProcessor answer = new WireTapProcessor(endpoint, getPattern());
+
+ executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
+ if (executorService == null) {
+ executorService =
routeContext.getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
"WireTap");
+ }
+ WireTapProcessor answer = new WireTapProcessor(endpoint, getPattern(),
executorService);
if (newExchangeProcessorRef != null) {
newExchangeProcessor =
routeContext.lookup(newExchangeProcessorRef, Processor.class);
@@ -76,8 +81,6 @@ public class WireTapDefinition extends S
answer.setNewExchangeExpression(newExchangeExpression.createExpression(routeContext));
}
- executorService =
ExecutorServiceHelper.getConfiguredExecutorService(routeContext, this);
- answer.setExecutorService(executorService);
return answer;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/BatchProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/BatchProcessor.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/BatchProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/BatchProcessor.java
Sat Mar 13 10:54:45 2010
@@ -45,7 +45,7 @@ import org.apache.commons.logging.LogFac
* A base class for any kind of {...@link Processor} which implements some
kind of batch processing.
*
* @version $Revision$
- * @deprecated will be removed in Camel 2.4
+ * @deprecated may be removed in the future
*/
@Deprecated
public class BatchProcessor extends ServiceSupport implements Processor,
Navigate<Processor> {
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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -25,7 +25,6 @@ import java.util.concurrent.ExecutionExc
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -105,7 +104,7 @@ public class MulticastProcessor extends
private final boolean isParallelProcessing;
private final boolean streaming;
private final boolean stopOnException;
- private ExecutorService executorService;
+ private final ExecutorService executorService;
public MulticastProcessor(CamelContext camelContext, Collection<Processor>
processors) {
this(camelContext, processors, null);
@@ -126,10 +125,6 @@ public class MulticastProcessor extends
this.executorService = executorService;
this.streaming = streaming;
this.stopOnException = stopOnException;
-
- if (isParallelProcessing() && getExecutorService() == null) {
- this.executorService =
camelContext.getExecutorServiceStrategy().newCachedThreadPool(this,
"Multicast");
- }
}
@Override
@@ -149,6 +144,8 @@ public class MulticastProcessor extends
// so use try .. catch to cater for this
try {
if (isParallelProcessing()) {
+ // ensure an executor is set when running in parallel
+ ObjectHelper.notNull(executorService, "executorService", this);
doProcessParallel(result, pairs, isStreaming());
} else {
doProcessSequential(result, pairs);
@@ -356,6 +353,9 @@ public class MulticastProcessor extends
}
protected void doStart() throws Exception {
+ if (isParallelProcessing() && executorService == null) {
+ throw new IllegalArgumentException("ParallelProcessing is enabled
but ExecutorService has not been set");
+ }
ServiceHelper.startServices(processors);
}
@@ -368,7 +368,6 @@ public class MulticastProcessor extends
// only shutdown thread pool on shutdown
if (executorService != null) {
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
- executorService = null;
}
}
@@ -415,14 +414,6 @@ public class MulticastProcessor extends
return isParallelProcessing;
}
- public ExecutorService getExecutorService() {
- return executorService;
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
public List<Processor> next() {
if (!hasNext()) {
return 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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -41,17 +41,20 @@ public class OnCompletionProcessor exten
private static final transient Log LOG =
LogFactory.getLog(OnCompletionProcessor.class);
private final CamelContext camelContext;
private final Processor processor;
- private ExecutorService executorService;
+ private final ExecutorService executorService;
private boolean onCompleteOnly;
private boolean onFailureOnly;
private Predicate onWhen;
- public OnCompletionProcessor(CamelContext camelContext, Processor
processor, boolean onCompleteOnly, boolean onFailureOnly, Predicate onWhen) {
+ public OnCompletionProcessor(CamelContext camelContext, Processor
processor, ExecutorService executorService,
+ boolean onCompleteOnly, boolean
onFailureOnly, Predicate onWhen) {
notNull(camelContext, "camelContext");
notNull(processor, "processor");
+ notNull(executorService, "executorService");
this.camelContext = camelContext;
// wrap processor in UnitOfWork so what we send out runs in a UoW
this.processor = new UnitOfWorkProcessor(processor);
+ this.executorService = executorService;
this.onCompleteOnly = onCompleteOnly;
this.onFailureOnly = onFailureOnly;
this.onWhen = onWhen;
@@ -71,8 +74,6 @@ public class OnCompletionProcessor exten
// only shutdown thread pool on shutdown
if (executorService != null) {
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
- // must null it so we can restart
- executorService = null;
}
}
@@ -97,7 +98,7 @@ public class OnCompletionProcessor exten
// must use a copy as we dont want it to cause side effects of
the original exchange
final Exchange copy = prepareExchange(exchange);
- getExecutorService().submit(new Callable<Exchange>() {
+ executorService.submit(new Callable<Exchange>() {
public Exchange call() throws Exception {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing onComplete: " + copy);
@@ -124,7 +125,7 @@ public class OnCompletionProcessor exten
// the caused exception is stored as a property
(Exchange.EXCEPTION_CAUGHT) on the exchange
copy.setException(null);
- getExecutorService().submit(new Callable<Exchange>() {
+ executorService.submit(new Callable<Exchange>() {
public Exchange call() throws Exception {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing onFailure: " + copy);
@@ -179,21 +180,6 @@ public class OnCompletionProcessor exten
return copy;
}
- public ExecutorService getExecutorService() {
- if (executorService == null) {
- executorService = createExecutorService();
- }
- return executorService;
- }
-
- protected ExecutorService createExecutorService() {
- return
camelContext.getExecutorServiceStrategy().newCachedThreadPool(this,
this.toString());
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
@Override
public String toString() {
return "OnCompletionProcessor[" + processor + "]";
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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -44,13 +44,14 @@ 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;
- protected ExecutorService producerExecutorService;
- protected int poolSize = DEFAULT_THREADPOOL_SIZE;
- protected ExceptionHandler exceptionHandler;
+ private final CamelContext camelContext;
+ private final Processor target;
+ private final BlockingQueue<Exchange> completedTasks = new
LinkedBlockingQueue<Exchange>();
+ private ExecutorService executorService;
+ private ExecutorService producerExecutorService;
+ private int poolSize = DEFAULT_THREADPOOL_SIZE;
+ private ExceptionHandler exceptionHandler;
+
public SendAsyncProcessor(Endpoint destination, Processor target) {
super(destination);
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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -23,7 +23,6 @@ import java.util.concurrent.Future;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.ShutdownableService;
import org.apache.camel.WaitForTaskToComplete;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
@@ -42,7 +41,7 @@ import org.apache.camel.util.ObjectHelpe
public class ThreadsProcessor extends DelegateProcessor implements Processor {
protected final CamelContext camelContext;
- protected ExecutorService executorService;
+ protected final ExecutorService executorService;
protected WaitForTaskToComplete waitForTaskToComplete;
public ThreadsProcessor(CamelContext camelContext, Processor output,
ExecutorService executorService, WaitForTaskToComplete waitForTaskToComplete) {
@@ -69,7 +68,7 @@ public class ThreadsProcessor extends De
Callable<Exchange> task = createTask(output, copy);
// submit the task
- Future<Exchange> future = getExecutorService().submit(task);
+ Future<Exchange> future = executorService.submit(task);
// compute if we should wait for task to complete or not
WaitForTaskToComplete wait = waitForTaskToComplete;
@@ -102,20 +101,12 @@ public class ThreadsProcessor extends De
};
}
- public ExecutorService getExecutorService() {
- if (executorService == null) {
- executorService =
camelContext.getExecutorServiceStrategy().newCachedThreadPool(this, "Threads");
- }
- return executorService;
- }
-
@Override
protected void doShutdown() throws Exception {
super.doShutdown();
// only shutdown thread pool on shutdown
if (executorService != null) {
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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -39,20 +39,24 @@ import org.apache.camel.util.ObjectHelpe
public class WireTapProcessor extends SendProcessor {
private final CamelContext camelContext;
- private ExecutorService executorService;
+ private final ExecutorService executorService;
// expression or processor used for populating a new exchange to send
// as opposed to traditional wiretap that sends a copy of the original
exchange
private Expression newExchangeExpression;
private Processor newExchangeProcessor;
- public WireTapProcessor(Endpoint destination) {
+ public WireTapProcessor(Endpoint destination, ExecutorService
executorService) {
super(destination);
+ ObjectHelper.notNull(executorService, "executorService");
+ this.executorService = executorService;
this.camelContext = destination.getCamelContext();
}
- public WireTapProcessor(Endpoint destination, ExchangePattern pattern) {
+ public WireTapProcessor(Endpoint destination, ExchangePattern pattern,
ExecutorService executorService) {
super(destination, pattern);
+ ObjectHelper.notNull(executorService, "executorService");
+ this.executorService = executorService;
this.camelContext = destination.getCamelContext();
}
@@ -62,8 +66,6 @@ public class WireTapProcessor extends Se
// only shutdown thread pool on shutdown
if (executorService != null) {
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
- // must null it so we can restart
- executorService = null;
}
}
@@ -90,13 +92,14 @@ public class WireTapProcessor extends Se
/**
* Wiretaps the exchange.
*
+ * @param producer the producer
* @param exchange the exchange to wire tap
*/
protected void processWireTap(final Producer producer, final Exchange
exchange) {
// use submit instead of execute to force it to use a new thread,
execute might
// decide to use current thread, so we must submit a new task
// as we don't care for the response we dont hold the future object
and wait for the result
- getExecutorService().submit(new Callable<Exchange>() {
+ executorService.submit(new Callable<Exchange>() {
public Exchange call() throws Exception {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing wiretap: " + exchange);
@@ -152,21 +155,6 @@ public class WireTapProcessor extends Se
return answer;
}
- public ExecutorService getExecutorService() {
- if (executorService == null) {
- executorService = createExecutorService();
- }
- return executorService;
- }
-
- protected ExecutorService createExecutorService() {
- return
getDestination().getCamelContext().getExecutorServiceStrategy().newCachedThreadPool(this,
this.toString());
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
public Processor getNewExchangeProcessor() {
return newExchangeProcessor;
}
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=922524&r1=922523&r2=922524&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 10:54:45 2010
@@ -70,8 +70,8 @@ public class AggregateProcessor extends
private final Processor processor;
private final AggregationStrategy aggregationStrategy;
private final Expression correlationExpression;
+ private final ExecutorService executorService;
private TimeoutMap<Object, Exchange> timeoutMap;
- private ExecutorService executorService;
private ExceptionHandler exceptionHandler;
private AggregationRepository<Object> aggregationRepository = new
MemoryAggregationRepository();
private Map<Object, Object> closedCorrelationKeys;
@@ -92,15 +92,18 @@ public class AggregateProcessor extends
private AtomicInteger batchConsumerCounter = new AtomicInteger();
public AggregateProcessor(CamelContext camelContext, Processor processor,
- Expression correlationExpression,
AggregationStrategy aggregationStrategy) {
+ Expression correlationExpression,
AggregationStrategy aggregationStrategy,
+ ExecutorService executorService) {
ObjectHelper.notNull(camelContext, "camelContext");
ObjectHelper.notNull(processor, "processor");
ObjectHelper.notNull(correlationExpression, "correlationExpression");
ObjectHelper.notNull(aggregationStrategy, "aggregationStrategy");
+ ObjectHelper.notNull(executorService, "executorService");
this.camelContext = camelContext;
this.processor = processor;
this.correlationExpression = correlationExpression;
this.aggregationStrategy = aggregationStrategy;
+ this.executorService = executorService;
}
@Override
@@ -323,14 +326,6 @@ public class AggregateProcessor extends
});
}
- public ExecutorService getExecutorService() {
- return executorService;
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
public Predicate getCompletionPredicate() {
return completionPredicate;
}
@@ -472,16 +467,6 @@ public class AggregateProcessor extends
ServiceHelper.startService(aggregationRepository);
- if (executorService == null) {
- if (isParallelProcessing()) {
- // we are running in parallel so create a cached thread pool
which grows/shrinks automatic
- executorService =
camelContext.getExecutorServiceStrategy().newCachedThreadPool(this,
"Aggregator");
- } else {
- // use a single threaded if we are not running in parallel
- executorService =
camelContext.getExecutorServiceStrategy().newSingleThreadExecutor(this,
"Aggregator");
- }
- }
-
// start timeout service if its in use
if (getCompletionTimeout() > 0 || getCompletionTimeoutExpression() !=
null) {
ScheduledExecutorService scheduler =
camelContext.getExecutorServiceStrategy().newScheduledThreadPool(this,
"AggregateTimeoutChecker", 1);
@@ -505,10 +490,7 @@ public class AggregateProcessor extends
@Override
protected void doShutdown() throws Exception {
// only shutdown thread pool when we are shutting down
- if (executorService != null) {
-
camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
- executorService = null;
- }
+ camelContext.getExecutorServiceStrategy().shutdownNow(executorService);
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationCollection.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationCollection.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationCollection.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationCollection.java
Sat Mar 13 10:54:45 2010
@@ -30,6 +30,7 @@ import org.apache.camel.Expression;
* Other implementations could for instance just add all exchanges as a batch.
*
* @version $Revision$
+ * @deprecated will be removed in Camel 2.4
*/
@Deprecated
public interface AggregationCollection extends Collection<Exchange> {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/DefaultAggregationCollection.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/DefaultAggregationCollection.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/DefaultAggregationCollection.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/DefaultAggregationCollection.java
Sat Mar 13 10:54:45 2010
@@ -35,6 +35,7 @@ import org.apache.commons.logging.LogFac
* correlation key.
*
* @version $Revision$
+ * @deprecated will be removed in Camel 2.4
*/
@Deprecated
public class DefaultAggregationCollection extends AbstractCollection<Exchange>
implements AggregationCollection {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/PredicateAggregationCollection.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/PredicateAggregationCollection.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/PredicateAggregationCollection.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/PredicateAggregationCollection.java
Sat Mar 13 10:54:45 2010
@@ -29,6 +29,7 @@ import org.apache.camel.Predicate;
* a particular correlation key
*
* @version $Revision$
+ * @deprecated will be removed in Camel 2.4
*/
@Deprecated
public class PredicateAggregationCollection extends
DefaultAggregationCollection {
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java?rev=922524&r1=922523&r2=922524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
Sat Mar 13 10:54:45 2010
@@ -16,6 +16,8 @@
*/
package org.apache.camel.processor.aggregator;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.CamelExchangeException;
@@ -37,11 +39,19 @@ import org.apache.camel.spi.ExceptionHan
*/
public class AggregateProcessorTest extends ContextTestSupport {
+ private ExecutorService executorService;
+
@Override
public boolean isUseRouteBuilder() {
return false;
}
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ executorService = Executors.newSingleThreadExecutor();
+ }
+
public void testAggregateProcessorCompletionPredicate() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("A+B+END");
@@ -52,7 +62,7 @@ public class AggregateProcessorTest exte
AggregationStrategy as = new BodyInAggregatingStrategy();
Predicate complete = body().contains("END");
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionPredicate(complete);
ap.setEagerCheckCompletion(false);
ap.start();
@@ -93,7 +103,7 @@ public class AggregateProcessorTest exte
AggregationStrategy as = new BodyInAggregatingStrategy();
Predicate complete = body().isEqualTo("END");
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionPredicate(complete);
ap.setEagerCheckCompletion(true);
ap.start();
@@ -141,7 +151,7 @@ public class AggregateProcessorTest exte
Expression corr = header("id");
AggregationStrategy as = new BodyInAggregatingStrategy();
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionSize(3);
ap.setEagerCheckCompletion(eager);
ap.start();
@@ -189,7 +199,7 @@ public class AggregateProcessorTest exte
Expression corr = header("id");
AggregationStrategy as = new BodyInAggregatingStrategy();
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionTimeout(3000);
ap.setEagerCheckCompletion(eager);
ap.start();
@@ -235,7 +245,7 @@ public class AggregateProcessorTest exte
AggregationStrategy as = new BodyInAggregatingStrategy();
Predicate complete = body().contains("END");
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionPredicate(complete);
ap.setIgnoreBadCorrelationKeys(true);
@@ -275,7 +285,7 @@ public class AggregateProcessorTest exte
AggregationStrategy as = new BodyInAggregatingStrategy();
Predicate complete = body().contains("END");
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionPredicate(complete);
ap.start();
@@ -322,7 +332,7 @@ public class AggregateProcessorTest exte
AggregationStrategy as = new BodyInAggregatingStrategy();
Predicate complete = body().contains("END");
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionPredicate(complete);
ap.setCloseCorrelationKeyOnCompletion(1000);
@@ -369,7 +379,7 @@ public class AggregateProcessorTest exte
Expression corr = header("id");
AggregationStrategy as = new BodyInAggregatingStrategy();
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setCompletionSize(100);
ap.setCompletionFromBatchConsumer(true);
@@ -464,7 +474,7 @@ public class AggregateProcessorTest exte
Expression corr = header("id");
AggregationStrategy as = new BodyInAggregatingStrategy();
- AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as);
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr,
as, executorService);
ap.setEagerCheckCompletion(true);
ap.setCompletionPredicate(body().isEqualTo("END"));
if (handler != null) {
