diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml index 0499aa2c2..c4846c70a 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml @@ -12,7 +12,7 @@ - + diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml index fd37fd1b6..f7f45a50e 100644 --- a/spring-batch-integration/pom.xml +++ b/spring-batch-integration/pom.xml @@ -123,7 +123,6 @@ - 1.0.0.CI-SNAPSHOT - + 1.0.0.RC2 \ No newline at end of file diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index 876971734..dbd3e0cf3 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java @@ -12,7 +12,7 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; -import org.springframework.integration.endpoint.MessagingGateway; +import org.springframework.integration.gateway.MessagingGateway; import org.springframework.util.Assert; public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream { @@ -27,7 +27,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private MessagingGateway messagingGateway; - // TODO: abstract the state or make a factory for this writer? private LocalState localState = new LocalState(); private long throttleLimit = DEFAULT_THROTTLE_LIMIT; @@ -81,7 +80,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo logger.debug("Finished waiting for results in step listener."); } catch (RuntimeException e) { - logger.debug("Detected failure waiting for results in step listener."); + logger.debug("Detected failure waiting for results in step listener.", e); stepExecution.setStatus(BatchStatus.FAILED); return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage()); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java index bf9ed81ef..df2a38b4f 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java @@ -9,7 +9,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -30,9 +29,6 @@ public class SmokeTests { @Qualifier("smokeout") private PollableChannel smokeout; - @Autowired - private MessageBus bus; - // This has to be static because the MessageBus registers the handler // more than once (every time a test instance is created), but only one of // them will get the message. @@ -52,7 +48,6 @@ public class SmokeTests { @Test public void testVanillaSendAndReceive() throws Exception { - bus.start(); smokein.send(new GenericMessage("foo")); @SuppressWarnings("unchecked") Message message = (Message) smokeout.receive(100); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java index 84d38080f..316d36a27 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java @@ -30,7 +30,6 @@ import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -46,9 +45,6 @@ public class ChunkMessageItemWriterIntegrationTests { private ChunkMessageChannelItemWriter writer = new ChunkMessageChannelItemWriter(); - @Autowired - private MessageBus bus; - @Autowired @Qualifier("requests") private MessageChannel requests; @@ -90,15 +86,12 @@ public class ChunkMessageItemWriterIntegrationTests { message = replies.receive(10); } - bus.start(); - } @After public void tearDown() { while (replies.receive(10L) != null) { } - bus.stop(); } @Test diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java index 615256d60..6f3eba334 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java @@ -41,7 +41,7 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.message.MessageConsumer; +import org.springframework.integration.message.MessageHandler; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.util.ReflectionUtils; @@ -66,8 +66,8 @@ public class FileToMessagesJobFactoryBeanTests { itemReader.setLineMapper(new PassThroughLineMapper()); factory.setItemReader(itemReader); factory.setChannel(channel); - channel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + channel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { // TODO: Ask Mark: unsafe cast... receiver.add((String) message.getPayload()); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java index 365268614..988417e55 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java @@ -27,7 +27,6 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.core.io.Resource; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Splitter; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -52,9 +51,6 @@ public class ResourceSplitterIntegrationTests { @Qualifier("requests") private PollableChannel requests; - @Autowired - private MessageBus bus; - /* * This is so cool (but see INT-190)...
* @@ -71,7 +67,6 @@ public class ResourceSplitterIntegrationTests { @SuppressWarnings("unchecked") @Test public void testVanillaConversion() throws Exception { - bus.start(); resources.send(new GenericMessage("classpath:*-context.xml")); Message message = (Message) requests.receive(200L); assertNotNull(message); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java index 7d6c4f99c..3eb4acc40 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java @@ -30,7 +30,7 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.message.MessageConsumer; +import org.springframework.integration.message.MessageHandler; import org.springframework.util.ReflectionUtils; /** @@ -67,8 +67,8 @@ public class MessageChannelItemWriterTests { @Test public void testWriteWithRollback() throws Exception { DirectChannel channel = new DirectChannel(); - channel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + channel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { throw new RuntimeException("Planned failure"); } }); @@ -86,8 +86,8 @@ public class MessageChannelItemWriterTests { @Test public void testWriteWithRollbackOnEndpoint() throws Exception { DirectChannel channel = new DirectChannel(); - channel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + channel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { throw new RuntimeException("Planned failure"); } }); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests.java index a3ab180ed..40e2e245b 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests.java @@ -17,7 +17,6 @@ package org.springframework.batch.integration.job; import static org.junit.Assert.assertEquals; -import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.core.BatchStatus; @@ -27,7 +26,6 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -38,9 +36,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration() @RunWith(SpringJUnit4ClassRunner.class) public class MessageOrientedStepIntegrationTests { - - @Autowired - private MessageBus bus; @Autowired private JobLauncher jobLauncher; @@ -49,14 +44,8 @@ public class MessageOrientedStepIntegrationTests { @Qualifier("job") private Job job; - @After - public void shutdown() { - bus.stop(); - } - @Test public void testLaunchJob() throws Exception { - bus.start(); JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java index 9ed3b1868..16940458b 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java @@ -39,7 +39,7 @@ import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.MessageConsumer; +import org.springframework.integration.message.MessageHandler; import org.springframework.util.ReflectionUtils; /** @@ -97,8 +97,8 @@ public class MessageOrientedStepTests { */ @Test public void testExecuteWithTimeout() throws Exception { - requestChannel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + requestChannel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { } }); step.setExecutionTimeout(1000); @@ -113,8 +113,8 @@ public class MessageOrientedStepTests { @Test public void testVanillaExecute() throws Exception { - requestChannel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + requestChannel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); jobExecution.setStatus(BatchStatus.COMPLETED); replyChannel.send(message); @@ -125,8 +125,8 @@ public class MessageOrientedStepTests { @Test public void testExecuteWithFailure() throws Exception { - requestChannel.subscribe(new MessageConsumer() { - public void onMessage(Message message) { + requestChannel.subscribe(new MessageHandler() { + public void handleMessage(Message message) { JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); jobExecution.registerThrowable(new RuntimeException("Planned failure")); replyChannel.send(message); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java index 560fdfe04..082427eb1 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java @@ -16,7 +16,6 @@ import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.integration.JobSupport; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -31,9 +30,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) public class JobLaunchingMessageHandlerIntegrationTests { - @Autowired - private MessageBus bus; - @Autowired @Qualifier("requests") private MessageChannel requestChannel; @@ -47,7 +43,6 @@ public class JobLaunchingMessageHandlerIntegrationTests { @Before public void setUp() { responseChannel.purge(null); - bus.start(); } @Test diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerTests.java index 60a663e5a..6674e2d34 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerTests.java @@ -14,8 +14,6 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.integration.JobSupport; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; @@ -26,9 +24,6 @@ public class JobLaunchingMessageHandlerTests extends AbstractJUnit4SpringContext StubJobLauncher jobLauncher; - @Autowired - public MessageBus messageBus; - @Before public void setUp() { jobLauncher = new StubJobLauncher(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java index a5d3c6837..2955c0530 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java @@ -11,13 +11,14 @@ import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Poller; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -26,7 +27,7 @@ import org.springframework.util.StringUtils; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @MessageEndpoint -public class RepeatTransactionalPollingIntegrationTests { +public class RepeatTransactionalPollingIntegrationTests implements ApplicationContextAware { private Log logger = LogFactory.getLog(getClass()); @@ -34,10 +35,13 @@ public class RepeatTransactionalPollingIntegrationTests { private List list = new ArrayList(); - @Autowired - private MessageBus bus; + private Lifecycle bus; private volatile int count = 0; + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + bus = (Lifecycle) applicationContext; + } @ServiceActivator(inputChannel = "requests", outputChannel = "replies") public String process(String message) { @@ -90,7 +94,7 @@ public class RepeatTransactionalPollingIntegrationTests { lifecycle.start(); int timeout = 0; while (processed.size() < count && timeout++ < maxTries) { - Thread.sleep(10); + Thread.sleep(5); } lifecycle.stop(); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java index 43a804ebd..d82afdb4d 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java @@ -11,12 +11,14 @@ import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -25,7 +27,7 @@ import org.springframework.util.StringUtils; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @MessageEndpoint -public class RetryRepeatTransactionalPollingIntegrationTests { +public class RetryRepeatTransactionalPollingIntegrationTests implements ApplicationContextAware { private Log logger = LogFactory.getLog(getClass()); @@ -37,8 +39,11 @@ public class RetryRepeatTransactionalPollingIntegrationTests { @Autowired private SimpleService service; - @Autowired - private MessageBus bus; + private Lifecycle bus; + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + bus = (Lifecycle) applicationContext; + } private volatile int count = 0; @@ -85,7 +90,7 @@ public class RetryRepeatTransactionalPollingIntegrationTests { lifecycle.start(); int timeout = 0; while (service.getProcessed().size() < count && timeout++ < maxTries) { - Thread.sleep(10); + Thread.sleep(5); } lifecycle.stop(); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java index 93def00a8..103e18334 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java @@ -11,12 +11,14 @@ import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -25,7 +27,7 @@ import org.springframework.util.StringUtils; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @MessageEndpoint -public class RetryTransactionalPollingIntegrationTests { +public class RetryTransactionalPollingIntegrationTests implements ApplicationContextAware { private Log logger = LogFactory.getLog(getClass()); @@ -37,8 +39,11 @@ public class RetryTransactionalPollingIntegrationTests { @Autowired private SimpleService service; - @Autowired - private MessageBus bus; + private Lifecycle bus; + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + bus = (Lifecycle) applicationContext; + } private volatile int count = 0; diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java index 0e5159fc4..2d0ad7bb2 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java @@ -11,13 +11,14 @@ import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Poller; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.bus.MessageBus; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -26,7 +27,7 @@ import org.springframework.util.StringUtils; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @MessageEndpoint -public class TransactionalPollingIntegrationTests { +public class TransactionalPollingIntegrationTests implements ApplicationContextAware { private Log logger = LogFactory.getLog(getClass()); @@ -34,8 +35,11 @@ public class TransactionalPollingIntegrationTests { private List list = new ArrayList(); - @Autowired - private MessageBus bus; + private Lifecycle bus; + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + bus = (Lifecycle) applicationContext; + } private volatile int count = 0; diff --git a/spring-batch-integration/src/test/resources/job-execution-context.xml b/spring-batch-integration/src/test/resources/job-execution-context.xml index 4fa57e18a..36a483900 100644 --- a/spring-batch-integration/src/test/resources/job-execution-context.xml +++ b/spring-batch-integration/src/test/resources/job-execution-context.xml @@ -8,7 +8,7 @@ http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"> - + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml index dcccb56e1..57f21ae9a 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml @@ -10,9 +10,10 @@ http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + + - + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml index aacba30eb..37381fe45 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml @@ -10,10 +10,10 @@ http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml index 745a80ef6..1c1cbebb1 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml @@ -11,9 +11,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + - + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml index 5c0c5d2cb..633b3347f 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml @@ -11,9 +11,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml index af7df096c..13dd198c2 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml @@ -10,10 +10,10 @@ http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml index 4385e0e8e..70dbb7fe8 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml @@ -10,10 +10,10 @@ http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml index 06c400cf5..068ed7b32 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml @@ -9,7 +9,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemListener.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemListener.java new file mode 100644 index 000000000..eb4d92e82 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemListener.java @@ -0,0 +1,38 @@ +package org.springframework.batch.sample.common; + +import javax.sql.DataSource; + +import org.springframework.batch.core.listener.StepListenerSupport; +import org.springframework.batch.item.ItemReader; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.util.Assert; + +/** + * Thread-safe database {@link ItemReader} implementing the process indicator + * pattern. + */ +public class StagingItemListener extends StepListenerSupport implements InitializingBean { + + private SimpleJdbcTemplate jdbcTemplate; + + public void setDataSource(DataSource dataSource) { + jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + public final void afterPropertiesSet() throws Exception { + Assert.notNull(jdbcTemplate, "You must provide a DataSource."); + } + + @Override + public void afterRead(Long id) { + int count = jdbcTemplate.update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?", + StagingItemWriter.DONE, id, StagingItemWriter.NEW); + if (count != 1) { + throw new OptimisticLockingFailureException("The staging record with ID=" + id + + " was updated concurrently when trying to mark as complete (updated " + count + " records."); + } + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java index eb55f81ef..37f77cd6f 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java @@ -1,96 +1,64 @@ package org.springframework.batch.sample.common; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; -import java.util.Iterator; -import java.util.List; import javax.sql.DataSource; import org.apache.commons.lang.SerializationUtils; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.batch.core.ExitStatus; -import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ReaderNotOpenException; -import org.springframework.beans.factory.DisposableBean; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.database.JdbcCursorItemReader; import org.springframework.beans.factory.InitializingBean; -import org.springframework.dao.DataAccessException; -import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.jdbc.core.PreparedStatementSetter; +import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.simple.ParameterizedRowMapper; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; -import org.springframework.util.Assert; /** * Thread-safe database {@link ItemReader} implementing the process indicator * pattern. */ -public class StagingItemReader implements ItemReader, StepExecutionListener, InitializingBean, DisposableBean { - - private static Log logger = LogFactory.getLog(StagingItemReader.class); - - private StepExecution stepExecution; - - private final Object lock = new Object(); - - private volatile boolean initialized = false; - - private volatile Iterator keys; +public class StagingItemReader implements ItemReader, ItemStream, InitializingBean { + private JdbcCursorItemReader delegate; + private SimpleJdbcTemplate jdbcTemplate; + private long jobId; + + public void setJobId(long jobId) { + this.jobId = jobId; + } + public void setDataSource(DataSource dataSource) { jdbcTemplate = new SimpleJdbcTemplate(dataSource); - } - - public void destroy() throws Exception { - initialized = false; - keys = null; + delegate = new JdbcCursorItemReader(); + delegate.setDataSource(dataSource); + delegate.setSql("SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID"); + delegate.setPreparedStatementSetter(new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + ps.setLong(1, jobId); + ps.setString(2, StagingItemWriter.NEW); + } + }); + delegate.setMapper(new RowMapper() { + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + return rs.getLong(1); + } + }); } public final void afterPropertiesSet() throws Exception { - Assert.notNull(jdbcTemplate, "You must provide a DataSource."); + delegate.afterPropertiesSet(); } - private List retrieveKeys() { + public T read() throws Exception { - synchronized (lock) { - - return jdbcTemplate.query( - - "SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID", - - new ParameterizedRowMapper() { - public Long mapRow(ResultSet rs, int rowNum) throws SQLException { - return rs.getLong(1); - } - }, - - stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW); - - } - - } - - public T read() throws DataAccessException { - - if (!initialized) { - throw new ReaderNotOpenException("ItemStream must be open before it can be read."); - } - - Long id = null; - synchronized (lock) { - if (keys.hasNext()) { - id = keys.next(); - } - } - logger.debug("Retrieved key from list: " + id); - - if (id == null) { - return null; - } + Long id = delegate.read(); @SuppressWarnings("unchecked") T result = (T) jdbcTemplate.queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?", new ParameterizedRowMapper() { @@ -100,55 +68,20 @@ public class StagingItemReader implements ItemReader, StepExecutionListene } }, id); - // Update now - changes will rollback if there is a problem later. - int count = jdbcTemplate.update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?", - StagingItemWriter.DONE, id, StagingItemWriter.NEW); - if (count != 1) { - throw new OptimisticLockingFailureException("The staging record with ID=" + id - + " was updated concurrently when trying to mark as complete (updated " + count + " records."); - } - return result; } - /* - * (non-Javadoc) - * - * @see - * org.springframework.batch.core.domain.StepListener#afterStep(StepExecution - * ) - */ - public ExitStatus afterStep(StepExecution stepExecution) { - return null; + public void close(ExecutionContext executionContext) throws ItemStreamException { + delegate.close(executionContext); } - /* - * (non-Javadoc) - * - * @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org. - * springframework.batch.core.domain.StepExecution) - */ - public void beforeStep(StepExecution stepExecution) { - this.stepExecution = stepExecution; - synchronized (lock) { - if (keys == null) { - keys = retrieveKeys().iterator(); - logger.info("Keys obtained for staging."); - initialized = true; - } - } + public void open(ExecutionContext executionContext) throws ItemStreamException { + delegate.open(executionContext); } - /* - * (non-Javadoc) - * - * @see - * org.springframework.batch.core.domain.StepListener#onErrorInStep(java - * .lang.Throwable) - */ - public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { - return null; + public void update(ExecutionContext executionContext) throws ItemStreamException { + delegate.update(executionContext); } } diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index c12461453..6effa93ac 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -46,6 +46,11 @@ + + + + + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java index b1dabd5ff..94e55f835 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java @@ -13,6 +13,7 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; import org.springframework.test.context.ContextConfiguration; @@ -54,12 +55,11 @@ public class StagingItemReaderTests { new JobParameters(), "testJob"))); writer.beforeStep(stepExecution); writer.write(Arrays.asList(new String[] { "FOO", "BAR", "SPAM", "BUCKET" })); - reader.beforeStep(stepExecution); + reader.setJobId(jobId); } @AfterTransaction public void onTearDownAfterTransaction() throws Exception { - reader.destroy(); simpleJdbcTemplate.update("DELETE FROM BATCH_STAGING"); } @@ -118,7 +118,13 @@ public class StagingItemReaderTests { String.class, id); assertEquals(StagingItemWriter.NEW, before); - Object item = reader.read(); + Object item; + try { + item = reader.read(); + } + catch (Exception e) { + throw new UnexpectedJobExecutionException("Reader error", e); + } assertEquals("FOO", item); transactionStatus.setRollbackOnly();