Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

On destruction, ConcurrentHandler calls shutdownNow() on its Executor rather than shutdown().

Browse Source
This commit is contained in:
Mark Fisher
2008-04-16 19:50:32 +00:00
parent a0072b3e66
commit e9673586c4
2 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java
View File

@@ -68,7 +68,7 @@ public class ConcurrentHandler implements MessageHandler, DisposableBean {
}
public void destroy() {
this.executor.shutdown();
this.executor.shutdownNow();
}
public Message<?> handle(Message<?> message) {

24
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
View File

@@ -184,21 +184,19 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
if (this.handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.handler).setChannelRegistry(this.channelRegistry);
}
if (this.concurrencyPolicy != null || this.handler instanceof ConcurrentHandler) {
if (!(this.handler instanceof ConcurrentHandler)) {
int capacity = concurrencyPolicy.getQueueCapacity();
BlockingQueue<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() :
new ArrayBlockingQueue<Runnable>(capacity);
ExecutorService executor = new ThreadPoolExecutor(
concurrencyPolicy.getCoreSize(), concurrencyPolicy.getMaxSize(),
concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue);
this.handler = new ConcurrentHandler(this.handler, executor);
}
ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler;
if (this.concurrencyPolicy != null && !(this.handler instanceof ConcurrentHandler)) {
int capacity = this.concurrencyPolicy.getQueueCapacity();
BlockingQueue<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() : new ArrayBlockingQueue<Runnable>(capacity);
ExecutorService executor = new ThreadPoolExecutor(
this.concurrencyPolicy.getCoreSize(), this.concurrencyPolicy.getMaxSize(),
this.concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue);
this.handler = new ConcurrentHandler(this.handler, executor);
}
if (this.handler instanceof ConcurrentHandler) {
if (this.errorHandler != null) {
concurrentHandler.setErrorHandler(this.errorHandler);
((ConcurrentHandler) this.handler).setErrorHandler(this.errorHandler);
}
concurrentHandler.setReplyHandler(this.replyHandler);
((ConcurrentHandler) this.handler).setReplyHandler(this.replyHandler);
}
this.initialized = true;
}