diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java index 78a1ec12b5..a69ab622af 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java +++ b/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. *

* 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 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(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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java index f2b618e24f..e6c3344094 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java +++ b/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 null, 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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencerEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencerEndpoint.java index 74f50e9333..de3f0e1304 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencerEndpoint.java +++ b/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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index 15edffae28..b927a8bc84 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/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()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java index 046814a901..337478d306 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java +++ b/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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedulers.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedulers.java new file mode 100644 index 0000000000..068c7704d4 --- /dev/null +++ b/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); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java index 6dcfc8b7b6..9396d85beb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java +++ b/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(); + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerEndpointTests.java index 8be2517686..43f186a1ca 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerEndpointTests.java +++ b/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(); + } }