diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java index d596700e3..91281a58b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java @@ -146,7 +146,7 @@ public class JobExecution extends Entity { /** * Register a step execution with the current job execution. - * @param stepName TODO + * @param stepName the name of the step the new execution is associated with */ public StepExecution createStepExecution(String stepName) { StepExecution stepExecution = new StepExecution(stepName, this); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index edc7cd70e..530afa2ca 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -353,10 +353,9 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB); } else { - String message = ""; StringWriter writer = new StringWriter(); ex.printStackTrace(new PrintWriter(writer)); - message = writer.toString(); + String message = writer.toString(); exitStatus = ExitStatus.FAILED.addExitDescription(message); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java deleted file mode 100644 index 9f34593a1..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * Copyright 2002-2007 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.batch.repeat.support; - -import org.springframework.batch.repeat.RepeatException; - -/** - * Abstract implementation that can be extended for both Backport Concurrent and - * JDK 5 Concurrent. - * - * @author Ben Hale - * @author Dave Syer - */ -abstract class AbstractResultQueue extends RepeatInternalStateSupport implements ResultQueue { - - // Arbitrary lock object. - private final Object lock = new Object(); - - // Counter to monitor the difference between expected and actually collected - // results. When this reaches zero there are really no more results. - private volatile int count = 0; - - public boolean isExpecting() { - synchronized (lock) { - // Base the decision about whether we expect more results on a - // counter of the number of expected results actually collected. - return count > 0; - } - } - - public void expect() { - try { - synchronized (lock) { - aquireWait(); - count++; - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RepeatException("InterruptedException waiting for to acquire lock on input."); - } - } - - public void put(ResultHolder holder) { - // There should be no need to block here, or to use offer() - addResult(holder); - // Take from the waits queue now to allow another result to - // accumulate. But don't decrement the counter. - releaseWait(); - } - - public ResultHolder take() { - ResultHolder value; - try { - synchronized (lock) { - value = takeResult(); - // Decrement the counter only when the result is collected. - count--; - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RepeatException("InterruptedException while waiting for result."); - } - return value; - } - - /** - * Acquire permission for one more task on the queue. - * - * @throws InterruptedException - */ - protected abstract void aquireWait() throws InterruptedException; - - /** - * Release the permit that we were holding while a task was processed. - */ - protected abstract void releaseWait(); - - /** - * Add a {@link ResultHolder} with a finished result to the queue for - * collection. Should not block. May throw an exception to signal that the - * queue is full, but that would be an unexpected condition. - * - * @param resultHolder a {@link ResultHolder} - */ - protected abstract void addResult(ResultHolder resultHolder); - - /** - * Obtain a result from the queue, blocking until one becomes available. - * - * @return a {@link ResultHolder} with the completed result - * @throws InterruptedException if an interrupt is signalled - */ - protected abstract ResultHolder takeResult() throws InterruptedException; - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/JdkConcurrentResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/JdkConcurrentResultQueue.java index 5730ca89c..47cd52d97 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/JdkConcurrentResultQueue.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/JdkConcurrentResultQueue.java @@ -20,12 +20,14 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.Semaphore; +import org.springframework.batch.repeat.RepeatException; + /** * An implementation of the {@link ResultQueue} that uses the Java 5 Concurrent Utilities. * * @author Ben Hale */ -class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInternalState { +class JdkConcurrentResultQueue extends RepeatInternalStateSupport implements ResultQueue { // Accumulation of result objects as they finish. private final BlockingQueue results; @@ -33,6 +35,10 @@ class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInte // Accumulation of dummy objects flagging expected results in the future. private final Semaphore waits; + private final Object lock = new Object(); + + private volatile int count = 0; + JdkConcurrentResultQueue(int throttleLimit) { results = new LinkedBlockingQueue(); waits = new Semaphore(throttleLimit); @@ -58,4 +64,49 @@ class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInte return results.isEmpty(); } + public boolean isExpecting() { + synchronized (lock) { + // Base the decision about whether we expect more results on a + // counter of the number of expected results actually collected. + return count > 0; + } + } + + public void expect() { + try { + synchronized (lock) { + aquireWait(); + count++; + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RepeatException("InterruptedException waiting for to acquire lock on input."); + } + } + + public void put(ResultHolder holder) { + // There should be no need to block here, or to use offer() + addResult(holder); + // Take from the waits queue now to allow another result to + // accumulate. But don't decrement the counter. + releaseWait(); + } + + public ResultHolder take() { + ResultHolder value; + try { + synchronized (lock) { + value = takeResult(); + // Decrement the counter only when the result is collected. + count--; + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RepeatException("InterruptedException while waiting for result."); + } + return value; + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java index 357e9f72d..c69cd203c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java @@ -24,6 +24,8 @@ package org.springframework.batch.repeat.support; * * @author Dave Syer * @author Ben Hale + * + * TODO: BlockingQueue with a CompletionService? */ interface ResultQueue extends RepeatInternalState { 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 07ecb8854..878ca1db6 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,10 +12,10 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.util.Assert; public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream { @@ -28,9 +28,9 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private static final long DEFAULT_THROTTLE_LIMIT = 6; - private MessageTarget target; + private MessageChannel target; - private BlockingSource source; + private PollableChannel source; // TODO: abstract the state or make a factory for this writer? private LocalState localState = new LocalState(); @@ -46,11 +46,13 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo this.throttleLimit = throttleLimit; } - public void setSource(BlockingSource source) { + // TODO: refactor to ChannelAdapter? + public void setInputChannel(PollableChannel source) { this.source = source; } - public void setTarget(MessageTarget target) { + // TODO: re-evaluate the refactor to MessageChannel + public void setOutputChannel(MessageChannel target) { this.target = target; } @@ -145,7 +147,8 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * otherwise return null. */ private void getNextResult(long timeout) { - Message message = source.receive(timeout); + @SuppressWarnings("unchecked") + Message message = (Message) source.receive(timeout); if (message != null) { ChunkResponse payload = message.getPayload(); Long jobInstanceId = payload.getJobId(); @@ -174,7 +177,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } public int getSkipCount() { - // TODO Auto-generated method stub return stepExecution.getSkipCount(); } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java index 2d7e69a8a..abce8c644 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java @@ -4,9 +4,11 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.repeat.ExitStatus; import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.util.Assert; +@MessageEndpoint public class ChunkProcessorChunkHandler implements ChunkHandler, InitializingBean { private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class); @@ -32,7 +34,7 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, Initializ * (non-Javadoc) * @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection) */ - @Handler + @ServiceActivator public ChunkResponse handleChunk(ChunkRequest chunkRequest) { logger.debug("Handling chunk: " + chunkRequest); diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java index 6bbf61853..4a3d096f9 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java @@ -23,10 +23,10 @@ import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.beans.factory.annotation.Required; -import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.util.Assert; /** @@ -40,9 +40,9 @@ public class MessageOrientedStep extends AbstractStep { */ public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING"; - private MessageTarget target; + private MessageChannel outputChannel; - private BlockingSource source; + private PollableChannel source; private static int MINUTE = 1000 * 60; @@ -77,11 +77,11 @@ public class MessageOrientedStep extends AbstractStep { /** * Public setter for the target. - * @param target the target to set + * @param outputChannel the target to set */ @Required - public void setTarget(MessageTarget target) { - this.target = target; + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; } /** @@ -89,7 +89,7 @@ public class MessageOrientedStep extends AbstractStep { * @param source the source to set */ @Required - public void setSource(BlockingSource source) { + public void setInputChannel(PollableChannel source) { this.source = source; } @@ -113,7 +113,7 @@ public class MessageOrientedStep extends AbstractStep { executionContext.putString(WAITING, "true"); // TODO: need these two lines to be atomic getJobRepository().update(stepExecution); - target.send(new GenericMessage(request)); + outputChannel.send(new GenericMessage(request)); waitForReply(request.getJobId()); } @@ -132,7 +132,8 @@ public class MessageOrientedStep extends AbstractStep { while (count++ < maxCount) { // TODO: timeout? - Message message = source.receive(timeout); + @SuppressWarnings("unchecked") + Message message = (Message) source.receive(timeout); if (message != null) { diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java index 7a0f22d4d..85eb99c45 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java @@ -25,7 +25,7 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.annotation.Required; -import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.ServiceActivator; /** * @author Dave Syer @@ -58,7 +58,7 @@ public class StepExecutionMessageHandler { this.jobRepository = jobRepository; } - @Handler + @ServiceActivator public JobExecutionRequest handle(JobExecutionRequest request) { // Hand off immediately if the job has already failed diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java index 4ae22495e..c9b16dda2 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java @@ -6,7 +6,7 @@ import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.batch.core.launch.JobLauncher; -import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.ServiceActivator; /** * Message handler which uses strategies to convert a Message into a job and a @@ -27,7 +27,7 @@ public class JobLaunchingMessageHandler { this.jobLauncher = jobLauncher; } - @Handler + @ServiceActivator public JobExecution launch(JobLaunchRequest request) { Job job = request.getJob(); JobParameters jobParameters = request.getJobParameters(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobRepositorySupport.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobRepositorySupport.java index 6bfb82bf3..2b6135e61 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobRepositorySupport.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobRepositorySupport.java @@ -35,7 +35,7 @@ public class JobRepositorySupport implements JobRepository { */ public JobExecution createJobExecution(String jobName, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException { - return new JobExecution(new JobInstance(0L, jobParameters, job.getName())); + return new JobExecution(new JobInstance(0L, jobParameters, jobName)); } /* (non-Javadoc) @@ -76,4 +76,8 @@ public class JobRepositorySupport implements JobRepository { public void update(StepExecution stepExecution) { } + public boolean isJobInstanceExists(String jobName, JobParameters jobParameters) { + return false; + } + } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobSupport.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobSupport.java index 7cfdab45d..041dcd020 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobSupport.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/JobSupport.java @@ -14,7 +14,6 @@ public class JobSupport implements Job { } public void execute(JobExecution execution) throws JobExecutionException { - // TODO Auto-generated method stub } public String getName() { @@ -22,12 +21,10 @@ public class JobSupport implements Job { } public boolean isRestartable() { - // TODO Auto-generated method stub return false; } public JobParametersIncrementer getJobParametersIncrementer() { - // TODO Auto-generated method stub return null; } 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 55adf9d81..173bc5c42 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 @@ -7,10 +7,10 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.test.context.ContextConfiguration; @@ -18,7 +18,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration(locations = "/integration-context.xml") @RunWith(SpringJUnit4ClassRunner.class) -@MessageEndpoint(input = "smokein", output = "smokeout") +@MessageEndpoint public class SmokeTests { @Autowired @@ -27,14 +27,14 @@ public class SmokeTests { @Autowired @Qualifier("smokeout") - private BlockingSource smokeout; + private PollableChannel smokeout; // 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. private volatile static int count = 0; - @Handler + @ServiceActivator(inputChannel = "smokein", outputChannel = "smokeout") public String process(String message) { count++; String result = message + ": " + count; @@ -49,7 +49,8 @@ public class SmokeTests { @Test public void testVanillaSendAndReceive() throws Exception { smokein.send(new GenericMessage("foo")); - Message message = smokeout.receive(100); + @SuppressWarnings("unchecked") + Message message = (Message) smokeout.receive(100); String result = (String) (message == null ? null : message.getPayload()); assertEquals("foo: 1", result); assertEquals(1, count); 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 6227e7ba9..8367a903b 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 @@ -32,6 +32,7 @@ import org.springframework.batch.repeat.ExitStatus; 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.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -47,7 +48,7 @@ public class ChunkMessageItemWriterIntegrationTests { @Autowired @Qualifier("requests") - private PollableChannel requests; + private MessageChannel requests; @Autowired @Qualifier("replies") @@ -59,7 +60,6 @@ public class ChunkMessageItemWriterIntegrationTests { private static long jobCounter; - @SuppressWarnings("unchecked") @Before public void setUp() { @@ -71,18 +71,13 @@ public class ChunkMessageItemWriterIntegrationTests { factory.setItemWriter(writer); factory.setCommitInterval(4); - writer.setSource(replies); - writer.setTarget(requests); + writer.setInputChannel(replies); + writer.setOutputChannel(requests); TestItemWriter.count = 0; // Drain queues - Message message = requests.receive(10); - while (message!=null) { - System.err.println(message); - message = requests.receive(10); - } - message = replies.receive(10); + Message message = replies.receive(10); while (message!=null) { System.err.println(message); message = replies.receive(10); @@ -92,7 +87,6 @@ public class ChunkMessageItemWriterIntegrationTests { @After public void tearDown() { - while(requests.receive(10L)!=null) {} while(replies.receive(10L)!=null) {} } @@ -126,7 +120,7 @@ public class ChunkMessageItemWriterIntegrationTests { waitForResults(6, 10); assertEquals(6, TestItemWriter.count); - assertEquals(6, stepExecution.getItemCount()); + assertEquals(6, stepExecution.getReadCount()); } @@ -153,7 +147,7 @@ public class ChunkMessageItemWriterIntegrationTests { waitForResults(8, 10); assertEquals(8, TestItemWriter.count); - assertEquals(6, stepExecution.getItemCount()); + assertEquals(6, stepExecution.getReadCount()); } @@ -185,7 +179,7 @@ public class ChunkMessageItemWriterIntegrationTests { waitForResults(1, 10); assertEquals(1, TestItemWriter.count); - assertEquals(0, stepExecution.getItemCount()); + assertEquals(0, stepExecution.getReadCount()); } @@ -260,7 +254,7 @@ public class ChunkMessageItemWriterIntegrationTests { } assertEquals(0, TestItemWriter.count); - assertEquals(0, stepExecution.getItemCount()); + assertEquals(0, stepExecution.getReadCount()); } 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 8d7d60efb..d3244c8e0 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 @@ -20,8 +20,9 @@ import static org.junit.Assert.assertNotNull; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; -import org.junit.After; import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.BatchStatus; @@ -40,8 +41,8 @@ import org.springframework.beans.factory.annotation.Required; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageConsumer; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.util.ReflectionUtils; @@ -53,7 +54,8 @@ public class FileToMessagesJobFactoryBeanTests { private static final String FILE_INPUT_PATH = ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH; private FileToMessagesJobFactoryBean
factory = new FileToMessagesJobFactoryBean
(); - private ThreadLocalChannel receiver = new ThreadLocalChannel(); + private DirectChannel channel = new DirectChannel(); + private List
receiver = new ArrayList
(); private JobRepositorySupport jobRepository; @Before @@ -64,16 +66,15 @@ public class FileToMessagesJobFactoryBeanTests { FlatFileItemReader
itemReader = new FlatFileItemReader
(); itemReader.setFieldSetMapper(new PassThroughFieldSetMapper()); factory.setItemReader(itemReader); - DirectChannel channel = new DirectChannel(); factory.setChannel(channel); - channel.subscribe(this.receiver); + channel.subscribe(new MessageConsumer() { + public void onMessage(Message message) { + // TODO: Ask Mark: unsafe cast... + receiver.add((FieldSet) message.getPayload()); + } + }); } - @After - public void tearDown() { - while(receiver.receive(10L)!=null) {} - } - /** * Test method for * {@link org.springframework.batch.integration.file.FileToMessagesJobFactoryBean#setBeanName(java.lang.String)}. @@ -170,7 +171,6 @@ public class FileToMessagesJobFactoryBeanTests { assertEquals(true, factory.isSingleton()); } - @SuppressWarnings("unchecked") @Test public void testVanillaJobExecution() throws Exception { @@ -183,17 +183,12 @@ public class FileToMessagesJobFactoryBeanTests { assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); FieldSet payload; - Message
message; // first line from properties file - message = (Message
) receiver.receive(100L); - assertNotNull(message); - payload = message.getPayload(); + payload = receiver.get(0); assertNotNull(payload); // second line from properties file - message = (Message
) receiver.receive(100L); - assertNotNull(message); - payload = message.getPayload(); + payload = receiver.get(1); assertNotNull(payload); } 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 41ee5be78..cf26f5ca7 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 @@ -40,7 +40,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration() @RunWith(SpringJUnit4ClassRunner.class) -@MessageEndpoint(input = "resources", output = "requests") +@MessageEndpoint public class ResourceSplitterIntegrationTests { @Autowired @@ -57,7 +57,7 @@ public class ResourceSplitterIntegrationTests { * The incoming message is a Resource pattern, and it is converted to the * correct payload type with Spring's default strategy */ - @Splitter + @Splitter(inputChannel = "resources", outputChannel = "requests") public Resource[] handle(Resource[] message) { List list = Arrays.asList(message); System.err.println(list); 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 807b1dab9..cb914680e 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 @@ -29,24 +29,27 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.ThreadLocalChannel; -import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessageConsumer; import org.springframework.util.ReflectionUtils; /** * @author Dave Syer - * + * */ public class MessageChannelItemWriterTests { /** - * Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#setChannel(org.springframework.integration.channel.MessageChannel)}. + * Test method for + * {@link org.springframework.batch.integration.item.MessageChannelItemWriter#setChannel(org.springframework.integration.channel.MessageChannel)} + * . */ @Test public void testSetChannel() { - Method method = ReflectionUtils.findMethod(MessageChannelItemWriter.class, "setChannel", new Class[] {MessageChannel.class}); + Method method = ReflectionUtils.findMethod(MessageChannelItemWriter.class, "setChannel", + new Class[] { MessageChannel.class }); assertNotNull(method); Annotation[] annotations = AnnotationUtils.getAnnotations(method); assertEquals(1, annotations.length); @@ -55,11 +58,9 @@ public class MessageChannelItemWriterTests { @Test public void testWrite() throws Exception { - DirectChannel channel = new DirectChannel(); ThreadLocalChannel receiver = new ThreadLocalChannel(); - channel.subscribe(receiver); MessageChannelItemWriter writer = new MessageChannelItemWriter(); - writer.setChannel(channel); + writer.setChannel(receiver); writer.write(Collections.singletonList("foo")); Message message = receiver.receive(10); assertNotNull(message); @@ -69,8 +70,8 @@ public class MessageChannelItemWriterTests { @Test public void testWriteWithRollback() throws Exception { DirectChannel channel = new DirectChannel(); - channel.subscribe(new MessageTarget() { - public boolean send(Message message) { + channel.subscribe(new MessageConsumer() { + public void onMessage(Message message) { throw new RuntimeException("Planned failure"); } }); @@ -88,7 +89,7 @@ public class MessageChannelItemWriterTests { @Test public void testWriteWithRollbackOnEndpoint() throws Exception { DirectChannel channel = new DirectChannel(); - DefaultEndpoint endpoint = new DefaultEndpoint(new MessageHandler() { + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new MessageHandler() { public Message handle(Message message) { throw new RuntimeException("Planned failure"); } @@ -101,7 +102,8 @@ public class MessageChannelItemWriterTests { fail("Expected RuntimeException"); } catch (RuntimeException e) { - // INT-377: this assertion fails because the exception is wrapped too tightly + // INT-377: this assertion fails because the exception is wrapped + // too tightly assertEquals("Planned failure", e.getCause().getMessage()); } } 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 a6fb37a9b..60a5899c7 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 @@ -34,12 +34,12 @@ import org.springframework.batch.integration.JobRepositorySupport; import org.springframework.beans.factory.annotation.Required; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.ThreadLocalChannel; -import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessageConsumer; import org.springframework.util.ReflectionUtils; /** @@ -56,28 +56,23 @@ public class MessageOrientedStepTests { private PollableChannel replyChannel; - @SuppressWarnings("unchecked") @Before public void createStep() { replyChannel = new ThreadLocalChannel(); requestChannel = new DirectChannel(); step.setName("step"); - step.setTarget(requestChannel); - step.setSource(replyChannel); + step.setOutputChannel(requestChannel); + step.setInputChannel(replyChannel); step.setStartLimit(10); step.setJobRepository(new JobRepositorySupport()); JobInstance jobInstance = new JobInstance(0L, new JobParameters(), "job"); jobExecution = new JobExecution(jobInstance); } - /** - * Test method for - * {@link org.springframework.batch.integration.job.MessageOrientedStep#setTarget(MessageTarget)}. - */ @Test public void testSetRequestChannel() { - Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setTarget", - new Class[] { MessageTarget.class }); + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setOutputChannel", + new Class[] { MessageChannel.class }); assertNotNull(method); Annotation[] annotations = AnnotationUtils.getAnnotations(method); assertEquals(1, annotations.length); @@ -86,8 +81,8 @@ public class MessageOrientedStepTests { @Test public void testSetReplyChannel() { - Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource", - new Class[] { BlockingSource.class }); + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setInputChannel", + new Class[] { PollableChannel.class }); assertNotNull(method); Annotation[] annotations = AnnotationUtils.getAnnotations(method); assertEquals(1, annotations.length); @@ -96,11 +91,16 @@ public class MessageOrientedStepTests { /** * Test method for - * {@link org.springframework.batch.integration.job.MessageOrientedStep#execute(org.springframework.batch.core.StepExecution)}. + * {@link org.springframework.batch.integration.job.MessageOrientedStep#execute(org.springframework.batch.core.StepExecution)} + * . * @throws Exception */ @Test public void testExecuteWithTimeout() throws Exception { + requestChannel.subscribe(new MessageConsumer() { + public void onMessage(Message message) { + } + }); try { step.setExecutionTimeout(1000); step.setPollingInterval(100); @@ -116,11 +116,11 @@ public class MessageOrientedStepTests { @Test public void testVanillaExecute() throws Exception { - requestChannel.subscribe(new MessageTarget() { - public boolean send(Message message) { + requestChannel.subscribe(new MessageConsumer() { + public void onMessage(Message message) { JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); jobExecution.setStatus(BatchStatus.COMPLETED); - return replyChannel.send(message); + replyChannel.send(message); } }); step.execute(jobExecution.createStepExecution(step.getName())); @@ -128,11 +128,11 @@ public class MessageOrientedStepTests { @Test public void testExecuteWithFailure() throws Exception { - requestChannel.subscribe(new MessageTarget() { - public boolean send(Message message) { + requestChannel.subscribe(new MessageConsumer() { + public void onMessage(Message message) { JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); jobExecution.registerThrowable(new RuntimeException("Planned failure")); - return replyChannel.send(message); + replyChannel.send(message); } }); try { diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java index 0e1545f68..c097fd5b1 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java @@ -79,8 +79,8 @@ public class StepExecutionMessageHandlerTests { public void testVanillaHandle() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport(); StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobRepository.createJobExecution( - job.getName(), new JobParameters()))); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobRepository.createJobExecution("job", + new JobParameters()))); assertEquals(1, message.getJobExecution().getStepExecutions().size()); assertEquals(BatchStatus.COMPLETED, message.getStatus()); } @@ -89,8 +89,8 @@ public class StepExecutionMessageHandlerTests { public void testHandleWithInputs() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport(); StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution( - job.getName(), new JobParameters())); + JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution("job", + new JobParameters())); jobExecutionRequest.getJobExecution().getExecutionContext().putString("foo", "bar"); JobExecutionRequest message = handler.handle(jobExecutionRequest); assertEquals(1, message.getJobExecution().getStepExecutions().size()); @@ -102,8 +102,8 @@ public class StepExecutionMessageHandlerTests { public void testHandleWithInputsAndOutputs() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport(); StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution( - job.getName(), new JobParameters())); + JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution("job", + new JobParameters())); jobExecutionRequest.getJobExecution().getExecutionContext().putString("foo", "bar"); // The step has to add the output attribute to the context handler.setStep(new StepSupport("step") { @@ -122,7 +122,7 @@ public class StepExecutionMessageHandlerTests { public void testHandleFailedJob() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport(); StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); + JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); jobExecution.setStatus(BatchStatus.FAILED); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertEquals(0, message.getJobExecution().getStepExecutions().size()); @@ -132,8 +132,8 @@ public class StepExecutionMessageHandlerTests { public void testHandleRestart() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport() { @Override - public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { - StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance)); + public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) { + StepExecution stepExecution = new StepExecution(stepName, new JobExecution(jobInstance)); stepExecution.setStatus(BatchStatus.FAILED); stepExecution.setExecutionContext(new ExecutionContext() { { @@ -144,20 +144,13 @@ public class StepExecutionMessageHandlerTests { return stepExecution; } - /* - * (non-Javadoc) - * - * @seeorg.springframework.integration.batch.JobRepositorySupport# - * getStepExecutionCount(org.springframework.batch.core.JobInstance, - * org.springframework.batch.core.Step) - */ @Override - public int getStepExecutionCount(JobInstance jobInstance, Step step) { + public int getStepExecutionCount(JobInstance jobInstance, String stepName) { return 1; } }; StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); + JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertNotNull(message); assertEquals(1, jobExecution.getStepExecutions().size()); @@ -169,8 +162,8 @@ public class StepExecutionMessageHandlerTests { public void testHandleRestartAlreadyComplete() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport() { @Override - public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { - StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance)); + public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) { + StepExecution stepExecution = new StepExecution(stepName, new JobExecution(jobInstance)); stepExecution.setStatus(BatchStatus.COMPLETED); stepExecution.setExecutionContext(new ExecutionContext() { { @@ -181,7 +174,7 @@ public class StepExecutionMessageHandlerTests { } }; StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); + JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertNotNull(message); assertEquals(1, jobExecution.getStepExecutions().size()); @@ -196,18 +189,18 @@ public class StepExecutionMessageHandlerTests { public void testHandleRestartStartLimitExceeded() throws Exception { JobRepositorySupport jobRepository = new JobRepositorySupport() { @Override - public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { - return new StepExecution(step.getName(), new JobExecution(jobInstance)); + public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) { + return new StepExecution(stepName, new JobExecution(jobInstance)); } @Override - public int getStepExecutionCount(JobInstance jobInstance, Step step) { + public int getStepExecutionCount(JobInstance jobInstance, String stepName) { // sufficiently high restart count return 100; } }; StepExecutionMessageHandler handler = createHandler(jobRepository); - JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); + JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertNotNull(message); assertEquals(1, jobExecution.getStepExecutions().size()); 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 6d6054157..eb3f26acb 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,12 +16,12 @@ 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.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHeaders; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessagingException; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -32,7 +32,7 @@ public class JobLaunchingMessageHandlerIntegrationTests { @Autowired @Qualifier("requests") - private MessageTarget requestChannel; + private MessageChannel requestChannel; @Autowired @Qualifier("response") @@ -54,9 +54,9 @@ public class JobLaunchingMessageHandlerIntegrationTests { try { requestChannel.send(trigger); } - catch (MessageDeliveryException e) { + catch (MessagingException e) { String message = e.getMessage(); - assertTrue("Wrong message: " + message, message.contains("reply target")); + assertTrue("Wrong message: " + message, message.contains("reply channel")); } Message executionMessage = (Message) responseChannel.receive(1000); @@ -79,7 +79,7 @@ public class JobLaunchingMessageHandlerIntegrationTests { assertNotNull("No response received", executionMessage); JobExecution execution = executionMessage.getPayload(); - assertNotNull("JobExectuion not returned", execution); + assertNotNull("JobExecution not returned", execution); } } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java index 0b297bcf0..01c33fde7 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java @@ -19,7 +19,7 @@ import java.util.Properties; import org.springframework.batch.core.converter.DefaultJobParametersConverter; import org.springframework.batch.integration.JobSupport; -import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.ServiceActivator; /** * @author Dave Syer @@ -27,7 +27,7 @@ import org.springframework.integration.annotation.Handler; */ public class JobRequestConverter { - @Handler + @ServiceActivator public JobLaunchRequest convert(String jobName) { // TODO: get these from message header Properties properties = new Properties(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java index ffe06221b..081b721c9 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java @@ -17,6 +17,9 @@ import org.springframework.aop.support.DefaultPointcutAdvisor; import org.springframework.aop.support.NameMatchMethodPointcut; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.retry.interceptor.MethodArgumentsKeyGenerator; import org.springframework.batch.retry.interceptor.MethodInvocationRecoverer; import org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor; @@ -24,11 +27,11 @@ import org.springframework.batch.support.transaction.ResourcelessTransactionMana import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.endpoint.SourcePoller; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageExchangeTemplate; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.SchedulableTask; @@ -36,6 +39,8 @@ import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.integration.scheduling.spi.ProviderTaskScheduler; import org.springframework.integration.scheduling.spi.SimpleScheduleServiceProvider; import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.StringUtils; public class PollableSourceRetryTests { @@ -70,16 +75,16 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); - return processed.add((String) payload); + processed.add((String) payload); } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1); + MessageChannel target = getChannel(handler); + SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 2, 40); @@ -98,8 +103,8 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); processed.add((String) payload); @@ -107,8 +112,8 @@ public class PollableSourceRetryTests { } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); - PollingDispatcher trigger = getPollingDispatcher(source, target, null, 1); + MessageChannel target = getChannel(handler); + SourcePoller trigger = getSourcePoller(source, target, null, 1); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 2, 20); @@ -129,8 +134,8 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); processed.add((String) payload); @@ -138,8 +143,8 @@ public class PollableSourceRetryTests { } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1); + MessageChannel target = getChannel(handler); + SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 2, 40); @@ -162,21 +167,20 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); - boolean result = processed.add((String) payload); + processed.add((String) payload); if ("fail".equals(payload)) { throw new RuntimeException("Planned failure: " + payload); } - return result; } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1); + MessageChannel target = getChannel(handler); + SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 5, 50); @@ -202,22 +206,24 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); - boolean result = processed.add((String) payload); + processed.add((String) payload); if ("fail".equals(payload)) { throw new RuntimeException("Planned failure: " + payload); } - return result; } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 3); - TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); + MessageChannel target = getChannel(handler); + SourcePoller trigger = getSourcePoller(source, target, null, 1); + SchedulableTask task = (SchedulableTask) getProxy(trigger, SchedulableTask.class, new Advice[] { + new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()), + getRepeatOperationsInterceptor(3) }, "run"); + TaskScheduler scheduler = getSchedulerWithErrorHandler(task); waitForResults(scheduler, 6, 100); @@ -242,29 +248,29 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { if (message == null) { - return false; + return; } Object payload = message.getPayload(); logger.debug("Handling: " + payload); - boolean result = processed.add((String) payload); + processed.add((String) payload); // INT-184 this won't work if it is a "real" handler that throws // MessageHandlingException if ("fail".equals(payload)) { throw new RuntimeException("Planned failure: " + payload); } - return result; } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); + MessageChannel target = getChannel(handler); // this was the old dispatch advice chain - target = (MessageTarget) getProxy(target, MessageTarget.class, + target = (MessageChannel) getProxy(target, MessageChannel.class, new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send"); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1); + SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); + TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 4, 40); @@ -290,34 +296,37 @@ public class PollableSourceRetryTests { list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,fail,c,d,e,f,g,h,j,k"))); int beforeCount = list.size(); - MessageTarget handler = new MessageTarget() { - public boolean send(Message message) { + MessageConsumer handler = new MessageConsumer() { + public void onMessage(Message message) { Object payload = message.getPayload(); logger.debug("Handling: " + payload); - boolean result = processed.add((String) payload); + processed.add((String) payload); if ("fail".equals(payload)) { throw new RuntimeException("Planned failure: " + payload); } - return result; } }; PollableSource source = getPollableSource(list); - MessageTarget target = getChannel(handler); + MessageChannel target = getChannel(handler); + // this was the old dispatch advice chain - target = (MessageTarget) getProxy(target, MessageTarget.class, + target = (MessageChannel) getProxy(target, MessageChannel.class, new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send"); - PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 3); - TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); + SourcePoller trigger = getSourcePoller(source, target, null, 1); + SchedulableTask task = (SchedulableTask) getProxy(trigger, SchedulableTask.class, new Advice[] { + new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()), + getRepeatOperationsInterceptor(3) }, "run"); + TaskScheduler scheduler = getSchedulerWithErrorHandler(task); waitForResults(scheduler, 6, 100); System.err.println(processed); System.err.println(list); - assertEquals(6, processed.size()); assertFalse("No messages got to processor", processed.isEmpty()); - // One roll back and then start again with a,b,d,e - assertEquals(beforeCount - 5, list.size()); + assertEquals(7, processed.size()); + // 6 items were removed from the list + assertEquals(beforeCount - 6, list.size()); assertEquals("a", processed.get(0)); assertEquals("fail", processed.get(1)); // retry makes it fail once then recover... @@ -327,26 +336,18 @@ public class PollableSourceRetryTests { } - private PollingDispatcher getPollingDispatcher(PollableSource source, MessageTarget target, - PlatformTransactionManager transactionManager, int commitInterval) { - MessageExchangeTemplate template = getExchangeTemplate(transactionManager); - PollingDispatcher dispatcher = new PollingDispatcher(source, new PollingSchedule(100), null, template); - dispatcher.setMaxMessagesPerPoll(commitInterval); - dispatcher.subscribe(target); - return dispatcher; + private SourcePoller getSourcePoller(PollableSource source, MessageChannel channel, + PlatformTransactionManager transactionManager, int maxMessagesPerPoll) { + SourcePoller poller = new SourcePoller(source, channel, new PollingSchedule(100)); + poller.setTransactionManager(transactionManager); + poller.setMaxMessagesPerPoll(maxMessagesPerPoll); + return poller; } - private MessageExchangeTemplate getExchangeTemplate(PlatformTransactionManager transactionManager) { - MessageExchangeTemplate template = new MessageExchangeTemplate(); - template.setTransactionManager(transactionManager); - template.afterPropertiesSet(); - return template; - } - - private DirectChannel getChannel(MessageTarget target) { + private DirectChannel getChannel(MessageConsumer handler) { DirectChannel channel = new DirectChannel(); channel.setBeanName("input"); - channel.subscribe(target); + channel.subscribe(handler); return channel; } @@ -398,7 +399,8 @@ public class PollableSourceRetryTests { * @param methodArgumentsKeyGenerator * @return */ - private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor(MethodArgumentsKeyGenerator methodArgumentsKeyGenerator) { + private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor( + MethodArgumentsKeyGenerator methodArgumentsKeyGenerator) { StatefulRetryOperationsInterceptor advice = new StatefulRetryOperationsInterceptor(); advice.setRecoverer(new MethodInvocationRecoverer() { @SuppressWarnings("unchecked") @@ -416,21 +418,18 @@ public class PollableSourceRetryTests { return advice; } + private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int commitInterval) { + RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor(); + RepeatTemplate repeatTemplate = new RepeatTemplate(); + repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); + advice.setRepeatOperations(repeatTemplate); + return advice; + } /** * @param commitInterval * @return */ - // private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int - // commitInterval) { - // RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor(); - // RepeatTemplate repeatTemplate = new RepeatTemplate(); - // repeatTemplate.setCompletionPolicy(new - // SimpleCompletionPolicy(commitInterval)); - // advice.setRepeatOperations(repeatTemplate); - // return advice; - // } - private Object getProxy(Object target, Class intf, Advice[] advices, String methodName) { ProxyFactory factory = new ProxyFactory(target); for (int i = 0; i < advices.length; i++) { 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 011bd4899..21b4a040e 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 @@ -1,7 +1,7 @@ - - - - + - - - - + + + + - @@ -34,7 +30,5 @@ - - - + \ 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 72351415d..a309dee6e 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 @@ -12,7 +12,9 @@ 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 307a32369..c765146ba 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 @@ -1,11 +1,7 @@ - - - + + - + - - - + + + - + - + - + \ No newline at end of file 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 48fa16248..ff67dd09d 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 @@ -1,9 +1,7 @@ - - - - + - - + - - - - + - - - \ No newline at end of file