diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index b196d2df61..e4c02aa189 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -214,7 +214,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware } private boolean send(final Message message, final int index) { - boolean result = false; if (index == 0) { for (EndpointInterceptor interceptor : interceptors) { if (!interceptor.preSend(message)) { @@ -223,21 +222,20 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware } } if (index == interceptors.size()) { - return this.doSend(message); + boolean result = this.doSend(message); + for (int i = index - 1; i >= 0; i--) { + EndpointInterceptor interceptor = this.interceptors.get(i); + interceptor.postSend(message, result); + } + return result; } EndpointInterceptor nextInterceptor = interceptors.get(index); - result = nextInterceptor.aroundSend(message, new MessageTarget() { + return nextInterceptor.aroundSend(message, new MessageTarget() { @SuppressWarnings("unchecked") public boolean send(Message message) { return AbstractEndpoint.this.send(message, index + 1); } }); - if (index == this.interceptors.size()) { - for (EndpointInterceptor interceptor : this.interceptors) { - interceptor.postSend(message, result); - } - } - return result; } private boolean doSend(Message message) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java index 496ccc307f..691cca53c5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java @@ -17,6 +17,8 @@ package org.springframework.integration.endpoint.interceptor; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; @@ -105,6 +107,16 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter if (this.executor instanceof DisposableBean) { ((DisposableBean) this.executor).destroy(); } + if (this.executor instanceof ConcurrentTaskExecutor) { + Executor innerExecutor = ((ConcurrentTaskExecutor) this.executor).getConcurrentExecutor(); + if (innerExecutor instanceof ExecutorService) { + ExecutorService executorService = (ExecutorService) innerExecutor; + executorService.shutdown(); + if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { + executorService.shutdownNow(); + } + } + } } @Override diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java index cd97b42c0b..1296982d15 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java @@ -213,7 +213,7 @@ public class SimpleTaskScheduler extends AbstractTaskScheduler implements Dispos logger.warn("error occurred in task but no 'errorHandler' is available", t); } } - if (this.shouldRepeat) { + if (this.shouldRepeat && isRunning()) { TaskRunner runner = new TaskRunner(this.task); runner.setShouldRepeat(true); executor.execute(runner); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java index 3188c4a8f3..20680015e4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -111,7 +111,7 @@ public class WireTapTests { QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); Message message = new StringMessage("testing"); - Thread.sleep(3); + Thread.sleep(50); mainChannel.send(message); Message original = mainChannel.receive(0); Message duplicate = secondaryChannel.receive(0);