From d9d15cab76075434bb5cfab6cdbddfe8029f0754 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 18 Feb 2008 22:58:45 +0000 Subject: [PATCH] Simplified the task scheduler configuration and exposed setters/constructors for ScheduledExecutorService (INT-113). --- .../adapter/PollingSourceAdapter.java | 5 +- .../integration/bus/MessageBus.java | 64 ++++++----------- .../dispatcher/DefaultMessageDispatcher.java | 44 ++++-------- .../SimpleMessagingTaskScheduler.java | 71 +++++++------------ .../stream/ByteStreamTargetAdapterTests.java | 22 ++++-- .../CharacterStreamTargetAdapterTests.java | 18 +++-- .../config/ChannelParserTests.java | 10 ++- .../DefaultMessageDispatcherTests.java | 39 +++++----- 8 files changed, 115 insertions(+), 158 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 7b8a638689..de1cfa8420 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -17,6 +17,7 @@ package org.springframework.integration.adapter; import java.util.Collection; +import java.util.concurrent.Executors; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannel; @@ -97,9 +98,7 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements if (logger.isInfoEnabled()) { logger.info("no task scheduler has been provided, will create one"); } - SimpleMessagingTaskScheduler taskScheduler = new SimpleMessagingTaskScheduler(); - taskScheduler.setCorePoolSize(1); - this.scheduler = taskScheduler; + this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor()); } if (!this.scheduler.isRunning()) { this.scheduler.start(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 29ed28b585..9114ede538 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -20,6 +20,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -47,7 +49,6 @@ import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.integration.scheduling.Subscription; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; /** @@ -58,21 +59,24 @@ import org.springframework.util.Assert; */ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lifecycle { - private Log logger = LogFactory.getLog(this.getClass()); + private static final int DEFAULT_DISPATCHER_POOL_SIZE = 10; - private ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - private Map endpoints = new ConcurrentHashMap(); + private final Log logger = LogFactory.getLog(this.getClass()); - private Map dispatchers = new ConcurrentHashMap(); + private final ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - private List lifecycleSourceAdapters = new CopyOnWriteArrayList(); + private final Map endpoints = new ConcurrentHashMap(); - private MessagingTaskScheduler taskScheduler; + private final Map dispatchers = new ConcurrentHashMap(); - private int dispatcherPoolSize = 10; + private final List lifecycleSourceAdapters = new CopyOnWriteArrayList(); - private boolean autoCreateChannels; + private volatile MessagingTaskScheduler taskScheduler; + + private volatile ScheduledExecutorService executor; + + private volatile boolean autoCreateChannels; private volatile boolean autoStartup = true; @@ -82,7 +86,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private volatile boolean running; - private Object lifecycleMonitor = new Object(); + private final Object lifecycleMonitor = new Object(); public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { @@ -99,23 +103,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public void setMessagingTaskScheduler(MessagingTaskScheduler taskScheduler) { - Assert.notNull(taskScheduler, "task scheduler must not be null"); - if (taskScheduler instanceof SimpleMessagingTaskScheduler) { - ((SimpleMessagingTaskScheduler) taskScheduler).setCorePoolSize(dispatcherPoolSize); - } - this.taskScheduler = taskScheduler; - } - /** - * Set the size for the dispatcher thread pool. + * Set the {@link ScheduledExecutorService} to use for scheduling message dispatchers. */ - public void setDispatcherPoolSize(int dispatcherPoolSize) { - Assert.isTrue(dispatcherPoolSize > 0, "'dispatcherPoolSize' must be at least 1"); - this.dispatcherPoolSize = dispatcherPoolSize; - if (this.taskScheduler != null && this.taskScheduler instanceof SimpleMessagingTaskScheduler) { - ((SimpleMessagingTaskScheduler) this.taskScheduler).setCorePoolSize(dispatcherPoolSize); - } + public void setScheduledExecutorService(ScheduledExecutorService executor) { + this.executor = executor; } /** @@ -170,25 +162,16 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (this.getErrorChannel() == null) { this.setErrorChannel(new SimpleChannel(Integer.MAX_VALUE)); } - if (this.taskScheduler == null) { - this.setMessagingTaskScheduler(createDefaultScheduler()); + if (this.executor == null) { + this.executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE); } + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(this.executor); + scheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); + this.taskScheduler = scheduler; this.initialized = true; } } - private MessagingTaskScheduler createDefaultScheduler() { - CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); - threadFactory.setThreadNamePrefix("dispatcher-executor-"); - threadFactory.setThreadGroup(new ThreadGroup("dispatcher-executors")); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); - scheduler.setCorePoolSize(this.dispatcherPoolSize); - scheduler.setThreadFactory(threadFactory); - scheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); - scheduler.afterPropertiesSet(); - return scheduler; - } - public MessageChannel getErrorChannel() { return this.channelRegistry.getErrorChannel(); } @@ -206,8 +189,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.initialize(); } channel.setName(name); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setMessagingTaskScheduler(this.taskScheduler); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, this.taskScheduler); this.dispatchers.put(channel, dispatcher); this.channelRegistry.registerChannel(name, channel); if (logger.isInfoEnabled()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java index 39f5bd5dd4..740606bc13 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -27,15 +27,14 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; +import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; -import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.util.Assert; /** @@ -47,7 +46,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, MessagingTaskSchedulerAware { +public class DefaultMessageDispatcher implements SchedulingMessageDispatcher { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -55,33 +54,28 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me private final MessageRetriever retriever; - private MessagingTaskScheduler scheduler; + private final MessagingTaskScheduler scheduler; - private Schedule defaultSchedule = new PollingSchedule(5); + private volatile Schedule defaultSchedule = new PollingSchedule(5); private final ConcurrentMap> scheduledHandlers = new ConcurrentHashMap>(); private final AtomicLong totalMessagesProcessed = new AtomicLong(); - private volatile boolean starting; - private volatile boolean running; private final Object lifecycleMonitor = new Object(); - public DefaultMessageDispatcher(MessageChannel channel) { + public DefaultMessageDispatcher(MessageChannel channel, MessagingTaskScheduler scheduler) { Assert.notNull(channel, "'channel' must not be null"); + Assert.notNull(scheduler, "'scheduler' must not be null"); this.channel = channel; + this.scheduler = scheduler; this.retriever = new ChannelPollingMessageRetriever(this.channel); } - public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) { - Assert.notNull(scheduler, "'scheduler' must not be null"); - this.scheduler = scheduler; - } - public void setDefaultSchedule(Schedule defaultSchedule) { Assert.notNull(defaultSchedule, "'defaultSchedule' must not be null"); this.defaultSchedule = defaultSchedule; @@ -123,33 +117,23 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me public void start() { synchronized (this.lifecycleMonitor) { - if (this.isRunning() || this.starting) { + if (this.running) { return; } - this.starting = true; - } - if (this.scheduler == null) { - if (logger.isInfoEnabled()) { - logger.info("no scheduler was provided, will create one"); + if (this.scheduler == null) { + throw new MessagingConfigurationException("'scheduler' is required"); + } + if (!this.scheduler.isRunning()) { + this.scheduler.start(); } - this.scheduler = new SimpleMessagingTaskScheduler(); - } - if (!this.scheduler.isRunning()) { - this.scheduler.start(); - } - synchronized (this.lifecycleMonitor) { for (Schedule schedule : this.scheduledHandlers.keySet()) { scheduleDispatcherTask(schedule); } this.running = true; - this.starting = false; } } private void scheduleDispatcherTask(Schedule schedule) { - if (!this.isRunning()) { - this.start(); - } List handlers = this.scheduledHandlers.get(schedule); for (MessageHandler handler : handlers) { if (handler instanceof Lifecycle) { @@ -160,7 +144,7 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me } public void stop() { - if (!this.isRunning()) { + if (!this.running) { return; } synchronized (this.lifecycleMonitor) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java index a8a8404397..1f09dcf4e7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java @@ -21,12 +21,9 @@ import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; -import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.util.ErrorHandler; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; /** @@ -35,78 +32,60 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler implements InitializingBean { +public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler { - private ScheduledExecutorService executor; + private final ScheduledExecutorService executor; - private int corePoolSize = 10; + private volatile ErrorHandler errorHandler; - private ThreadFactory threadFactory; + private final Set pendingTasks = new CopyOnWriteArraySet(); - private String threadNamePrefix = this.getClass().getSimpleName() + "-"; - - private ErrorHandler errorHandler; - - private Set pendingTasks = new CopyOnWriteArraySet(); + private volatile boolean starting; private volatile boolean running; - private Object lifecycleMonitor = new Object(); + private final Object lifecycleMonitor = new Object(); - public void setExecutor(ScheduledExecutorService executor) { + public SimpleMessagingTaskScheduler(int corePoolSize) { + this(new ScheduledThreadPoolExecutor(corePoolSize)); + } + + public SimpleMessagingTaskScheduler(ScheduledExecutorService executor) { Assert.notNull(executor, "'executor' must not be null"); this.executor = executor; } - public void setCorePoolSize(int corePoolSize) { - Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be greater than 0"); - this.corePoolSize = corePoolSize; - } - - public void setThreadFactory(ThreadFactory threadFactory) { - this.threadFactory = threadFactory; - } - - public void setThreadNamePrefix(String threadNamePrefix) { - Assert.notNull(threadNamePrefix, "'threadNamePrefix' must not be null"); - this.threadNamePrefix = threadNamePrefix; - } public void setErrorHandler(ErrorHandler errorHandler) { this.errorHandler = errorHandler; } - public void afterPropertiesSet() { - if (this.executor == null) { - if (this.threadFactory == null) { - this.threadFactory = new CustomizableThreadFactory(this.threadNamePrefix); - } - this.executor = new ScheduledThreadPoolExecutor(this.corePoolSize, this.threadFactory); - } - } - public boolean isRunning() { return this.running; } public void start() { - if (this.executor == null) { - this.afterPropertiesSet(); - } synchronized (this.lifecycleMonitor) { - this.running = true; - for (Runnable task : this.pendingTasks) { - this.schedule(task); + if (this.running || this.starting) { + return; } - this.pendingTasks.clear(); + this.starting = true; } + for (Runnable task : this.pendingTasks) { + this.schedule(task); + } + this.pendingTasks.clear(); + this.running = true; + this.starting = false; } public void stop() { - if (this.isRunning()) { - this.running = false; - this.executor.shutdownNow(); + synchronized (this.lifecycleMonitor) { + if (this.running) { + this.executor.shutdownNow(); + this.running = false; + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index fde0d7db1b..4f5b47d0a8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -29,6 +29,7 @@ import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher @@ -64,7 +65,8 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(3); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -83,7 +85,8 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -102,7 +105,8 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -121,7 +125,8 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -145,7 +150,8 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -168,7 +174,8 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -191,7 +198,8 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index f5b24e11ce..cbd9365956 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -29,12 +29,16 @@ import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher */ public class CharacterStreamTargetAdapterTests { + private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + + @Test public void testSingleString() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); @@ -49,7 +53,7 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); @@ -67,7 +71,7 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setShouldAppendNewLine(true); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); @@ -87,7 +91,7 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); @@ -104,7 +108,7 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(10); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); adapter.setShouldAppendNewLine(true); dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); @@ -120,7 +124,7 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); @@ -138,7 +142,7 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); @@ -157,7 +161,7 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); adapter.setShouldAppendNewLine(true); dispatcher.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index fe55cc6006..25821fe921 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -37,6 +37,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher @@ -66,7 +67,8 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault"); channel.send(new StringMessage("test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(1); TestHandler handler1 = new TestHandler(counter, latch); @@ -85,7 +87,8 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured"); channel.send(new StringMessage("test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(1); TestHandler handler1 = new TestHandler(counter, latch); @@ -104,7 +107,8 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel"); channel.send(new StringMessage("test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(2); TestHandler handler1 = new TestHandler(counter, latch); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index 9d0ec76a8f..2748fb31c3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -52,6 +53,9 @@ import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; */ public class DefaultMessageDispatcherTests { + private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(10)); + + @Test public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException { final AtomicInteger counter1 = new AtomicInteger(); @@ -61,12 +65,9 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor())); dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor())); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); - scheduler.start(); - dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -82,7 +83,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor())); dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor())); dispatcher.start(); @@ -102,7 +103,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler1, createExecutor()); inactiveHandler.destroy(); dispatcher.addHandler(inactiveHandler); @@ -126,7 +127,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler2, createExecutor()); inactiveHandler.destroy(); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor())); @@ -143,7 +144,7 @@ public class DefaultMessageDispatcherTests { public void testDispatcherWithNoExecutorsDoesNotFail() { SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.start(); } @@ -160,7 +161,7 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor())); dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) { @Override @@ -170,9 +171,7 @@ public class DefaultMessageDispatcherTests { }); dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor())); SimpleChannel errorChannel = new SimpleChannel(); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -194,7 +193,7 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(handler1); dispatcher.addHandler(new MessageHandler() { public Message handle(Message message) { @@ -216,15 +215,13 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(false)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); dispatcher.addHandler(handler1); dispatcher.addHandler(handler2); SimpleChannel errorChannel = new SimpleChannel(); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -248,7 +245,7 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) { @Override public Message handle(Message message) { @@ -292,7 +289,7 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setShouldFailOnRejectionLimit(false); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) { @Override public Message handle(Message message) { @@ -346,7 +343,7 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setShouldFailOnRejectionLimit(false); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) { @Override public Message handle(Message message) { @@ -387,7 +384,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1); DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2); endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); @@ -413,7 +410,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, attemptedLatch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1); DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2); endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); @@ -456,7 +453,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(); endpoint1.setHandler(handler1); endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));