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 11d2cc8e9..2f427abaf 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 @@ -16,9 +16,10 @@ import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; -import org.springframework.integration.channel.MessageChannel; +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.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; @@ -37,9 +38,9 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private static final long DEFAULT_THROTTLE_LIMIT = 6; - private MessageChannel requestChannel; + private MessageTarget target; - private MessageChannel replyChannel; + private BlockingSource source; // TODO: abstract the state or make a factory for this writer? private LocalState localState = new LocalState(); @@ -55,12 +56,12 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo this.throttleLimit = throttleLimit; } - public void setReplyChannel(MessageChannel replyChannel) { - this.replyChannel = replyChannel; + public void setSource(BlockingSource source) { + this.source = source; } - public void setRequestChannel(MessageChannel requestChannel) { - this.requestChannel = requestChannel; + public void setTarget(MessageTarget target) { + this.target = target; } public void write(T item) throws Exception { @@ -95,7 +96,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo logger.debug("Dispatching chunk: " + processed); ChunkRequest request = new ChunkRequest(processed, localState.getJobId(), localState.getSkipCount()); GenericMessage> message = new GenericMessage>(request); - requestChannel.send(message); + target.send(message); localState.expected++; } @@ -176,9 +177,10 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * otherwise return null. */ private void getNextResult(long timeout) { - Message message = replyChannel.receive(timeout); + // TODO: use the timeout + Message message = source.receive(timeout); if (message != null) { - ChunkResponse payload = (ChunkResponse) message.getPayload(); + ChunkResponse payload = message.getPayload(); Long jobInstanceId = payload.getJobId(); Assert.state(jobInstanceId != null, "Message did not contain job instance id."); Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id [" 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 95b5fa7ca..6bbf61853 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,9 +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.channel.MessageChannel; +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.util.Assert; /** @@ -39,9 +40,9 @@ public class MessageOrientedStep extends AbstractStep { */ public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING"; - private MessageChannel requestChannel; + private MessageTarget target; - private MessageChannel replyChannel; + private BlockingSource source; private static int MINUTE = 1000 * 60; @@ -75,21 +76,21 @@ public class MessageOrientedStep extends AbstractStep { } /** - * Public setter for the requestChannel. - * @param requestChannel the requestChannel to set + * Public setter for the target. + * @param target the target to set */ @Required - public void setRequestChannel(MessageChannel requestChannel) { - this.requestChannel = requestChannel; + public void setTarget(MessageTarget target) { + this.target = target; } /** - * Public setter for the replyChannel. - * @param replyChannel the replyChannel to set + * Public setter for the source. + * @param source the source to set */ @Required - public void setReplyChannel(MessageChannel replyChannel) { - this.replyChannel = replyChannel; + public void setSource(BlockingSource source) { + this.source = source; } /* @@ -112,7 +113,7 @@ public class MessageOrientedStep extends AbstractStep { executionContext.putString(WAITING, "true"); // TODO: need these two lines to be atomic getJobRepository().update(stepExecution); - requestChannel.send(new GenericMessage(request)); + target.send(new GenericMessage(request)); waitForReply(request.getJobId()); } @@ -128,14 +129,14 @@ public class MessageOrientedStep extends AbstractStep { long maxCount = executionTimeout / timeout; long count = 0; - // TODO: use a ReponseCorrelator?, or just a SynchronousChannel while (count++ < maxCount) { - Message message = replyChannel.receive(timeout); + // TODO: timeout? + Message message = source.receive(timeout); if (message != null) { - JobExecutionRequest payload = (JobExecutionRequest) message.getPayload(); + JobExecutionRequest payload = message.getPayload(); Long jobInstanceId = payload.getJobId(); Assert.state(jobInstanceId != null, "Message did not contain job instance id."); Assert.state(jobInstanceId.equals(expectedJobId), "Message contained wrong job instance id [" 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 06d4ab800..55adf9d81 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 @@ -10,6 +10,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.test.context.ContextConfiguration; @@ -26,7 +27,7 @@ public class SmokeTests { @Autowired @Qualifier("smokeout") - private MessageChannel smokeout; + private BlockingSource 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 @@ -48,7 +49,7 @@ public class SmokeTests { @Test public void testVanillaSendAndReceive() throws Exception { smokein.send(new GenericMessage("foo")); - Message message = smokeout.receive(100); + 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 2b0c9e1ab..9d9b6ae12 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,7 +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; import org.springframework.test.context.ContextConfiguration; @@ -47,11 +47,11 @@ public class ChunkMessageItemWriterIntegrationTests { @Autowired @Qualifier("requests") - private MessageChannel requests; + private PollableChannel requests; @Autowired @Qualifier("replies") - private MessageChannel replies; + private PollableChannel replies; private SimpleStepFactoryBean factory; @@ -59,6 +59,7 @@ public class ChunkMessageItemWriterIntegrationTests { private static long jobCounter; + @SuppressWarnings("unchecked") @Before public void setUp() { @@ -71,8 +72,8 @@ public class ChunkMessageItemWriterIntegrationTests { factory.setItemWriter(writer); factory.setCommitInterval(4); - writer.setReplyChannel(replies); - writer.setRequestChannel(requests); + writer.setSource(replies); + writer.setTarget(requests); TestItemWriter.count = 0; 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 89918ba0a..7d81c85ca 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 @@ -28,6 +28,7 @@ import org.springframework.core.io.Resource; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Splitter; 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.test.context.ContextConfiguration; @@ -48,7 +49,7 @@ public class ResourceSplitterIntegrationTests { @Autowired @Qualifier("requests") - private MessageChannel requests; + private PollableChannel requests; /* * This is so cool (but see INT-190)...
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java index 0e1d9ab3a..3d5589288 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java @@ -22,7 +22,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.item.ItemWriter; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.Message; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -36,7 +36,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class MessageChannelItemWriterIntegrationTests { @Autowired - private MessageChannel channel; + private PollableChannel channel; @Autowired private ItemWriter itemWriter; 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 a803f93e7..03bdfd3bd 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 @@ -33,11 +33,13 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.integration.JobRepositorySupport; import org.springframework.beans.factory.annotation.Required; import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.util.ReflectionUtils; @@ -53,15 +55,16 @@ public class MessageOrientedStepTests { private DirectChannel requestChannel; - private MessageChannel replyChannel; + private PollableChannel replyChannel; + @SuppressWarnings("unchecked") @Before public void createStep() { replyChannel = new ThreadLocalChannel(); requestChannel = new DirectChannel(); step.setName("step"); - step.setRequestChannel(requestChannel); - step.setReplyChannel(replyChannel); + step.setTarget(requestChannel); + step.setSource(replyChannel); step.setStartLimit(10); step.setJobRepository(new JobRepositorySupport()); JobInstance jobInstance = new JobInstance(0L, new JobParameters(), "job"); @@ -70,12 +73,12 @@ public class MessageOrientedStepTests { /** * Test method for - * {@link org.springframework.batch.integration.job.MessageOrientedStep#setRequestChannel(org.springframework.integration.channel.MessageChannel)}. + * {@link org.springframework.batch.integration.job.MessageOrientedStep#setTarget(MessageTarget)}. */ @Test public void testSetRequestChannel() { - Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setRequestChannel", - new Class[] { MessageChannel.class }); + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setTarget", + new Class[] { MessageTarget.class }); assertNotNull(method); Annotation[] annotations = AnnotationUtils.getAnnotations(method); assertEquals(1, annotations.length); @@ -84,12 +87,12 @@ public class MessageOrientedStepTests { /** * Test method for - * {@link org.springframework.batch.integration.job.MessageOrientedStep#setReplyChannel(org.springframework.integration.channel.MessageChannel)}. + * {@link org.springframework.batch.integration.job.MessageOrientedStep#setSource(MessageSource)}. */ @Test public void testSetReplyChannel() { - Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setReplyChannel", - new Class[] { MessageChannel.class }); + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource", + new Class[] { BlockingSource.class }); assertNotNull(method); Annotation[] annotations = AnnotationUtils.getAnnotations(method); assertEquals(1, annotations.length); 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 160939a49..c05c44264 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 @@ -15,7 +15,7 @@ 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.MessageHandlingException; @@ -30,11 +30,11 @@ public class JobLaunchingMessageHandlerIntegrationTests { @Autowired @Qualifier("requests") - private MessageChannel requestChannel; + private PollableChannel requestChannel; @Autowired @Qualifier("response") - private MessageChannel responseChannel; + private PollableChannel responseChannel; private JobSupport job = new JobSupport("testJob"); 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 8d9f381db..7f94caee9 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 @@ -37,6 +37,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; @@ -385,7 +386,7 @@ public class PollableSourceRetryTests { * @return */ private DirectChannel getChannel(MessageTarget handler, MessageSource source) { - DirectChannel channel = new DirectChannel(source); + DirectChannel channel = new DirectChannel(); channel.setName("input"); channel.subscribe(handler); return channel; @@ -400,7 +401,7 @@ public class PollableSourceRetryTests { lifecycle.stop(); } - private MessageSource getPollableSource(List list) { + private PollableSource getPollableSource(List list) { final ItemReader reader = new ListItemReader(list) { public String read() { String item = super.read(); @@ -408,10 +409,12 @@ public class PollableSourceRetryTests { return item; } }; - MessageSource source = new MessageSource() { + PollableSource source = new PollableSource() { public Message receive() { try { - return new GenericMessage(reader.read()); + String payload = reader.read(); + if (payload==null) return null; + return new GenericMessage(payload); } catch (RuntimeException e) { throw e;