From 67b6ea048aac949d960087652f59422582ff8c85 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 4 Aug 2009 01:25:53 +0000 Subject: [PATCH] Refactored all task scheduling and poller triggering to use functionality now included in the Spring 3.0 core, and removed Spring Integration specific code that is now handled by that corresponding code in the core. --- .../config/DefaultConfigurationTests.java | 18 +- .../config/DefaultConfigurationTests.java | 14 +- .../config/DefaultConfigurationTests.java | 14 +- .../jms/config/DefaultConfigurationTests.java | 14 +- .../config/DefaultConfigurationTests.java | 14 +- .../rmi/config/DefaultConfigurationTests.java | 14 +- .../config/DefaultConfigurationTests.java | 14 +- .../ByteStreamWritingMessageHandlerTests.java | 17 +- ...acterStreamWritingMessageHandlerTests.java | 17 +- .../config/DefaultConfigurationTests.java | 14 +- .../integration/test/util/TestUtils.java | 17 +- .../MarshallingWebServiceInboundGateway.java | 2 +- .../ws/config/DefaultConfigurationTests.java | 14 +- .../xml/config/DefaultConfigurationTests.java | 14 +- .../AbstractMessageBarrierHandler.java | 8 +- .../integration/channel/ExecutorChannel.java | 2 +- .../MessagePublishingErrorHandler.java | 6 +- .../channel/PublishSubscribeChannel.java | 2 +- .../ApplicationEventMulticasterParser.java | 14 +- ...ltConfiguringBeanFactoryPostProcessor.java | 12 +- .../context/IntegrationContextUtils.java | 20 +- .../context/IntegrationObjectSupport.java | 4 +- .../endpoint/AbstractEndpoint.java | 4 +- .../endpoint/AbstractPollingEndpoint.java | 10 +- .../integration/handler/DelayHandler.java | 1 - .../scheduling/CronSequenceGenerator.java | 328 ------------------ .../integration/scheduling/CronTrigger.java | 65 ---- .../scheduling/IntervalTrigger.java | 12 +- .../scheduling/PollerMetadata.java | 4 +- .../scheduling/SimpleTaskScheduler.java | 320 ----------------- .../integration/scheduling/TaskScheduler.java | 40 --- .../integration/scheduling/Trigger.java | 40 --- .../integration/util/ErrorHandler.java | 28 -- .../util/ErrorHandlingTaskExecutor.java | 5 +- .../aggregator/AggregatorEndpointTests.java | 22 +- .../aggregator/ResequencerTests.java | 12 +- .../ApplicationContextMessageBusTests.java | 8 +- ...dDispatcherConfigurationScenarioTests.java | 38 +- .../config/MessageBusParserTests.java | 2 +- .../PublishSubscribeChannelParserTests.java | 4 +- .../integration/config/StubTaskScheduler.java | 24 +- .../integration/config/TestErrorHandler.java | 6 +- ...figuringBeanFactoryPostProcessorTests.java | 21 +- ...nitionHandlerAwareEndpointParserTests.java | 47 ++- .../PollingConsumerEndpointTests.java | 36 +- .../scheduling/CronTriggerTests.java | 101 ++++-- .../integration/util/TestUtils.java | 26 +- 47 files changed, 399 insertions(+), 1070 deletions(-) delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSequenceGenerator.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronTrigger.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Trigger.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandler.java diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/DefaultConfigurationTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/DefaultConfigurationTests.java index ab07769c53..b206de47e7 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -44,23 +47,26 @@ public class DefaultConfigurationTests { @Test public void verifyErrorChannel() { - Object errorChannel = context.getBean("errorChannel"); + Object errorChannel = context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); assertNotNull(errorChannel); assertEquals(PublishSubscribeChannel.class, errorChannel.getClass()); } @Test public void verifyNullChannel() { - Object nullChannel = context.getBean("nullChannel"); + Object nullChannel = context.getBean(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); assertNotNull(nullChannel); assertEquals(NullChannel.class, nullChannel.getClass()); } @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/DefaultConfigurationTests.java b/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/DefaultConfigurationTests.java index 69f5e81d3d..dab26b32cc 100644 --- a/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/DefaultConfigurationTests.java b/org.springframework.integration.httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/DefaultConfigurationTests.java index 37df408aaa..d1bac26c6a 100644 --- a/org.springframework.integration.httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/DefaultConfigurationTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/DefaultConfigurationTests.java index d58150222c..fea0cc00ef 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/DefaultConfigurationTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/DefaultConfigurationTests.java index ea6e9cf723..ec2646a815 100644 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.rmi/src/test/java/org/springframework/integration/rmi/config/DefaultConfigurationTests.java b/org.springframework.integration.rmi/src/test/java/org/springframework/integration/rmi/config/DefaultConfigurationTests.java index 41feea3c19..1850185aef 100644 --- a/org.springframework.integration.rmi/src/test/java/org/springframework/integration/rmi/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.rmi/src/test/java/org/springframework/integration/rmi/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/DefaultConfigurationTests.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/DefaultConfigurationTests.java index 7cb4e1cee2..c19835ad0c 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java index f0857fbae0..3bcf05841f 100644 --- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java +++ b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -29,13 +29,13 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.Trigger; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Mark Fisher @@ -52,7 +52,7 @@ public class ByteStreamWritingMessageHandlerTests { private TestTrigger trigger = new TestTrigger(); - private SimpleTaskScheduler scheduler; + private ThreadPoolTaskScheduler scheduler; @Before @@ -61,9 +61,9 @@ public class ByteStreamWritingMessageHandlerTests { handler = new ByteStreamWritingMessageHandler(stream); this.channel = new QueueChannel(10); this.endpoint = new PollingConsumer(channel, handler); - scheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor()); + scheduler = new ThreadPoolTaskScheduler(); this.endpoint.setTaskScheduler(scheduler); - scheduler.start(); + scheduler.afterPropertiesSet(); trigger.reset(); endpoint.setTrigger(trigger); } @@ -235,8 +235,7 @@ public class ByteStreamWritingMessageHandlerTests { private volatile CountDownLatch latch = new CountDownLatch(1); - - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { + public Date nextExecutionTime(TriggerContext triggerContext) { if (!hasRun.getAndSet(true)) { return new Date(); } diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java index 7ce1072193..8a510f2d7e 100644 --- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java +++ b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -28,13 +28,13 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.Trigger; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Mark Fisher @@ -51,7 +51,7 @@ public class CharacterStreamWritingMessageHandlerTests { private TestTrigger trigger = new TestTrigger(); - private SimpleTaskScheduler scheduler; + private ThreadPoolTaskScheduler scheduler; @Before @@ -61,9 +61,9 @@ public class CharacterStreamWritingMessageHandlerTests { this.channel = new QueueChannel(10); trigger.reset(); this.endpoint = new PollingConsumer(channel, handler); - scheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor()); + scheduler = new ThreadPoolTaskScheduler(); this.endpoint.setTaskScheduler(scheduler); - scheduler.start(); + scheduler.afterPropertiesSet(); trigger.reset(); endpoint.setTrigger(trigger); } @@ -201,8 +201,7 @@ public class CharacterStreamWritingMessageHandlerTests { private volatile CountDownLatch latch = new CountDownLatch(1); - - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { + public Date nextExecutionTime(TriggerContext triggerContext) { if (!hasRun.getAndSet(true)) { return new Date(); } diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests.java index 59f90ff159..ade9a8461b 100644 --- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.test/src/main/java/org/springframework/integration/test/util/TestUtils.java b/org.springframework.integration.test/src/main/java/org/springframework/integration/test/util/TestUtils.java index c6d8c5ea44..c734d6cc0b 100644 --- a/org.springframework.integration.test/src/main/java/org/springframework/integration/test/util/TestUtils.java +++ b/org.springframework.integration.test/src/main/java/org/springframework/integration/test/util/TestUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -31,9 +31,8 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.scheduling.IntervalTrigger; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; /** @@ -75,11 +74,11 @@ public abstract class TestUtils { } public static TaskScheduler createTaskScheduler(int poolSize) { - ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - executor.setCorePoolSize(poolSize); - executor.setRejectedExecutionHandler(new CallerRunsPolicy()); - executor.afterPropertiesSet(); - return new SimpleTaskScheduler(executor); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(poolSize); + scheduler.setRejectedExecutionHandler(new CallerRunsPolicy()); + scheduler.afterPropertiesSet(); + return scheduler; } private static void registerBean(String beanName, Object bean, BeanFactory beanFactory) { diff --git a/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/MarshallingWebServiceInboundGateway.java b/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/MarshallingWebServiceInboundGateway.java index 5fd44fa7ed..d42bc73a1e 100644 --- a/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/MarshallingWebServiceInboundGateway.java +++ b/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/MarshallingWebServiceInboundGateway.java @@ -23,7 +23,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.gateway.SimpleMessagingGateway; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.TaskScheduler; import org.springframework.oxm.Marshaller; import org.springframework.oxm.Unmarshaller; import org.springframework.ws.server.endpoint.AbstractMarshallingPayloadEndpoint; diff --git a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/DefaultConfigurationTests.java b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/DefaultConfigurationTests.java index 785efaa871..94392f8d07 100644 --- a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration.xml/src/test/java/org/springframework/integration/xml/config/DefaultConfigurationTests.java b/org.springframework.integration.xml/src/test/java/org/springframework/integration/xml/config/DefaultConfigurationTests.java index ebf2b4f6dd..5835e4e241 100644 --- a/org.springframework.integration.xml/src/test/java/org/springframework/integration/xml/config/DefaultConfigurationTests.java +++ b/org.springframework.integration.xml/src/test/java/org/springframework/integration/xml/config/DefaultConfigurationTests.java @@ -22,11 +22,14 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,9 +61,12 @@ public class DefaultConfigurationTests { @Test public void verifyTaskScheduler() { - Object taskScheduler = context.getBean("taskScheduler"); - assertNotNull(taskScheduler); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java index 5cbb78a265..ca467711a7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java @@ -23,7 +23,6 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -41,8 +40,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; -import org.springframework.integration.scheduling.IntervalTrigger; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; /** @@ -218,8 +216,8 @@ public abstract class AbstractMessageBarrierHandler failedMessage = (t instanceof MessagingException) ? ((MessagingException) t).getFailedMessage() : null; MessageChannel errorChannel = this.resolveErrorChannel(failedMessage); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 786464fdb5..137ea37c28 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -20,8 +20,8 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.util.ErrorHandler; import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.scheduling.support.ErrorHandler; /** * A channel that sends Messages to each of its subscribers. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java index 0a89a5ef62..8642898e7d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java @@ -23,12 +23,11 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.BeanDefinitionStoreException; 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.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.core.SpringVersion; -import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.util.StringUtils; /** @@ -57,8 +56,15 @@ public class ApplicationEventMulticasterParser extends AbstractSingleBeanDefinit builder.addPropertyReference("taskExecutor", taskExecutorRef); } else { - TaskExecutor taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(1, 10, 0, "event-multicaster-"); - builder.addPropertyValue("taskExecutor", taskExecutor); + BeanDefinitionBuilder executorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"); + executorBuilder.addPropertyValue("corePoolSize", 1); + executorBuilder.addPropertyValue("maxPoolSize", 10); + executorBuilder.addPropertyValue("queueCapacity", 0); + executorBuilder.addPropertyValue("threadNamePrefix", "event-multicaster-"); + String executorBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName( + executorBuilder.getBeanDefinition(), parserContext.getRegistry()); + builder.addPropertyReference("taskExecutor", executorBeanName); } String springVersion = SpringVersion.getVersion(); if (springVersion != null && springVersion.startsWith("2")) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java index ec4573c279..5ff0adf8d7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java @@ -16,6 +16,8 @@ package org.springframework.integration.config.xml; +import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -27,7 +29,6 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.core.task.TaskExecutor; import org.springframework.integration.context.IntegrationContextUtils; /** @@ -111,12 +112,13 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce if (!registry.isBeanNameInUse(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) { if (logger.isInfoEnabled()) { logger.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME + - "' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created."); + "' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created."); } - TaskExecutor taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(2, 100, 0, "task-scheduler-"); BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".scheduling.SimpleTaskScheduler"); - schedulerBuilder.addConstructorArgValue(taskExecutor); + "org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler"); + schedulerBuilder.addPropertyValue("poolSize", 10); + schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-"); + schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy()); BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition( IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.MessagePublishingErrorHandler"); errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index 96a9dc3971..73e56a9700 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -16,16 +16,11 @@ package org.springframework.integration.context; -import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; - import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.scheduling.PollerMetadata; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * @author Mark Fisher @@ -70,17 +65,4 @@ public abstract class IntegrationContextUtils { return (T) bean; } - public static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int coreSize, int maxSize, int queueCapacity, String threadPrefix) { - ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - executor.setCorePoolSize(coreSize); - executor.setMaxPoolSize(maxSize); - executor.setQueueCapacity(queueCapacity); - if (StringUtils.hasText(threadPrefix)) { - executor.setThreadFactory(new CustomizableThreadFactory(threadPrefix)); - } - executor.setRejectedExecutionHandler(new CallerRunsPolicy()); - executor.afterPropertiesSet(); - return executor; - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 31c560cd7a..530b02a566 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -24,7 +24,7 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index b38843f138..8dd5d989ae 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -22,7 +22,7 @@ import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.TaskScheduler; /** * The base class for Message Endpoint implementations. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index fea979e1ab..d7ba23c7bf 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -28,9 +28,9 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.MessagePublishingErrorHandler; -import org.springframework.integration.scheduling.Trigger; -import org.springframework.integration.util.ErrorHandler; import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.support.ErrorHandler; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; @@ -223,8 +223,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private boolean innerPoll() { TransactionTemplate txTemplate = getTransactionTemplate(); if (txTemplate != null) { - return (Boolean) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { + return txTemplate.execute(new TransactionCallback() { + public Boolean doInTransaction(TransactionStatus status) { return doPoll(); } }); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java index 58ac2147a2..45c158dbc1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -40,7 +40,6 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandler; -import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.util.Assert; /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSequenceGenerator.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSequenceGenerator.java deleted file mode 100644 index fa0c4b1509..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSequenceGenerator.java +++ /dev/null @@ -1,328 +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.BitSet; -import java.util.Calendar; -import java.util.Date; -import java.util.GregorianCalendar; - -import org.springframework.util.StringUtils; - -/** - * Date sequence generator for a Crontab pattern allowing - * client to specify a pattern that the sequence matches. The pattern is a list - * of 6 single space separated fields representing (second, minute, hour, day, - * month, weekday). Month and weekday names can be given as the first three - * letters of the English names.
- *
- * - * Example patterns - *
    - *
  • "0 0 * * * *" = the top of every hour of every day.
    • - *
  • "*/10 * * * * *" = every ten seconds.
    • - *
  • "0 0 8-10 * * *" = 8, 9 and 10 o'clock of every day.
    • - *
  • "0 0 8-10/30 * * *" = 8:00, 8:30, 9:00, 9:30 and 10 o'clock every day.
    • - *
  • "0 0 9-17 * * MON-FRI" = on the hour nine-to-five weekdays
    • - *
  • "0 0 0 25 12 ?" = every Christmas Day at midnight
    • - *
- * - * @author Dave Syer - */ -class CronSequenceGenerator { - - private final BitSet seconds = new BitSet(60); - - private final BitSet minutes = new BitSet(60); - - private final BitSet hours = new BitSet(24); - - private final BitSet daysOfWeek = new BitSet(7); - - private final BitSet daysOfMonth = new BitSet(31); - - private final BitSet months = new BitSet(12); - - private final String pattern; - - /** - * Construct a {@link CronSequenceGenerator} from the pattern provided. - * - * @param pattern a space separated list of time fields - * - * @throws IllegalArgumentException if the pattern cannot be parsed - */ - public CronSequenceGenerator(String pattern) throws IllegalArgumentException { - this.pattern = pattern; - parse(pattern); - } - - /** - * Get the next {@link Date} in the sequence matching the Cron pattern and - * after the value provided. The return value will have a whole number of - * seconds, and will be after the input value. - * - * @param date a seed value - * @return the next value matching the pattern - */ - public Date next(Date date) { - - Calendar calendar = new GregorianCalendar(); - calendar.setTime(date); - - // Truncate to the next whole second - calendar.add(Calendar.SECOND, 1); - calendar.set(Calendar.MILLISECOND, 0); - - int second = calendar.get(Calendar.SECOND); - second = findNext(seconds, second, 60, calendar, Calendar.SECOND); - - int minute = calendar.get(Calendar.MINUTE); - minute = findNext(minutes, minute, 60, calendar, Calendar.MINUTE, Calendar.SECOND); - - int hour = calendar.get(Calendar.HOUR_OF_DAY); - hour = findNext(hours, hour, 24, calendar, Calendar.HOUR_OF_DAY, Calendar.MINUTE, Calendar.SECOND); - - int dayOfWeek = calendar.get(Calendar.DAY_OF_WEEK); - int dayOfMonth = calendar.get(Calendar.DAY_OF_MONTH); - dayOfMonth = findNextDay(calendar, daysOfMonth, dayOfMonth, daysOfWeek, dayOfWeek, 366); - - int month = calendar.get(Calendar.MONTH); - month = findNext(months, month, 12, calendar, Calendar.MONTH, Calendar.DAY_OF_MONTH, Calendar.HOUR_OF_DAY, - Calendar.MINUTE, Calendar.SECOND); - - return calendar.getTime(); - - } - - /** - * @param calendar - * @return - */ - private int findNextDay(Calendar calendar, BitSet daysOfMonth, int dayOfMonth, BitSet daysOfWeek, int dayOfWeek, - int max) { - int count = 0; - // the DAY_OF_WEEK values in java.util.Calendar start with 1 (Sunday), - // but in the cron pattern, they start with 0, so we subtract 1 here - while ((!daysOfMonth.get(dayOfMonth) || !daysOfWeek.get(dayOfWeek-1)) && count++ < max) { - calendar.add(Calendar.DAY_OF_MONTH, 1); - dayOfMonth = calendar.get(Calendar.DAY_OF_MONTH); - dayOfWeek = calendar.get(Calendar.DAY_OF_WEEK); - reset(calendar, Calendar.HOUR_OF_DAY, Calendar.MINUTE, Calendar.SECOND); - } - if (count > max) { - throw new IllegalStateException("Overflow in day for expression=" + pattern); - } - return dayOfMonth; - } - - /** - * Search the bits provided for the next set bit after the value provided, - * and reset the calendar. - * - * @param bits a {@link BitSet} representing the allowed values of the field - * @param value the current value of the field - * @param max the largest value that the field can have - * @param calendar the calendar to increment as we move through the bits - * @param field the field to increment in the calendar (@see - * {@link Calendar} for the static constants defining valid fields) - * @param lowerOrders the Calendar field ids that should be reset (i.e. the - * ones of lower significance than the field of interest) - * - * @return the value of the calendar field that is next in the sequence - */ - private int findNext(BitSet bits, int value, int max, Calendar calendar, int field, int... lowerOrders) { - int nextValue = bits.nextSetBit(value); - //roll over if needed - if (nextValue == -1) { - calendar.add(field, max - value); - nextValue = bits.nextSetBit(0); - } - if (nextValue != value) { - calendar.set(field, nextValue); - reset(calendar, lowerOrders); - } - return nextValue; - } - - /** - * Reset the calendar setting all the fields provided to zero. - * - * @param calendar - * @param fields - */ - private void reset(Calendar calendar, int... fields) { - for (int field : fields) { - calendar.set(field, 0); - } - } - - /** - * @param expression - */ - private void parse(String expression) throws IllegalArgumentException { - String[] fields = StringUtils.tokenizeToStringArray(expression, " "); - if (fields.length != 6) { - throw new IllegalArgumentException(String.format("" - + "cron expression must consist of 6 fields (found %d in %s)", fields.length, expression)); - } - setNumberHits(seconds, fields[0], 60); - setNumberHits(minutes, fields[1], 60); - setNumberHits(hours, fields[2], 24); - setDaysOfMonth(daysOfMonth, fields[3], 31); - setNumberHits(months, replaceOrdinals(fields[4], "JAN,FEB,MAR,APR,MAY,JUN,JUL,AUG,SEP,OCT,NOV,DEC"), 12); - setDays(daysOfWeek, replaceOrdinals(fields[5], "SUN,MON,TUE,WED,THU,FRI,SAT"), 8); - if (daysOfWeek.get(7)) { - // Sunday can be represented as 0 or 7 - daysOfWeek.set(0); - daysOfWeek.clear(7); - } - } - - /** - * Replace the values in the commaSeparatedList (case insensitive) with - * their index in the list. - * - * @param value - * @param commaSeparatedList - * @return a new string with the values from the list replaced - */ - private String replaceOrdinals(String value, String commaSeparatedList) { - String[] list = StringUtils.commaDelimitedListToStringArray(commaSeparatedList); - for (int i = 0; i < list.length; i++) { - String item = list[i].toUpperCase(); - value = StringUtils.replace(value.toUpperCase(), item, "" + i); - } - return value; - } - - /** - * @param bits - * @param field - * @param max - */ - private void setDaysOfMonth(BitSet bits, String field, int max) { - // Days of month start with 1 (in Cron and Calendar) so add one - setDays(bits, field, max+1); - // ... and remove it from the front - bits.clear(0); - } - - /** - * @param bits - * @param field - * @param max - */ - private void setDays(BitSet bits, String field, int max) { - if (field.contains("?")) { - field = "*"; - } - setNumberHits(bits, field, max); - } - - /** - * @param bits - * @param value - * @param max - * @return - */ - private void setNumberHits(BitSet bits, String value, int max) { - - String[] fields = StringUtils.delimitedListToStringArray(value, ","); - - for (String field : fields) { - - if (!field.contains("/")) { - - // Not an incrementer so it must be a range (possibly empty) - int[] range = getRange(field, max); - bits.set(range[0], range[1] + 1); - - } - else { - - String[] split = StringUtils.delimitedListToStringArray(field, "/"); - if (split.length > 2) { - throw new IllegalArgumentException("Incrementer has more than two fields: " + field); - } - int[] range = getRange(split[0], max); - if (!split[0].contains("-")) { - range[1] = max - 1; - } - int delta = Integer.valueOf(split[1]); - for (int i = range[0]; i <= range[1]; i += delta) { - bits.set(i); - } - - } - } - } - - /** - * @param field - * @return - */ - private int[] getRange(String field, int max) { - int[] result = new int[2]; - if (field.contains("*")) { - result[0] = 0; - result[1] = max-1; - return result; - } - if (!field.contains("-")) { - result[0] = result[1] = Integer.valueOf(field); - } - else { - String[] split = StringUtils.delimitedListToStringArray(field, "-"); - if (split.length > 2) { - throw new IllegalArgumentException("Range has more than two fields: " + field); - } - result[0] = Integer.valueOf(split[0]); - result[1] = Integer.valueOf(split[1]); - } - return result; - } - - /** - * @see Object#equals(Object) - */ - @Override - public boolean equals(Object obj) { - if (!(obj instanceof CronSequenceGenerator)) { - return false; - } - CronSequenceGenerator cron = (CronSequenceGenerator) obj; - return cron.months.equals(months) && cron.daysOfMonth.equals(daysOfMonth) && cron.daysOfWeek.equals(daysOfWeek) - && cron.hours.equals(hours) && cron.minutes.equals(minutes) && cron.seconds.equals(seconds); - } - - /** - * @see Object#hashCode() - */ - @Override - public int hashCode() { - return 37 + 17 * months.hashCode() + 29 * daysOfMonth.hashCode() + 37 * daysOfWeek.hashCode() + 41 - * hours.hashCode() + 53 * minutes.hashCode() + 61 * seconds.hashCode(); - } - - @Override - public String toString() { - return getClass().getSimpleName() + ": " + pattern; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronTrigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronTrigger.java deleted file mode 100644 index 3ad1e66cc3..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronTrigger.java +++ /dev/null @@ -1,65 +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.Date; - -/** - * A trigger that uses a cron expression. See {@link CronSequenceGenerator} - * for a detailed description of the expression pattern syntax. - * - * @author Mark Fisher - */ -public class CronTrigger implements Trigger { - - private final CronSequenceGenerator cronSequenceGenerator; - - - /** - * Create a trigger for the given cron expression. - * See {@link CronSequenceGenerator}. - */ - public CronTrigger(String expression) throws IllegalArgumentException { - this.cronSequenceGenerator = new CronSequenceGenerator(expression); - } - - - /** - * Return the next time a task should run. Determined by consulting this - * trigger's cron expression compared with the lastCompleteTime. If the - * lastCompleteTime is null, the current time is used. - */ - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { - Date date = (lastCompleteTime != null) ? lastCompleteTime : new Date(); - return this.cronSequenceGenerator.next(date); - } - - @Override - public boolean equals(Object other) { - if (other == null || !(other instanceof CronTrigger)) { - return false; - } - return this.cronSequenceGenerator.equals( - ((CronTrigger) other).cronSequenceGenerator); - } - - @Override - public int hashCode() { - return this.cronSequenceGenerator.hashCode(); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java index 98b6160a68..09f2ef8836 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -19,6 +19,8 @@ package org.springframework.integration.scheduling; import java.util.Date; import java.util.concurrent.TimeUnit; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; import org.springframework.util.Assert; /** @@ -78,14 +80,14 @@ public class IntervalTrigger implements Trigger { /** * Returns the next time a task should run. */ - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { - if (lastScheduledRunTime == null) { + public Date nextExecutionTime(TriggerContext triggerContext) { + if (triggerContext.lastScheduledExecutionTime() == null) { return new Date(System.currentTimeMillis() + this.initialDelay); } else if (this.fixedRate) { - return new Date(lastScheduledRunTime.getTime() + this.interval); + return new Date(triggerContext.lastScheduledExecutionTime().getTime() + this.interval); } - return new Date(lastCompleteTime.getTime() + this.interval); + return new Date(triggerContext.lastCompletionTime().getTime() + this.interval); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index 9162230d1e..ca0e0f364a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -19,8 +19,8 @@ package org.springframework.integration.scheduling; import java.util.List; import org.aopalliance.aop.Advice; - import org.springframework.core.task.TaskExecutor; +import org.springframework.scheduling.Trigger; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java deleted file mode 100644 index dafd57ae5c..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java +++ /dev/null @@ -1,320 +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.Collections; -import java.util.Date; -import java.util.Set; -import java.util.TreeSet; -import java.util.concurrent.DelayQueue; -import java.util.concurrent.Delayed; -import java.util.concurrent.FutureTask; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; -import java.util.concurrent.locks.ReentrantLock; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.context.ApplicationEvent; -import org.springframework.context.ApplicationListener; -import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.channel.BeanFactoryChannelResolver; -import org.springframework.integration.channel.MessagePublishingErrorHandler; -import org.springframework.integration.util.ErrorHandler; -import org.springframework.scheduling.SchedulingException; -import org.springframework.util.Assert; - -/** - * An implementation of {@link TaskScheduler} that delegates to any instance - * of {@link TaskExecutor}. - * - *

This class implements ApplicationListener and provides an {@link #autoStartup} - * property. If true, the scheduler will start automatically upon - * receiving the {@link ContextRefreshedEvent}. Otherwise, it will require an - * explicit invocation of its {@link #start()} method. The default value is - * true. To require explicit startup, provide a value of - * false to the {@link #setAutoStartup(boolean)} method. - * - * @author Mark Fisher - * @author Marius Bogoevici - */ -public class SimpleTaskScheduler implements TaskScheduler, BeanFactoryAware, ApplicationListener, DisposableBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final TaskExecutor executor; - - private volatile boolean autoStartup = true; - - private volatile ErrorHandler errorHandler; - - private volatile SchedulerTask schedulerTask = null; - - private final DelayQueue> scheduledTasks = new DelayQueue>(); - - private final Set> executingTasks = Collections.synchronizedSet(new TreeSet>()); - - private volatile boolean running; - - private final ReentrantLock lifecycleLock = new ReentrantLock(); - - - public SimpleTaskScheduler(TaskExecutor executor) { - Assert.notNull(executor, "executor must not be null"); - this.executor = executor; - } - - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - public void setErrorHandler(ErrorHandler errorHandler) { - this.errorHandler = errorHandler; - } - - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - if (this.errorHandler == null) { - this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory)); - } - } - - public final ScheduledFuture schedule(Runnable task, Trigger trigger) { - Assert.notNull(task, "task must not be null"); - TriggeredTask triggeredTask = new TriggeredTask(task, trigger); - return this.schedule(triggeredTask, null, null); - } - - private ScheduledFuture schedule(TriggeredTask triggeredTask, Date lastScheduledRunTime, Date lastCompleteTime) { - Date nextRunTime = triggeredTask.trigger.getNextRunTime(lastScheduledRunTime, lastCompleteTime); - if (nextRunTime != null) { - triggeredTask.setScheduledTime(nextRunTime); - this.scheduledTasks.offer(triggeredTask); - } - return triggeredTask; - } - - - // Lifecycle implementation - - public final boolean isRunning() { - this.lifecycleLock.lock(); - try { - return this.running; - } - finally { - this.lifecycleLock.unlock(); - } - } - - public final void start() { - this.lifecycleLock.lock(); - try { - if (!this.running) { - this.executor.execute(this.schedulerTask = new SchedulerTask()); - this.running = true; - if (logger.isInfoEnabled()) { - logger.info("started " + this); - } - } - } - finally { - this.lifecycleLock.unlock(); - } - } - - public final void stop() { - this.lifecycleLock.lock(); - try { - if (this.running) { - this.schedulerTask.deactivate(); - Thread executingThread = this.schedulerTask.executingThread.get(); - if (executingThread != null) { - executingThread.interrupt(); - } - this.scheduledTasks.clear(); - synchronized (this.executingTasks) { - for (TriggeredTask task : this.executingTasks) { - task.cancel(true); - } - this.executingTasks.clear(); - } - this.schedulerTask = null; - this.running = false; - if (logger.isInfoEnabled()) { - logger.info("stopped " + this); - } - } - } - finally { - this.lifecycleLock.unlock(); - } - } - - public final void onApplicationEvent(ApplicationEvent event) { - if (event instanceof ContextRefreshedEvent && this.autoStartup) { - this.start(); - } - } - - public void destroy() throws Exception { - this.stop(); - if (this.executor instanceof DisposableBean) { - if (logger.isInfoEnabled()) { - logger.info("shutting down TaskExecutor"); - } - ((DisposableBean) this.executor).destroy(); - } - } - - public boolean prefersShortLivedTasks() { - return true; - } - - public void execute(Runnable task) { - this.executor.execute(task); - } - - - private class SchedulerTask implements Runnable { - - private final AtomicReference executingThread = new AtomicReference(); - - private volatile boolean active = true; - - public void run() { - if (!this.executingThread.compareAndSet(null, Thread.currentThread())) { - throw new SchedulingException("The SchedulerTask is already running."); - } - while (this.active) { - try { - TriggeredTask task = SimpleTaskScheduler.this.scheduledTasks.take(); - //if this thread is not active anymore, clear - if (this.active) { - SimpleTaskScheduler.this.executor.execute(task); - } - else { - scheduledTasks.offer(task); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } - } - this.executingThread.set(null); - } - - public void deactivate() { - this.active = false; - } - } - - - /** - * Wrapper class that enables rescheduling of a task based on a Trigger. - */ - private class TriggeredTask extends FutureTask implements Delayed, ScheduledFuture { - - private final Trigger trigger; - - private volatile Date scheduledTime; - - - public TriggeredTask(Runnable task, Trigger trigger) { - super(new ErrorHandlingRunnableWrapper(task), null); - this.trigger = trigger; - } - - - public void setScheduledTime(Date scheduledTime) { - this.scheduledTime = scheduledTime; - } - - public void run() { - SimpleTaskScheduler.this.executingTasks.add(this); - super.runAndReset(); - SimpleTaskScheduler.this.executingTasks.remove(this); - if (SimpleTaskScheduler.this.isRunning() && !this.isCancelled()) { - SimpleTaskScheduler.this.schedule(this, this.scheduledTime, new Date()); - } - } - - public int compareTo(Delayed other) { - long thisDelay = this.getDelay(TimeUnit.MILLISECONDS); - long otherDelay = other.getDelay(TimeUnit.MILLISECONDS); - if (thisDelay < otherDelay) { - return -1; - } - if (thisDelay == otherDelay) { - return 0; - } - return 1; - } - - public long getDelay(TimeUnit unit) { - long now = new Date().getTime(); - long scheduled = (this.scheduledTime != null) ? this.scheduledTime.getTime() : now; - return (scheduled > now) ? unit.convert(scheduled - now, TimeUnit.MILLISECONDS) : 0; - } - - public synchronized boolean cancel(boolean mayInterruptIfRunning) { - if (!this.isCancelled()) { - SimpleTaskScheduler.this.scheduledTasks.remove(this); - } - return super.cancel(mayInterruptIfRunning); - } - } - - - /** - * Wrapper that catches any Throwable thrown by a target task and - * delegates to the {@link ErrorHandler} if available. If no error handler - * has been configured, the error will be logged at error-level. - */ - private class ErrorHandlingRunnableWrapper implements Runnable { - - private final Runnable target; - - - public ErrorHandlingRunnableWrapper(Runnable target) { - this.target = target; - } - - - public void run() { - try { - this.target.run(); - } - catch (Throwable t) { - if (SimpleTaskScheduler.this.errorHandler != null) { - SimpleTaskScheduler.this.errorHandler.handle(t); - } - else if (logger.isErrorEnabled()) { - logger.error("Error occurred in task but no 'errorHandler' is available.", t); - } - } - } - } - -} 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 deleted file mode 100644 index 29ea9b8785..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java +++ /dev/null @@ -1,40 +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.ScheduledFuture; - -import org.springframework.context.Lifecycle; - -/** - * Base interface for scheduling tasks. - * - * @author Mark Fisher - * @author Marius Bogoevici - * @author Iwein Fuld - */ -public interface TaskScheduler extends Lifecycle { - - /** - * Schedules a task for multiple executions according to a Trigger. - * - * @param task Task to be run multiple times - * @param trigger Trigger that determines at which times the task should be run - */ - ScheduledFuture schedule(Runnable task, Trigger trigger); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Trigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Trigger.java deleted file mode 100644 index efa12bea13..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Trigger.java +++ /dev/null @@ -1,40 +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.Date; - -/** - * A strategy for providing the next time a task should run. - * - * @author Mark Fisher - */ -public interface Trigger { - - /** - * Returns the next time that a task should run or null if the - * task should not run again. - * - * @param lastScheduledRunTime last time the relevant task was scheduled to - * run, or null if it has never been scheduled - * @param lastCompleteTime last time the relevant task finished or - * null if it did not run to completion - * @return next time that a task should run - */ - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandler.java deleted file mode 100644 index f24fd60f6c..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandler.java +++ /dev/null @@ -1,28 +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.util; - -/** - * Strategy for handling a {@link Throwable}. - * - * @author Mark Fisher - */ -public interface ErrorHandler { - - void handle(Throwable t); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java b/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java index 534ec3a7db..efd885cb81 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -17,6 +17,7 @@ package org.springframework.integration.util; import org.springframework.core.task.TaskExecutor; +import org.springframework.scheduling.support.ErrorHandler; import org.springframework.util.Assert; /** @@ -49,7 +50,7 @@ public class ErrorHandlingTaskExecutor implements TaskExecutor { task.run(); } catch (Throwable t) { - errorHandler.handle(t); + errorHandler.handleError(t); } } }); 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 a4590f6b47..1fd601a9aa 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -16,6 +16,11 @@ package org.springframework.integration.aggregator; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -23,10 +28,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.After; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -38,8 +39,7 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Mark Fisher @@ -49,7 +49,7 @@ public class AggregatorEndpointTests { private TaskExecutor taskExecutor; - private TaskScheduler taskScheduler; + private ThreadPoolTaskScheduler taskScheduler; private AbstractMessageAggregator aggregator; @@ -57,10 +57,10 @@ public class AggregatorEndpointTests { @Before public void configureAggregator() { this.taskExecutor = new SimpleAsyncTaskExecutor(); - this.taskScheduler = new SimpleTaskScheduler(taskExecutor); + this.taskScheduler = new ThreadPoolTaskScheduler(); + this.taskScheduler.afterPropertiesSet(); this.aggregator = new TestAggregator(); this.aggregator.setTaskScheduler(this.taskScheduler); - this.taskScheduler.start(); this.aggregator.start(); } @@ -342,7 +342,7 @@ public class AggregatorEndpointTests { @After public void stopTaskScheduler() { - this.taskScheduler.stop(); + this.taskScheduler.destroy(); this.aggregator.stop(); } 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 357eee3621..330c0ef7d8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -25,12 +25,13 @@ import static org.junit.Assert.assertThat; import org.junit.After; import org.junit.Before; import org.junit.Test; + import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.integration.util.TestUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Marius Bogoevici @@ -40,7 +41,7 @@ public class ResequencerTests { private Resequencer resequencer; - private TaskScheduler taskScheduler; + private ThreadPoolTaskScheduler taskScheduler; @Before @@ -48,7 +49,7 @@ public class ResequencerTests { this.resequencer = new Resequencer(); this.taskScheduler = TestUtils.createTaskScheduler(10); this.resequencer.setTaskScheduler(taskScheduler); - taskScheduler.start(); + this.taskScheduler.afterPropertiesSet(); this.resequencer.start(); } @@ -250,6 +251,7 @@ public class ResequencerTests { @After public void stopTaskScheduler() { this.resequencer.stop(); - this.taskScheduler.stop(); + this.taskScheduler.destroy(); } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index dbad270fae..7c3aeff7e9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -27,9 +27,6 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.BeanFactoryChannelResolver; -import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; @@ -191,9 +188,6 @@ public class ApplicationContextMessageBusTests { channelAdapter.setTrigger(new IntervalTrigger(1000)); channelAdapter.setOutputChannel(outputChannel); context.registerEndpoint("testChannel", channelAdapter); - ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); - MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(channelResolver); - errorHandler.setDefaultErrorChannel(errorChannel); context.refresh(); latch.await(2000, TimeUnit.MILLISECONDS); Message message = errorChannel.receive(5000); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java index 82c0c3317b..4d83b89381 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.channel; import static org.junit.Assert.assertFalse; @@ -21,9 +22,9 @@ import static org.mockito.Matchers.anyObject; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.never; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -37,6 +38,7 @@ import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; import org.mockito.runners.MockitoJUnit44Runner; import org.mockito.stubbing.Answer; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -52,22 +54,31 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; */ @RunWith(MockitoJUnit44Runner.class) public class MixedDispatcherConfigurationScenarioTests { + private static final int TOTAL_EXECUTIONS = 40; + private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor(); + private CountDownLatch allDone; private CountDownLatch start; private AtomicBoolean failed; + @Mock private List exceptionRegistry; + private ApplicationContext ac; + @Mock private MessageHandler handlerA; + @Mock private MessageHandler handlerB; + @Mock private MessageHandler handlerC; + @Mock - private Message message; + private Message message; @Before public void initialize() throws Exception { @@ -145,7 +156,7 @@ public class MixedDispatcherConfigurationScenarioTests { dispatcher.addHandler(handlerA); dispatcher.addHandler(handlerB); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { RuntimeException e = new RuntimeException(); allDone.countDown(); @@ -155,7 +166,7 @@ public class MixedDispatcherConfigurationScenarioTests { } }).when(handlerA).handleMessage(message); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { allDone.countDown(); return null; @@ -273,7 +284,7 @@ public class MixedDispatcherConfigurationScenarioTests { final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS); final Message message = this.message; final AtomicBoolean failed = new AtomicBoolean(false); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { failed.set(true); RuntimeException e = new RuntimeException(); @@ -282,13 +293,13 @@ public class MixedDispatcherConfigurationScenarioTests { throw e; } }).when(handlerA).handleMessage(message); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { allDone.countDown(); return null; } }).when(handlerB).handleMessage(message); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { allDone.countDown(); return null; @@ -403,7 +414,7 @@ public class MixedDispatcherConfigurationScenarioTests { dispatcher.addHandler(handlerB); dispatcher.addHandler(handlerC); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { RuntimeException e = new RuntimeException(); failed.set(true); @@ -411,13 +422,13 @@ public class MixedDispatcherConfigurationScenarioTests { throw e; } }).when(handlerA).handleMessage(message); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { allDone.countDown(); return null; } }).when(handlerB).handleMessage(message); - doAnswer(new Answer() { + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { allDone.countDown(); return null; @@ -439,7 +450,7 @@ public class MixedDispatcherConfigurationScenarioTests { } start.countDown(); allDone.await(); - // Mockiti threads migt still be lingering, so wait till they are all finished to avoid + // Mockito threads might still be lingering, so wait till they are all finished to avoid // Mockito concurrency this.waitTillAllFinished((SimpleAsyncTaskExecutor) ac.getBean("taskExecutor")); @@ -447,6 +458,7 @@ public class MixedDispatcherConfigurationScenarioTests { verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerC, never()).handleMessage(message); } + /** * * @param taskExecutor @@ -454,7 +466,6 @@ public class MixedDispatcherConfigurationScenarioTests { private void waitTillAllFinished(SimpleAsyncTaskExecutor taskExecutor){ for (int i = 0; i < 1000; i++) { if (taskExecutor.getThreadGroup().activeCount() == 0){ - //System.out.println("breaking"); break; } try { @@ -462,4 +473,5 @@ public class MixedDispatcherConfigurationScenarioTests { } catch (InterruptedException e) {/*ignore*/} } } + } 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 032423988c..6c3abfed28 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 @@ -30,7 +30,7 @@ import org.springframework.core.SpringVersion; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java index aa09fb52df..4bc8e2979a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -29,8 +29,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.util.ErrorHandler; import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.scheduling.support.ErrorHandler; /** * @author Mark Fisher diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/StubTaskScheduler.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/StubTaskScheduler.java index 19945943e0..6998eae88c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/StubTaskScheduler.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/StubTaskScheduler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -16,10 +16,11 @@ package org.springframework.integration.config; +import java.util.Date; import java.util.concurrent.ScheduledFuture; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.integration.scheduling.Trigger; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.Trigger; /** * @author Mark Fisher @@ -30,21 +31,24 @@ public class StubTaskScheduler implements TaskScheduler { return null; } - public void execute(Runnable task) { + public ScheduledFuture schedule(Runnable task, Date startTime) { + return null; } - public boolean prefersShortLivedTasks() { - return true; + public ScheduledFuture scheduleAtFixedRate(Runnable task, long period) { + return null; } - public boolean isRunning() { - return false; + public ScheduledFuture scheduleAtFixedRate(Runnable task, Date startTime, long period) { + return null; } - public void start() { + public ScheduledFuture scheduleWithFixedDelay(Runnable task, long delay) { + return null; } - public void stop() { + public ScheduledFuture scheduleWithFixedDelay(Runnable task, Date startTime, long delay) { + return null; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestErrorHandler.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestErrorHandler.java index 9572249841..df8efe4bed 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestErrorHandler.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -16,7 +16,7 @@ package org.springframework.integration.config; -import org.springframework.integration.util.ErrorHandler; +import org.springframework.scheduling.support.ErrorHandler; /** * @author Mark Fisher @@ -26,7 +26,7 @@ public class TestErrorHandler implements ErrorHandler { private volatile Throwable lastError; - public void handle(Throwable t) { + public void handleError(Throwable t) { this.lastError = t; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessorTests.java index 6a46f85516..764d5f0f0b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessorTests.java @@ -17,18 +17,19 @@ package org.springframework.integration.config.xml; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -45,20 +46,26 @@ public class DefaultConfiguringBeanFactoryPostProcessorTests { @Test public void errorChannelRegistered() { - MessageChannel errorChannel = (MessageChannel) context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); + Object errorChannel = context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); + assertNotNull(errorChannel); assertEquals(PublishSubscribeChannel.class, errorChannel.getClass()); } @Test public void nullChannelRegistered() { - MessageChannel nullChannel = (MessageChannel) context.getBean(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); + Object nullChannel = context.getBean(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); + assertNotNull(nullChannel); assertEquals(NullChannel.class, nullChannel.getClass()); } @Test public void taskSchedulerRegistered() { - TaskScheduler taskScheduler = (TaskScheduler) context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); - assertEquals(SimpleTaskScheduler.class, taskScheduler.getClass()); + Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass()); + Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler"); + assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass()); + Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel"); + assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java index fef4a9ab13..310a6edda4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.config.xml; import static org.junit.Assert.assertEquals; @@ -28,6 +29,7 @@ import junit.framework.Assert; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.xml.XmlBeanDefinitionReader; @@ -47,10 +49,9 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; + /** - * * @author Oleg Zhurakousky - * */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @@ -70,16 +71,19 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { String configProperty = testConfigurations.getProperty("splitter-inner-success-with-poller"); this.bootStrap(configProperty); } + @Test public void testInnerSplitterDefinitionSuccessWithPollerReversedOrder(){ String configProperty = testConfigurations.getProperty("splitter-inner-success-with-poller-reversed-order"); this.bootStrap(configProperty); } + @Test public void testRefSplitterDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("splitter-ref-success"); this.testSplitterDefinitionSuccess(configProperty); } + @Test(expected=BeanDefinitionStoreException.class) public void testInnerSplitterDefinitionFailureRefAndInner(){ String xmlConfig = testConfigurations.getProperty("splitter-failure-refAndBean"); @@ -91,6 +95,7 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { String configProperty = testConfigurations.getProperty("transformer-inner-success"); this.testTransformerDefinitionSuccess(configProperty); } + @Test public void testRefTransformerDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("transformer-ref-success"); @@ -108,6 +113,7 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { String configProperty = testConfigurations.getProperty("router-inner-success"); this.testRouterDefinitionSuccess(configProperty); } + @Test public void testRefRouterDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("router-ref-success"); @@ -125,11 +131,13 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { String configProperty = testConfigurations.getProperty("sa-inner-success"); this.testSADefinitionSuccess(configProperty); } + @Test public void testRefSADefinitionSuccess(){ String configProperty = testConfigurations.getProperty("sa-ref-success"); this.testSADefinitionSuccess(configProperty); } + @Test(expected=BeanDefinitionStoreException.class) public void testInnerSADefinitionFailureRefAndInner(){ String xmlConfig = testConfigurations.getProperty("sa-failure-refAndBean"); @@ -141,21 +149,25 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { String configProperty = testConfigurations.getProperty("aggregator-inner-success"); this.testAggregatorDefinitionSuccess(configProperty); } + @Test public void testInnerConcurrentAggregatorDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("aggregator-inner-concurrent-success"); this.testAggregatorDefinitionSuccess(configProperty); } + @Test public void testInnerConcurrentAggregatorDefinitionSuccessReorderBeanPoller(){ String configProperty = testConfigurations.getProperty("aggregator-inner-concurrent-success-reorder-bean-poller"); this.testAggregatorDefinitionSuccess(configProperty); } + @Test public void testRefAggregatorDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("aggregator-ref-success"); this.testAggregatorDefinitionSuccess(configProperty); } + @Test(expected=BeanDefinitionStoreException.class) public void testInnerAggregatorDefinitionFailureRefAndInner(){ String xmlConfig = testConfigurations.getProperty("aggregator-failure-refAndBean"); @@ -171,9 +183,9 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { @Test public void testRefFilterDefinitionSuccess(){ String configProperty = testConfigurations.getProperty("filter-ref-success"); - System.out.println(configProperty); this.testFilterDefinitionSuccess(configProperty); } + @Test(expected=BeanDefinitionStoreException.class) public void testInnerFilterDefinitionFailureRefAndInner(){ String xmlConfig = testConfigurations.getProperty("filter-failure-refAndBean"); @@ -184,8 +196,8 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { ApplicationContext ac = this.bootStrap(configProperty); EventDrivenConsumer splitter = (EventDrivenConsumer) ac.getBean("testSplitter"); Assert.assertNotNull(splitter); - MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload(new String[]{"One","Two"}); - Message inMessage = inChannelMessageBuilder.build(); + MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload(new String[]{"One","Two"}); + Message inMessage = inChannelMessageBuilder.build(); MessageChannel inChannel = (MessageChannel) ac.getBean("inChannel"); inChannel.send(inMessage); PollableChannel outChannel = (PollableChannel) ac.getBean("outChannel"); @@ -198,20 +210,21 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { ApplicationContext ac = this.bootStrap(configProperty); EventDrivenConsumer transformer = (EventDrivenConsumer) ac.getBean("testTransformer"); Assert.assertNotNull(transformer); - MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload(new String[]{"One","Two"}); - Message inMessage = inChannelMessageBuilder.build(); + MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload(new String[]{"One","Two"}); + Message inMessage = inChannelMessageBuilder.build(); DirectChannel inChannel = (DirectChannel) ac.getBean("inChannel"); inChannel.send(inMessage); PollableChannel outChannel = (PollableChannel) ac.getBean("outChannel"); String payload = (String) outChannel.receive().getPayload(); Assert.assertTrue(payload.equals("One,Two")); } + private void testRouterDefinitionSuccess(String configProperty){ ApplicationContext ac = this.bootStrap(configProperty); EventDrivenConsumer splitter = (EventDrivenConsumer) ac.getBean("testRouter"); Assert.assertNotNull(splitter); - MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload("1"); - Message inMessage = inChannelMessageBuilder.build(); + MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload("1"); + Message inMessage = inChannelMessageBuilder.build(); DirectChannel inChannel = (DirectChannel) ac.getBean("inChannel"); inChannel.send(inMessage); PollableChannel channel1 = (PollableChannel) ac.getBean("channel1"); @@ -222,28 +235,31 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { PollableChannel channel2 = (PollableChannel) ac.getBean("channel2"); Assert.assertTrue(channel2.receive().getPayload().equals("2")); } + private void testSADefinitionSuccess(String configProperty){ ApplicationContext ac = this.bootStrap(configProperty); EventDrivenConsumer splitter = (EventDrivenConsumer) ac.getBean("testServiceActivator"); Assert.assertNotNull(splitter); - MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload("1"); - Message inMessage = inChannelMessageBuilder.build(); + MessageBuilder inChannelMessageBuilder = MessageBuilder.withPayload("1"); + Message inMessage = inChannelMessageBuilder.build(); DirectChannel inChannel = (DirectChannel) ac.getBean("inChannel"); inChannel.send(inMessage); PollableChannel channel1 = (PollableChannel) ac.getBean("outChannel"); Assert.assertTrue(channel1.receive().getPayload().equals("1")); } + private void testAggregatorDefinitionSuccess(String configProperty){ ApplicationContext ac = this.bootStrap(configProperty); MessageChannel inChannel = (MessageChannel) ac.getBean("inChannel"); for (int i = 0; i < 5; i++) { Map headers = stubHeaders(i, 5, 1); - Message message = new GenericMessage(i, headers); + Message message = new GenericMessage(i, headers); inChannel.send(message); } PollableChannel output = (PollableChannel) ac.getBean("outChannel"); assertEquals(0 + 1 + 2 + 3 + 4, output.receive().getPayload()); } + private void testFilterDefinitionSuccess(String configProperty){ ApplicationContext ac = this.bootStrap(configProperty); MessageChannel input = (MessageChannel) ac.getBean("inChannel"); @@ -262,6 +278,7 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { ac.refresh(); return ac; } + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) { Map headers = new HashMap(); headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber); @@ -271,8 +288,9 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { return headers; } + @SuppressWarnings("unchecked") public static class TestSplitter{ - public Collection split(String[] payload){ + public Collection split(String[] payload) { return CollectionUtils.arrayToList(payload); } } @@ -288,6 +306,7 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { return (value.equals("1")) ? "channel1" : "channel2"; } } + public static class TestServiceActivator{ public String foo(String value) { return value; @@ -303,9 +322,9 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { return result; } } + public static class TestMessageFilter{ public boolean filter(String value) { - System.out.println(">>>> Filtering"); return value.equals("foo"); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java index b46b7a3446..eed92c1b62 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -34,15 +34,15 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.Trigger; -import org.springframework.integration.util.ErrorHandler; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.support.ErrorHandler; /** * @author Iwein Fuld @@ -65,25 +65,26 @@ public class PollingConsumerEndpointTests { private PollableChannel channelMock = createMock(PollableChannel.class); - private SimpleTaskScheduler taskScheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor()); + private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); @Before - public void init() throws InterruptedException { + public void init() throws Exception { consumer.counter.set(0); trigger.reset(); endpoint = new PollingConsumer(channelMock, consumer); - endpoint.setTaskScheduler(taskScheduler); + taskScheduler.setPoolSize(5); taskScheduler.setErrorHandler(errorHandler); - taskScheduler.start(); + endpoint.setTaskScheduler(taskScheduler); endpoint.setTrigger(trigger); endpoint.setReceiveTimeout(-1); + taskScheduler.afterPropertiesSet(); reset(channelMock); } @After - public void stop() { - taskScheduler.stop(); + public void stop() throws Exception { + taskScheduler.destroy(); } @@ -125,6 +126,15 @@ public class PollingConsumerEndpointTests { verify(channelMock); } + @Test + public void heavierLoadTest() throws Exception { + for (int i = 0; i < 1000; i++) { + this.init(); + this.multipleMessages(); + this.stop(); + } + } + @Test(expected = MessageRejectedException.class) public void rejectedMessage() throws Throwable { expect(channelMock.receive()).andReturn(badMessage); @@ -198,7 +208,7 @@ public class PollingConsumerEndpointTests { private volatile CountDownLatch latch = new CountDownLatch(1); - public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) { + public Date nextExecutionTime(TriggerContext triggerContext) { if (!this.hasRun.getAndSet(true)) { return new Date(); } @@ -229,7 +239,7 @@ public class PollingConsumerEndpointTests { private volatile Throwable lastError; - public void handle(Throwable t) { + public void handleError(Throwable t) { this.lastError = t; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java index 907e6258c3..156ba15fd4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -25,6 +25,9 @@ import java.util.GregorianCalendar; import org.junit.Before; import org.junit.Test; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.scheduling.support.SimpleTriggerContext; + /** * @author Dave Syer * @author Mark Fisher @@ -54,7 +57,9 @@ public class CronTriggerTests { @Test public void testMatchAll() throws Exception { CronTrigger trigger = new CronTrigger("* * * * * *"); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -79,7 +84,9 @@ public class CronTriggerTests { calendar.set(Calendar.SECOND, 10); Date date = calendar.getTime(); calendar.add(Calendar.SECOND, 1); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -88,7 +95,9 @@ public class CronTriggerTests { calendar.set(Calendar.SECOND, 11); Date date = calendar.getTime(); calendar.add(Calendar.SECOND, 59); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -107,7 +116,9 @@ public class CronTriggerTests { Date date = calendar.getTime(); calendar.add(Calendar.MINUTE, 1); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -117,9 +128,13 @@ public class CronTriggerTests { Date date = calendar.getTime(); calendar.add(Calendar.MINUTE, 1); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); calendar.add(Calendar.MINUTE, 1); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), date=trigger.nextExecutionTime(triggerContext2)); } @Test @@ -129,7 +144,9 @@ public class CronTriggerTests { calendar.set(Calendar.SECOND, 0); Date date = calendar.getTime(); calendar.add(Calendar.MINUTE, 59); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -143,9 +160,13 @@ public class CronTriggerTests { Date date = calendar.getTime(); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.HOUR_OF_DAY, 12); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); calendar.set(Calendar.HOUR_OF_DAY, 13); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext2)); } @Test @@ -157,10 +178,14 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); assertEquals(2, calendar.get(Calendar.DAY_OF_MONTH)); calendar.add(Calendar.DAY_OF_MONTH, 1); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext2)); assertEquals(3, calendar.get(Calendar.DAY_OF_MONTH)); } @@ -173,7 +198,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -186,7 +213,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -200,9 +229,13 @@ public class CronTriggerTests { calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); calendar.set(Calendar.DAY_OF_MONTH, 1); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); calendar.set(Calendar.DAY_OF_MONTH, 2); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext2)); } @Test @@ -215,10 +248,14 @@ public class CronTriggerTests { calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); calendar.set(Calendar.DAY_OF_MONTH, 31); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); calendar.set(Calendar.MONTH, 10); // November calendar.set(Calendar.DAY_OF_MONTH, 1); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext2)); } @Test @@ -232,9 +269,13 @@ public class CronTriggerTests { calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); calendar.set(Calendar.MONTH, 10); - assertEquals(calendar.getTime(), date=trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext1 = new SimpleTriggerContext(); + triggerContext1.update(null, null, date); + assertEquals(calendar.getTime(), date = trigger.nextExecutionTime(triggerContext1)); calendar.set(Calendar.MONTH, 11); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext2 = new SimpleTriggerContext(); + triggerContext2.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext2)); } @Test @@ -247,7 +288,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -261,7 +304,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } @Test @@ -273,7 +318,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); assertEquals(Calendar.TUESDAY, calendar.get(Calendar.DAY_OF_WEEK)); } @@ -286,7 +333,9 @@ public class CronTriggerTests { calendar.set(Calendar.HOUR_OF_DAY, 0); calendar.set(Calendar.MINUTE, 0); calendar.set(Calendar.SECOND, 0); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); assertEquals(Calendar.TUESDAY, calendar.get(Calendar.DAY_OF_WEEK)); } @@ -368,7 +417,9 @@ public class CronTriggerTests { private void assertMatchesNextSecond(CronTrigger trigger, Calendar calendar) { Date date = calendar.getTime(); roundup(calendar); - assertEquals(calendar.getTime(), trigger.getNextRunTime(null, date)); + SimpleTriggerContext triggerContext = new SimpleTriggerContext(); + triggerContext.update(null, null, date); + assertEquals(calendar.getTime(), trigger.nextExecutionTime(triggerContext)); } } 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 83fa03336b..96433000fe 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -26,14 +26,15 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.scheduling.IntervalTrigger; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.support.ErrorHandler; import org.springframework.util.Assert; /** @@ -70,16 +71,19 @@ public abstract class TestUtils { public static TestApplicationContext createTestApplicationContext() { TestApplicationContext context = new TestApplicationContext(); - registerBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, createTaskScheduler(10), context); + ErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(context)); + ThreadPoolTaskScheduler scheduler = createTaskScheduler(10); + scheduler.setErrorHandler(errorHandler); + registerBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler, context); return context; } - public static TaskScheduler createTaskScheduler(int poolSize) { - ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - executor.setCorePoolSize(poolSize); - executor.setRejectedExecutionHandler(new CallerRunsPolicy()); - executor.afterPropertiesSet(); - return new SimpleTaskScheduler(executor); + public static ThreadPoolTaskScheduler createTaskScheduler(int poolSize) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(poolSize); + scheduler.setRejectedExecutionHandler(new CallerRunsPolicy()); + scheduler.afterPropertiesSet(); + return scheduler; } private static void registerBean(String beanName, Object bean, BeanFactory beanFactory) {