Force AbstractMessageBarrierEndpoint and subclasses to use the TaskScheduler supplied via TaskSchedulerAware for scheduling the ReaperThread. Introducing a Schedulers helper class for creating a default TaskScheduler.

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java

@@ -22,19 +22,19 @@ import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.integration.channel.BlockingChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.scheduling.IntervalTrigger;
+import org.springframework.integration.scheduling.TaskSchedulerAware;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -43,7 +43,7 @@ import org.springframework.util.ObjectUtils;
  * Base class for {@link MessageBarrier}-based MessageHandlers.
  * A {@link MessageEndpoint} implementation that waits for a group of
  * {@link Message Messages} to arrive and processes them together.
- * Uses a {@link MessageBarrier} to store messages and to decide how
+ * Uses a {@link MessageBarr  ier} to store messages and to decide how
  * the messages should be released.
  * <p>
  * Each {@link Message} that is received by this endpoint will be associated with
@@ -60,7 +60,7 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
-public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHandlingEndpoint {
+public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHandlingEndpoint implements TaskSchedulerAware {
 
 	public final static long DEFAULT_SEND_TIMEOUT = 1000;
 
@@ -89,15 +89,9 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHand
 
 	protected volatile BlockingQueue<Object> trackedCorrelationIds;
 
-	protected final ScheduledExecutorService executor;
-
 	private volatile boolean initialized;
 
-
-	public AbstractMessageBarrierEndpoint(ScheduledExecutorService executor) {
-		this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor();		
-	}
-
+	private ScheduledFuture<?> reaperFutureTask;
 
 	/**
 	 * Specify a channel for sending Messages that arrive after their aggregation
@@ -154,10 +148,20 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHand
 	protected void initialize()  throws Exception {
 		super.initialize();
 		this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
-		this.executor.scheduleWithFixedDelay(new ReaperTask(),
-				this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS);
 		this.initialized = true;
 	}
+	
+	@Override
+	protected void onStart() {
+		super.onStart();
+		this.reaperFutureTask = this.getTaskScheduler().schedule(new ReaperTask(), new IntervalTrigger(reaperInterval, TimeUnit.MILLISECONDS));
+	}
+
+	@Override
+	protected void onStop() {
+		super.onStop();
+		this.reaperFutureTask.cancel(true);
+	}
 
 	@Override
 	protected final Message<?> handle(Message<?> message) {

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java

@@ -17,10 +17,10 @@
 package org.springframework.integration.aggregator;
 
 import java.util.List;
-import java.util.concurrent.ScheduledExecutorService;
 
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 
 /**
@@ -51,8 +51,8 @@ public class AggregatorEndpoint extends AbstractMessageBarrierEndpoint {
 	 * scheduling a background maintenance thread. If <code>null</code>, a new
 	 * single-threaded executor will be created.
 	 */
-	public AggregatorEndpoint(Aggregator aggregator, ScheduledExecutorService executor) {
-		super(executor);
+	public AggregatorEndpoint(Aggregator aggregator, TaskScheduler executor) {
+		super();
 		Assert.notNull(aggregator, "'aggregator' must not be null");
 		this.aggregator = aggregator;
 	}

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencerEndpoint.java

@@ -17,7 +17,6 @@
 package org.springframework.integration.aggregator;
 
 import java.util.List;
-import java.util.concurrent.ScheduledExecutorService;
 
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageHeaders;
@@ -41,15 +40,6 @@ public class ResequencerEndpoint extends AbstractMessageBarrierEndpoint {
 	private volatile boolean releasePartialSequences = true;
 
 
-	public ResequencerEndpoint() {
-		this(null);
-	}
-
-	public ResequencerEndpoint(ScheduledExecutorService executor) {
-		super(executor);
-	}
-
-
 	public void setReleasePartialSequences(boolean releasePartialSequences) {
 		this.releasePartialSequences = releasePartialSequences;
 	}

org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java

@@ -23,11 +23,9 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
@@ -46,11 +44,11 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.MessagingGateway;
+import org.springframework.integration.scheduling.Schedulers;
 import org.springframework.integration.scheduling.SimpleTaskScheduler;
 import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.TaskSchedulerAware;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.Assert;
 
 /**
@@ -165,12 +163,8 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 			}
 			Assert.notNull(this.applicationContext, "ApplicationContext must not be null");
 			if (this.taskScheduler == null) {
-				ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
-				executor.setCorePoolSize(DEFAULT_DISPATCHER_POOL_SIZE);
-				executor.setThreadFactory(new CustomizableThreadFactory("message-bus-"));
-				executor.setRejectedExecutionHandler(new CallerRunsPolicy());
-				executor.afterPropertiesSet();
-				this.taskScheduler = new SimpleTaskScheduler(executor);
+				this.taskScheduler = Schedulers.createDefaultTaskExecutor(DEFAULT_DISPATCHER_POOL_SIZE,
+						new CustomizableThreadFactory("message-bus-"));
 			}
 			if (this.getErrorChannel() == null) {
 				this.registerChannel(new DefaultErrorChannel());

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java

@@ -126,6 +126,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
 						"failed to start endpoint, no taskScheduler available");
 				this.pollerFuture = this.getTaskScheduler().schedule(this.poller, this.poller.getTrigger());
 			}
+			onStart();
 			this.running = true;
 		}
 	}
@@ -141,9 +142,27 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
 			else if (this.pollerFuture != null) {
 				this.pollerFuture.cancel(true);
 			}
+			onStop();
 			this.running = false;
 		}
 	}
+	
+	/**
+	 * Subclasses might override this to supply their own start code (e.g. if they start threads
+	 * on their own). This method will be called within the lifecycleMonitor.
+	 */
+	protected void onStart() {
+		
+	}
+	
+	/**
+	 * Subclasses might override this to supply their own stop code (e.g. if they stop threads
+	 * on their own).This method will be called within the lifecycleMonitor.
+	 * 
+	 */
+	protected void onStop() {
+		
+	}
 
 	public final void onMessage(Message<?> message) {
 		if (message == null || message.getPayload() == null) {

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedulers.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.scheduling;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
+
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+/**
+ * Helper class for creating predefined {@link TaskScheduler} classes.
+ * 
+ * @author Marius Bogoevici
+ */
+public class Schedulers {
+
+	public static TaskScheduler createDefaultTaskExecutor(int poolSize, ThreadFactory threadFactory) {
+		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+		executor.setCorePoolSize(poolSize);
+		executor.setThreadFactory(threadFactory);
+		executor.setRejectedExecutionHandler(new CallerRunsPolicy());
+		executor.afterPropertiesSet();
+		return new SimpleTaskScheduler(executor);
+	}
+	
+	public static TaskScheduler createDefaultTaskScheduler(int poolSize) {
+		return createDefaultTaskExecutor(poolSize, null);
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java

@@ -26,42 +26,45 @@ import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
-
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.StringMessage;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.integration.scheduling.Schedulers;
+import org.springframework.integration.scheduling.TaskScheduler;
 
 /**
  * @author Mark Fisher
  */
 public class AggregatorEndpointTests {
 
-	private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+	private final TaskScheduler taskScheduler = Schedulers.createDefaultTaskScheduler(10);
+	
+	private AggregatorEndpoint aggregator;
 
-
-	public AggregatorEndpointTests() {
-		this.executor.setMaxPoolSize(10);
-		this.executor.setQueueCapacity(0);
-		this.executor.afterPropertiesSet();
+	@Before
+	public void configureAggregator() {
+		this.aggregator = new AggregatorEndpoint(new TestAggregator());
+		this.aggregator.setTaskScheduler(this.taskScheduler);
+		this.taskScheduler.start();
+		this.aggregator.onStart();
 	}
 
-
 	@Test
 	public void testCompleteGroupWithinTimeout() throws InterruptedException {
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 3, 3, replyChannel);
 		CountDownLatch latch = new CountDownLatch(3);
-		executor.execute(new AggregatorTestTask(aggregator, message1, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message2, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message3, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message1, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message2, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message3, latch));
 		latch.await(1000, TimeUnit.MILLISECONDS);
 		Message<?> reply = replyChannel.receive(500);
 		assertNotNull(reply);
@@ -71,38 +74,36 @@ public class AggregatorEndpointTests {
 	@Test
 	public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException {
 		QueueChannel discardChannel = new QueueChannel();
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
-		aggregator.setTimeout(50);
-		aggregator.setReaperInterval(10);
-		aggregator.setDiscardChannel(discardChannel);
+		this.aggregator.setTimeout(50);
+		this.aggregator.setReaperInterval(10);
+		this.aggregator.setDiscardChannel(discardChannel);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message = createMessage("123", "ABC", 2, 1, replyChannel);
 		CountDownLatch latch = new CountDownLatch(1);
-		AggregatorTestTask task = new AggregatorTestTask(aggregator, message, latch);
-		executor.execute(task);
+		AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch);
+		this.taskScheduler.execute(task);
 		latch.await(2000, TimeUnit.MILLISECONDS);
 		assertEquals("task should have completed within timeout", 0, latch.getCount());
 		Message<?> reply = replyChannel.receive(0);
 		assertNull(reply);
-		Message<?> discardedMessage = discardChannel.receive(1000);
+		Message<?> discardedMessage = discardChannel.receive(2000);
 		assertNotNull(discardedMessage);
 		assertEquals(message, discardedMessage);
 	}
 
 	@Test
 	public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException {
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
-		aggregator.setTimeout(500);
-		aggregator.setReaperInterval(10);
-		aggregator.setSendPartialResultOnTimeout(true);
+		this.aggregator.setTimeout(500);
+		this.aggregator.setReaperInterval(10);
+		this.aggregator.setSendPartialResultOnTimeout(true);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
 		CountDownLatch latch = new CountDownLatch(2);
-		AggregatorTestTask task1 = new AggregatorTestTask(aggregator, message1, latch);
-		AggregatorTestTask task2 = new AggregatorTestTask(aggregator, message2, latch);
-		executor.execute(task1);
-		executor.execute(task2);
+		AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, message1, latch);
+		AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, message2, latch);
+		this.taskScheduler.execute(task1);
+		this.taskScheduler.execute(task2);
 		latch.await(3000, TimeUnit.MILLISECONDS);
 		assertEquals("handlers should have been invoked within time limit", 0, latch.getCount());
 		Message<?> reply = replyChannel.receive(3000);
@@ -114,7 +115,6 @@ public class AggregatorEndpointTests {
 
 	@Test
 	public void testMultipleGroupsSimultaneously() throws InterruptedException {
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
 		QueueChannel replyChannel1 = new QueueChannel();
 		QueueChannel replyChannel2 = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel1);
@@ -124,12 +124,12 @@ public class AggregatorEndpointTests {
 		Message<?> message5 = createMessage("def", "XYZ", 3, 2, replyChannel2);
 		Message<?> message6 = createMessage("ghi", "XYZ", 3, 3, replyChannel2);
 		CountDownLatch latch = new CountDownLatch(6);
-		executor.execute(new AggregatorTestTask(aggregator, message1, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message6, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message2, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message5, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message3, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message4, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message1, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message6, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message2, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message5, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message3, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message4, latch));
 		latch.await(1000, TimeUnit.MILLISECONDS);
 		Message<?> reply1 = replyChannel1.receive(500);
 		assertNotNull(reply1);
@@ -143,11 +143,10 @@ public class AggregatorEndpointTests {
 	public void testDiscardChannelForTrackedCorrelationId() {
 		QueueChannel replyChannel = new QueueChannel();
 		QueueChannel discardChannel = new QueueChannel();
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
-		aggregator.setDiscardChannel(discardChannel);
-		aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
+		this.aggregator.setDiscardChannel(discardChannel);
+		this.aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
 		assertEquals("test-1a", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
 		assertEquals("test-1b", discardChannel.receive(100).getPayload());
 	}
 
@@ -155,16 +154,15 @@ public class AggregatorEndpointTests {
 	public void testTrackedCorrelationIdsCapacityAtLimit() {
 		QueueChannel replyChannel = new QueueChannel();
 		QueueChannel discardChannel = new QueueChannel();
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
-		aggregator.setTrackedCorrelationIdCapacity(3);
-		aggregator.setDiscardChannel(discardChannel);
-		aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
+		this.aggregator.setTrackedCorrelationIdCapacity(3);
+		this.aggregator.setDiscardChannel(discardChannel);
+		this.aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
 		assertEquals("test-1a", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-2", 2, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-2", 2, 1, 1, replyChannel));
 		assertEquals("test-2", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-3", 3, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-3", 3, 1, 1, replyChannel));
 		assertEquals("test-3", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
 		assertEquals("test-1b", discardChannel.receive(100).getPayload());
 	}
 
@@ -172,42 +170,39 @@ public class AggregatorEndpointTests {
 	public void testTrackedCorrelationIdsCapacityPassesLimit() {
 		QueueChannel replyChannel = new QueueChannel();
 		QueueChannel discardChannel = new QueueChannel();
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
-		aggregator.setTrackedCorrelationIdCapacity(3);
-		aggregator.setDiscardChannel(discardChannel);
-		aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
+		this.aggregator.setTrackedCorrelationIdCapacity(3);
+		this.aggregator.setDiscardChannel(discardChannel);
+		this.aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
 		assertEquals("test-1a", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-2", 2, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-2", 2, 1, 1, replyChannel));
 		assertEquals("test-2", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-3", 3, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-3", 3, 1, 1, replyChannel));
 		assertEquals("test-3", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-4", 4, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-4", 4, 1, 1, replyChannel));
 		assertEquals("test-4", replyChannel.receive(100).getPayload());
-		aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
+		this.aggregator.handle(createMessage("test-1b", 1, 1, 1, replyChannel));
 		assertEquals("test-1b", replyChannel.receive(100).getPayload());
 		assertNull(discardChannel.receive(0));
 	}
 
 	@Test(expected=MessageHandlingException.class)
 	public void testExceptionThrownIfNoCorrelationId() throws InterruptedException {
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
 		Message<?> message = createMessage("123", null, 2, 1, new QueueChannel());
-		aggregator.handle(message);
+		this.aggregator.handle(message);
 	}
 	
 	@Test
 	public void testAdditionalMessageAfterCompletion() throws InterruptedException {
-		AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 3, 3, replyChannel);
 		Message<?> message4 = createMessage("abc", "ABC", 3, 3, replyChannel);
 		CountDownLatch latch = new CountDownLatch(4);
-		executor.execute(new AggregatorTestTask(aggregator, message1, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message2, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message3, latch));
-		executor.execute(new AggregatorTestTask(aggregator, message4, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message1, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message2, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message3, latch));
+		this.taskScheduler.execute(new AggregatorTestTask(this.aggregator, message4, latch));
 		latch.await(1000, TimeUnit.MILLISECONDS);
 		Message<?> reply = replyChannel.receive(500);
 		assertNotNull(reply);
@@ -216,26 +211,27 @@ public class AggregatorEndpointTests {
 	
 	@Test
 	public void testNullReturningAggregator() throws InterruptedException {
-		NullReturningAggregator aggregator = new NullReturningAggregator();
-		AggregatorEndpoint aggregatorEndpoint = new AggregatorEndpoint(aggregator);
+		NullReturningAggregator nullReturningAggregator = new NullReturningAggregator();
+		AggregatorEndpoint aggregator = new AggregatorEndpoint(nullReturningAggregator);
+//		aggregator.setTaskScheduler(this.taskScheduler);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 3, 3, replyChannel);
 		CountDownLatch latch = new CountDownLatch(3);
-		AggregatorTestTask task = new AggregatorTestTask(aggregatorEndpoint, message1, latch);
-		executor.execute(task);	
-		AggregatorTestTask task2 = new AggregatorTestTask(aggregatorEndpoint, message2, latch);
-		executor.execute(task2);
-		AggregatorTestTask task3 = new AggregatorTestTask(aggregatorEndpoint, message3, latch);
-		executor.execute(task3);
+		AggregatorTestTask task = new AggregatorTestTask(aggregator, message1, latch);
+		this.taskScheduler.execute(task);	
+		AggregatorTestTask task2 = new AggregatorTestTask(aggregator, message2, latch);
+		this.taskScheduler.execute(task2);
+		AggregatorTestTask task3 = new AggregatorTestTask(aggregator, message3, latch);
+		this.taskScheduler.execute(task3);
 		latch.await(1000, TimeUnit.MILLISECONDS);
 		assertNull(task.getException());
 		assertNull(task2.getException());
 		assertNull(task3.getException());
 		Message<?> reply = replyChannel.receive(500);
 		assertNull(reply);
-		assertEquals(true, aggregator.isAggregationComplete());
+		assertEquals(true, nullReturningAggregator.isAggregationComplete());
 	}
 
 
@@ -325,5 +321,11 @@ public class AggregatorEndpointTests {
 			}
 		}
 	}
+	
+	@After
+	public void stopTaskScheduler() {
+		this.taskScheduler.stop();
+		this.aggregator.onStop();
+	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerEndpointTests.java

@@ -20,30 +20,45 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
-
-import org.springframework.integration.aggregator.ResequencerEndpoint;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.scheduling.Schedulers;
+import org.springframework.integration.scheduling.TaskScheduler;
 
 /**
  * @author Marius Bogoevici
  */
 public class ResequencerEndpointTests {
 
+	private ResequencerEndpoint resequencer;
+	
+	private TaskScheduler taskScheduler;
+
+	
+	@Before
+	public void configureResequencer() {
+		this.resequencer = new ResequencerEndpoint();
+		this.taskScheduler = Schedulers.createDefaultTaskScheduler(10);
+		this.resequencer.setTaskScheduler(taskScheduler);
+		taskScheduler.start();
+		this.resequencer.onStart();
+	}
+	
 	@Test
 	public void testBasicResequencing() throws InterruptedException {
-		ResequencerEndpoint resequencer = new ResequencerEndpoint();
-		resequencer.setReleasePartialSequences(false);
+		this.resequencer.setReleasePartialSequences(false);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 3, 2, replyChannel);
-		resequencer.handle(message1);
-		resequencer.handle(message3);
-		resequencer.handle(message2);
+		this.resequencer.handle(message1);
+		this.resequencer.handle(message3);
+		this.resequencer.handle(message2);
 		Message<?> reply1 = replyChannel.receive(0);
 		Message<?> reply2 = replyChannel.receive(0);
 		Message<?> reply3 = replyChannel.receive(0);
@@ -57,16 +72,15 @@ public class ResequencerEndpointTests {
 
 	@Test
 	public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException {
-		ResequencerEndpoint resequencer = new ResequencerEndpoint();
-		resequencer.setReleasePartialSequences(true);
+		this.resequencer.setReleasePartialSequences(true);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
 		Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
-		resequencer.handle(message1);
-		resequencer.handle(message2);
-		resequencer.handle(message3);
+		this.resequencer.handle(message1);
+		this.resequencer.handle(message2);
+		this.resequencer.handle(message3);
 		Message<?> reply1 = replyChannel.receive(0);
 		Message<?> reply2 = replyChannel.receive(0);
 		Message<?> reply3 = replyChannel.receive(0);
@@ -77,7 +91,7 @@ public class ResequencerEndpointTests {
 		assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
 		assertNull(reply3);
 		// when sending the last message, the whole sequence must have been sent
-		resequencer.handle(message4);
+		this.resequencer.handle(message4);
 		reply3 = replyChannel.receive(0);
 		Message<?> reply4 = replyChannel.receive(0);
 		assertNotNull(reply3);
@@ -89,16 +103,15 @@ public class ResequencerEndpointTests {
 
 	@Test
 	public void testResequencingWithCompleteSequenceRelease() throws InterruptedException {
-		ResequencerEndpoint resequencer = new ResequencerEndpoint();
-		resequencer.setReleasePartialSequences(false);
+		this.resequencer.setReleasePartialSequences(false);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
 		Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
 		Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
 		Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
-		resequencer.handle(message1);
-		resequencer.handle(message2);
-		resequencer.handle(message3);
+		this.resequencer.handle(message1);
+		this.resequencer.handle(message2);
+		this.resequencer.handle(message3);
 		Message<?> reply1 = replyChannel.receive(0);
 		Message<?> reply2 = replyChannel.receive(0);
 		Message<?> reply3 = replyChannel.receive(0);
@@ -107,7 +120,7 @@ public class ResequencerEndpointTests {
 		assertNull(reply2);
 		assertNull(reply3);
 		// after sending the last message, the whole sequence should have been sent
-		resequencer.handle(message4);
+		this.resequencer.handle(message4);
 		reply1 = replyChannel.receive(0);
 		reply2 = replyChannel.receive(0);
 		reply3 = replyChannel.receive(0);
@@ -134,4 +147,9 @@ public class ResequencerEndpointTests {
 		return message;
 	}
 
+	@After
+	public void stopTaskScheduler() {
+		this.resequencer.onStop();
+		this.taskScheduler.stop();
+	}
 }