diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java index 29cc75dbfb..42cb9f1299 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java @@ -16,6 +16,8 @@ package org.springframework.integration.mail; +import java.util.Date; + import javax.mail.Message; import org.apache.commons.logging.Log; @@ -23,10 +25,10 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; -import org.springframework.core.task.TaskExecutor; import org.springframework.integration.endpoint.AbstractMessageProducingEndpoint; import org.springframework.integration.mail.monitor.AsyncMonitoringStrategy; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.scheduling.Trigger; import org.springframework.util.Assert; /** @@ -46,8 +48,6 @@ public class ListeningMailSource extends AbstractMessageProducingEndpoint implem private volatile boolean monitorRunning = false; - private volatile TaskExecutor taskExecutor; - public ListeningMailSource(FolderConnection folderConnection) { Assert.notNull(folderConnection, "FolderConnection must not be null"); @@ -55,10 +55,6 @@ public class ListeningMailSource extends AbstractMessageProducingEndpoint implem } - public void setTaskExecutor(TaskExecutor taskExecutor) { - this.taskExecutor = taskExecutor; - } - // Lifecycle implementation public boolean isRunning() { @@ -82,11 +78,14 @@ public class ListeningMailSource extends AbstractMessageProducingEndpoint implem protected void startMonitor() { synchronized (this.monitorRunnable) { if (!this.monitorRunning) { - if (this.taskExecutor == null) { - this.taskExecutor = this.getTaskScheduler(); - } - Assert.state(this.taskExecutor != null, "TaskExecutor is required"); - this.taskExecutor.execute(this.monitorRunnable); + Assert.state(this.getTaskScheduler() != null, "TaskScheduler is required"); + this.getTaskScheduler().schedule(this.monitorRunnable, + new Trigger() { + public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { + return new Date(); + } + } + ); } this.monitorRunning = true; } @@ -119,7 +118,12 @@ public class ListeningMailSource extends AbstractMessageProducingEndpoint implem public synchronized void interrupt() { - this.thread.interrupt(); + if (this.thread != null) { + this.thread.interrupt(); + } + else if (logger.isInfoEnabled()) { + logger.info("monitor is not running, cannot interrupt"); + } } public void run() { diff --git a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java index 80e3965f1e..13f16ec8e1 100644 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java +++ b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java @@ -24,41 +24,35 @@ import java.util.concurrent.ConcurrentLinkedQueue; import javax.mail.internet.MimeMessage; import org.easymock.classextension.EasyMock; -import org.junit.Before; import org.junit.Test; -import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; -import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; +import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.integration.util.TestUtils; /** * @author Jonas Partner */ public class SubscribableMailSourceTests { - private TaskExecutor executor; - - - @Before - public void setUp() { - executor = new ConcurrentTaskExecutor(); - } - - @Test public void testReceive() throws Exception { javax.mail.Message message = EasyMock.createMock(MimeMessage.class); StubFolderConnection folderConnection = new StubFolderConnection(message); QueueChannel channel = new QueueChannel(); + TaskScheduler scheduler = TestUtils.createTaskScheduler(5); + scheduler.start(); ListeningMailSource mailSource = new ListeningMailSource(folderConnection); - mailSource.setTaskExecutor(executor); + mailSource.setTaskScheduler(scheduler); mailSource.setOutputChannel(channel); mailSource.start(); Message result = channel.receive(1000); mailSource.stop(); assertNotNull(result); assertEquals("Wrong payload", message, result.getPayload()); + mailSource.stop(); + scheduler.stop(); } 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 9b4af0c914..ee35cc51ce 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 @@ -34,21 +34,17 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; import org.springframework.context.Lifecycle; -import org.springframework.context.event.ApplicationEventMulticaster; import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.context.event.SimpleApplicationEventMulticaster; -import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistryAware; 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.util.Assert; /** @@ -60,8 +56,6 @@ import org.springframework.util.Assert; */ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, ApplicationListener, DisposableBean { - private static final int DEFAULT_DISPATCHER_POOL_SIZE = 10; - private final Log logger = LogFactory.getLog(this.getClass()); private final Map channels = new ConcurrentHashMap(); @@ -76,8 +70,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private volatile ApplicationContext applicationContext; - private volatile boolean configureAsyncEventMulticaster = false; - private volatile boolean autoStartup = true; private volatile boolean initialized; @@ -112,17 +104,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A this.autoStartup = autoStartup; } - /** - * Set whether the bus should configure its asynchronous task executor - * to also be used by the ApplicationContext's 'applicationEventMulticaster'. - * This will only apply if the multicaster defined within the context - * is an instance of SimpleApplicationEventMulticaster (the default). - * This property is 'false' by default. - */ - public void setConfigureAsyncEventMulticaster(boolean configureAsyncEventMulticaster) { - this.configureAsyncEventMulticaster = configureAsyncEventMulticaster; - } - @SuppressWarnings("unchecked") private void registerChannels(ApplicationContext context) { Map channelBeans = (Map) context @@ -155,9 +136,9 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A return; } Assert.notNull(this.applicationContext, "ApplicationContext must not be null"); + //TODO: Assert.notNull(this.taskScheduler, "TaskScheduler must not be null"); if (this.taskScheduler == null) { - this.taskScheduler = Schedulers.createDefaultTaskExecutor(DEFAULT_DISPATCHER_POOL_SIZE, - new CustomizableThreadFactory("message-bus-")); + this.taskScheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor()); } if (this.getErrorChannel() == null) { this.registerChannel(new DefaultErrorChannel()); @@ -339,27 +320,12 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A ApplicationContext context = ((ContextRefreshedEvent) event).getApplicationContext(); this.registerChannels(context); this.registerGateways(context); - if (this.configureAsyncEventMulticaster) { - this.initialize(); - this.doConfigureAsyncEventMulticaster(context); - } if (this.autoStartup) { this.start(); } } } - private void doConfigureAsyncEventMulticaster(ApplicationContext context) { - String multicasterBeanName = AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME; - if (context.containsBean(multicasterBeanName)) { - ApplicationEventMulticaster multicaster = (ApplicationEventMulticaster) context - .getBean(multicasterBeanName); - if (multicaster instanceof SimpleApplicationEventMulticaster) { - ((SimpleApplicationEventMulticaster) multicaster).setTaskExecutor(this.taskScheduler); - } - } - } - public void addInterceptor(MessageBusInterceptor interceptor) { this.interceptors.add(interceptor); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageBusParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageBusParser.java index 0a7fb8aa9f..4123cb1bca 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageBusParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageBusParser.java @@ -16,26 +16,38 @@ package org.springframework.integration.config; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; + import org.w3c.dom.Element; import org.w3c.dom.Node; import org.w3c.dom.NodeList; import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.context.event.SimpleApplicationEventMulticaster; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.core.task.TaskExecutor; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.bus.MessageBusAwareBeanPostProcessor; import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; import org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor; +import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Parser for the <message-bus> element of the integration namespace. @@ -56,6 +68,10 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser { private static final String MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME = "internal.MessagingAnnotationPostProcessor"; + private static final String TASK_SCHEDULER_ATTRIBUTE = "task-scheduler"; + + private static final String ASYNC_EVENT_MULTICASTER_ATTRIBUTE = "configure-async-event-multicaster"; + @Override protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) @@ -72,15 +88,49 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser { @Override protected boolean isEligibleAttribute(String attributeName) { - return !"task-scheduler".equals(attributeName) && + return !TASK_SCHEDULER_ATTRIBUTE.equals(attributeName) && + !ASYNC_EVENT_MULTICASTER_ATTRIBUTE.equals(attributeName) && !"enable-annotations".equals(attributeName) && super.isEligibleAttribute(attributeName); } @Override - protected void postProcess(BeanDefinitionBuilder builder, Element element) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-scheduler"); + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + super.doParse(element, parserContext, builder); + String taskSchedulerRef = element.getAttribute(TASK_SCHEDULER_ATTRIBUTE); + TaskExecutor taskExecutor= null; + if (StringUtils.hasText(taskSchedulerRef)) { + builder.addPropertyReference("taskScheduler", taskSchedulerRef); + } + else { + taskExecutor = this.createTaskExecutor(100, "message-bus-"); + builder.addPropertyValue("taskScheduler", new SimpleTaskScheduler(taskExecutor)); + } + if ("true".equals(element.getAttribute(ASYNC_EVENT_MULTICASTER_ATTRIBUTE).toLowerCase())) { + if (taskExecutor == null) { + taskExecutor = this.createTaskExecutor(10, "event-multicaster-"); + } + BeanDefinitionBuilder eventMulticasterBuilder = BeanDefinitionBuilder.genericBeanDefinition( + SimpleApplicationEventMulticaster.class); + eventMulticasterBuilder.addPropertyValue("taskExecutor", taskExecutor); + eventMulticasterBuilder.addPropertyValue("collectionClass", CopyOnWriteArraySet.class); + BeanDefinitionHolder holder = new BeanDefinitionHolder(eventMulticasterBuilder.getBeanDefinition(), + AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME); + BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry()); + } this.processChildElements(builder, element); + this.addPostProcessors(element, parserContext); + } + + private TaskExecutor createTaskExecutor(int corePoolSize, String threadPrefix) { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(corePoolSize); + if (StringUtils.hasText(threadPrefix)) { + executor.setThreadFactory(new CustomizableThreadFactory(threadPrefix)); + } + executor.setRejectedExecutionHandler(new CallerRunsPolicy()); + executor.afterPropertiesSet(); + return executor; } @SuppressWarnings("unchecked") @@ -100,12 +150,6 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser { } } - @Override - protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - super.doParse(element, parserContext, builder); - this.addPostProcessors(element, parserContext); - } - /** * Adds extra post-processors to the context, to inject the objects configured by the MessageBus */ diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index d642c7d24b..640c342606 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -20,7 +20,7 @@ - Defines a message bus. + Defines the Message Bus for this Application Context. 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 deleted file mode 100644 index 068c7704d4..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedulers.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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/main/java/org/springframework/integration/scheduling/TaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java index 2990ba4d3d..8f759ac5c4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java @@ -19,7 +19,6 @@ package org.springframework.integration.scheduling; import java.util.concurrent.ScheduledFuture; import org.springframework.context.Lifecycle; -import org.springframework.scheduling.SchedulingTaskExecutor; /** * Base interface for scheduling messaging tasks. @@ -27,7 +26,7 @@ import org.springframework.scheduling.SchedulingTaskExecutor; * @author Mark Fisher * @author Marius Bogoevici */ -public interface TaskScheduler extends SchedulingTaskExecutor, Lifecycle { +public interface TaskScheduler extends Lifecycle { ScheduledFuture schedule(Runnable task, Trigger trigger); 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 85c6782a57..74add7b598 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 @@ -31,13 +31,15 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; 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.integration.scheduling.Schedulers; +import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; /** @@ -45,13 +47,17 @@ import org.springframework.integration.scheduling.TaskScheduler; */ public class AggregatorEndpointTests { - private final TaskScheduler taskScheduler = Schedulers.createDefaultTaskScheduler(10); - + private TaskExecutor taskExecutor; + + private TaskScheduler taskScheduler; + private AbstractMessageAggregator aggregator; @Before public void configureAggregator() { + this.taskExecutor = new SimpleAsyncTaskExecutor(); + this.taskScheduler = new SimpleTaskScheduler(taskExecutor); this.aggregator = new TestAggregator(); this.aggregator.setTaskScheduler(this.taskScheduler); this.taskScheduler.start(); @@ -65,9 +71,9 @@ public class AggregatorEndpointTests { Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); CountDownLatch latch = new CountDownLatch(3); - 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.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); latch.await(1000, TimeUnit.MILLISECONDS); Message reply = replyChannel.receive(500); assertNotNull(reply); @@ -84,7 +90,7 @@ public class AggregatorEndpointTests { Message message = createMessage("123", "ABC", 2, 1, replyChannel); CountDownLatch latch = new CountDownLatch(1); AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch); - this.taskScheduler.execute(task); + this.taskExecutor.execute(task); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("task should have completed within timeout", 0, latch.getCount()); Message reply = replyChannel.receive(0); @@ -105,8 +111,8 @@ public class AggregatorEndpointTests { CountDownLatch latch = new CountDownLatch(2); AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, message1, latch); AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, message2, latch); - this.taskScheduler.execute(task1); - this.taskScheduler.execute(task2); + this.taskExecutor.execute(task1); + this.taskExecutor.execute(task2); latch.await(3000, TimeUnit.MILLISECONDS); assertEquals("handlers should have been invoked within time limit", 0, latch.getCount()); Message reply = replyChannel.receive(3000); @@ -127,12 +133,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); - 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)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message6, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message5, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); latch.await(1000, TimeUnit.MILLISECONDS); Message reply1 = replyChannel1.receive(500); assertNotNull(reply1); @@ -202,10 +208,10 @@ public class AggregatorEndpointTests { Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); Message message4 = createMessage("abc", "ABC", 3, 3, replyChannel); CountDownLatch latch = new CountDownLatch(4); - 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)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); latch.await(1000, TimeUnit.MILLISECONDS); Message reply = replyChannel.receive(500); assertNotNull(reply); @@ -222,11 +228,11 @@ public class AggregatorEndpointTests { Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); CountDownLatch latch = new CountDownLatch(3); AggregatorTestTask task1 = new AggregatorTestTask(aggregator, message1, latch); - this.taskScheduler.execute(task1); + this.taskExecutor.execute(task1); AggregatorTestTask task2 = new AggregatorTestTask(aggregator, message2, latch); - this.taskScheduler.execute(task2); + this.taskExecutor.execute(task2); AggregatorTestTask task3 = new AggregatorTestTask(aggregator, message3, latch); - this.taskScheduler.execute(task3); + this.taskExecutor.execute(task3); latch.await(1000, TimeUnit.MILLISECONDS); assertNull(task1.getException()); assertNull(task2.getException()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 10f587cc30..d6c8fb29e4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -23,12 +23,13 @@ import static org.junit.Assert.assertNull; 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.scheduling.Schedulers; import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.integration.util.TestUtils; /** * @author Marius Bogoevici @@ -43,7 +44,7 @@ public class ResequencerTests { @Before public void configureResequencer() { this.resequencer = new Resequencer(); - this.taskScheduler = Schedulers.createDefaultTaskScheduler(10); + this.taskScheduler = TestUtils.createTaskScheduler(10); this.resequencer.setTaskScheduler(taskScheduler); taskScheduler.start(); this.resequencer.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml index 660c2cf12f..4bd2d9eea9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml @@ -4,7 +4,14 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd"> - + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index 27f92e715c..a1a4a66023 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -46,6 +46,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.integration.util.TestUtils; /** * @author Mark Fisher @@ -74,6 +75,7 @@ public class DefaultMessageBusTests { context.getBeanFactory().registerSingleton("testEndpoint", endpoint); context.refresh(); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); consumer.setChannelRegistry(bus); bus.start(); @@ -86,6 +88,7 @@ public class DefaultMessageBusTests { public void channelsWithoutHandlers() { GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); QueueChannel sourceChannel = new QueueChannel(); sourceChannel.setBeanName("sourceChannel"); @@ -144,6 +147,7 @@ public class DefaultMessageBusTests { context.getBeanFactory().registerSingleton("testEndpoint1", endpoint1); context.getBeanFactory().registerSingleton("testEndpoint2", endpoint2); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); bus.start(); inputChannel.send(new StringMessage("testing")); @@ -187,6 +191,7 @@ public class DefaultMessageBusTests { context.getBeanFactory().registerSingleton("testEndpoint1", endpoint1); context.getBeanFactory().registerSingleton("testEndpoint2", endpoint2); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); bus.start(); inputChannel.send(new StringMessage("testing")); @@ -214,6 +219,7 @@ public class DefaultMessageBusTests { channelAdapter.setBeanName("testChannel"); context.getBeanFactory().registerSingleton("testChannel", channelAdapter); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); bus.start(); latch.await(2000, TimeUnit.MILLISECONDS); @@ -257,6 +263,7 @@ public class DefaultMessageBusTests { endpoint.afterPropertiesSet(); context.getBeanFactory().registerSingleton("testEndpoint", endpoint); DefaultMessageBus bus = new DefaultMessageBus(); + bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); bus.setApplicationContext(context); bus.start(); errorChannel.send(new ErrorMessage(new RuntimeException("test-exception"))); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml index a2aee37bee..edd2da0760 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -4,7 +4,14 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index aca15e2b8c..53c1045512 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -37,8 +37,8 @@ import org.springframework.integration.bus.MessageBusInterceptorTests; import org.springframework.integration.bus.TestMessageBusAwareImpl; import org.springframework.integration.bus.TestMessageBusStartInterceptor; import org.springframework.integration.bus.TestMessageBusStopInterceptor; -import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** * @author Mark Fisher @@ -144,7 +144,7 @@ public class MessageBusParserTests { context.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME); DirectFieldAccessor accessor = new DirectFieldAccessor(multicaster); Object taskExecutor = accessor.getPropertyValue("taskExecutor"); - assertEquals(SimpleTaskScheduler.class, taskExecutor.getClass()); + assertEquals(ThreadPoolTaskExecutor.class, taskExecutor.getClass()); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index a71735e8f6..db8904fbf4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -52,6 +52,7 @@ import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.IntervalTrigger; import org.springframework.integration.scheduling.Trigger; +import org.springframework.integration.util.TestUtils; /** * @author Mark Fisher @@ -144,6 +145,7 @@ public class MessagingAnnotationPostProcessorTests { DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton( MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -209,6 +211,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -235,6 +238,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -259,6 +263,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -285,6 +290,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -309,6 +315,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -333,6 +340,7 @@ public class MessagingAnnotationPostProcessorTests { outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -378,6 +386,7 @@ public class MessagingAnnotationPostProcessorTests { DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton( MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); @@ -412,6 +421,7 @@ public class MessagingAnnotationPostProcessorTests { context.getBeanFactory().registerSingleton("outputChannel", outputChannel); DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); + messageBus.setTaskScheduler(TestUtils.createTaskScheduler(10)); messageBus.setApplicationContext(context); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java b/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java index 27c89d1380..bfb5d0445f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java @@ -16,7 +16,12 @@ package org.springframework.integration.util; +import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; + import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; /** @@ -51,4 +56,12 @@ public abstract class TestUtils { return (T) value; } + public static TaskScheduler createTaskScheduler(int poolSize) { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(poolSize); + executor.setRejectedExecutionHandler(new CallerRunsPolicy()); + executor.afterPropertiesSet(); + return new SimpleTaskScheduler(executor); + } + }