Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Simplified the task scheduler configuration and exposed setters/constructors for ScheduledExecutorService (INT-113).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-18 22:58:45 +00:00
parent 69b13fc3bf
commit d9d15cab76
8 changed files with 115 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -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<T> extends AbstractSourceAdapter<T> 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();

64
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
private final Log logger = LogFactory.getLog(this.getClass());
private Map<MessageChannel, SchedulingMessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, SchedulingMessageDispatcher>();
private final ChannelRegistry channelRegistry = new DefaultChannelRegistry();
private List<Lifecycle> lifecycleSourceAdapters = new CopyOnWriteArrayList<Lifecycle>();
private final Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
private MessagingTaskScheduler taskScheduler;
private final Map<MessageChannel, SchedulingMessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, SchedulingMessageDispatcher>();
private int dispatcherPoolSize = 10;
private final List<Lifecycle> lifecycleSourceAdapters = new CopyOnWriteArrayList<Lifecycle>();
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()) {

44
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java
View File

@@ -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<Schedule, List<MessageHandler>> scheduledHandlers = new ConcurrentHashMap<Schedule, List<MessageHandler>>();
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<MessageHandler> 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) {

71
spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java
View File

@@ -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<Runnable> pendingTasks = new CopyOnWriteArraySet<Runnable>();
private String threadNamePrefix = this.getClass().getSimpleName() + "-";
private ErrorHandler errorHandler;
private Set<Runnable> pendingTasks = new CopyOnWriteArraySet<Runnable>();
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;
}
}
}

22
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java
View File

@@ -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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);

18
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
View File

@@ -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>(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");

10
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java
View File

@@ -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);

39
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java
View File

@@ -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));