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 - * - * - * @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) {