From e9673586c4269bf95c9188c572f6f05b83b6d9ad Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 16 Apr 2008 19:50:32 +0000 Subject: [PATCH] On destruction, ConcurrentHandler calls shutdownNow() on its Executor rather than shutdown(). --- .../endpoint/ConcurrentHandler.java | 2 +- .../endpoint/DefaultMessageEndpoint.java | 24 +++++++++---------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java index 72c5e1e1ac..f01f56b7bd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java @@ -68,7 +68,7 @@ public class ConcurrentHandler implements MessageHandler, DisposableBean { } public void destroy() { - this.executor.shutdown(); + this.executor.shutdownNow(); } public Message handle(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 68de223104..ee186e4d8c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -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 queue = (capacity < 1) ? new SynchronousQueue() : - new ArrayBlockingQueue(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 queue = (capacity < 1) ? new SynchronousQueue() : new ArrayBlockingQueue(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; }