diff --git a/spring-batch-admin-parent/pom.xml b/spring-batch-admin-parent/pom.xml
index 134f3bcc9..6e159319a 100755
--- a/spring-batch-admin-parent/pom.xml
+++ b/spring-batch-admin-parent/pom.xml
@@ -682,6 +682,12 @@
${spring.integration.version}
compile
+
+ org.mockito
+ mockito-all
+ 1.9.5
+ test
+
diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml
index 524aef461..c11991565 100644
--- a/spring-batch-integration/pom.xml
+++ b/spring-batch-integration/pom.xml
@@ -140,6 +140,11 @@
org.springframework
spring-test
+
+ org.mockito
+ mockito-all
+ test
+
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
index 409a08fde..b0fa801d8 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
@@ -32,6 +32,7 @@ import org.springframework.util.Assert;
* Spring Batch {@link JobRepository} so that the shared state across those restarts can be managed centrally.
*
* @author Dave Syer
+ * @author Will Schipp
*
*/
@MessageEndpoint
@@ -44,6 +45,11 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
private MessagingOperations messagingGateway;
private String stepName;
+
+ /**
+ * pollable channel for the replies
+ */
+ private PollableChannel replyChannel;
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
@@ -95,6 +101,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
return messages;
}
+ public void setReplyChannel(PollableChannel replyChannel) {
+ this.replyChannel = replyChannel;
+ }
+
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the {@link MessagingOperations}, and then
* receives the result back as a list of {@link StepExecution} on a reply channel. Use the {@link #aggregate(List)}
@@ -108,7 +118,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
Set split = stepExecutionSplitter.split(masterStepExecution, gridSize);
int count = 0;
- PollableChannel replyChannel = new QueueChannel();
+
+ if (replyChannel == null) {
+ replyChannel = new QueueChannel();
+ }//end if
for (StepExecution stepExecution : split) {
Message request = createMessage(count++, split.size(), new StepExecutionRequest(
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandlerTest.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandlerTest.java
new file mode 100644
index 000000000..6c425a036
--- /dev/null
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandlerTest.java
@@ -0,0 +1,75 @@
+package org.springframework.batch.integration.partition;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.junit.Test;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.partition.StepExecutionSplitter;
+import org.springframework.integration.Message;
+import org.springframework.integration.core.MessagingOperations;
+import org.springframework.integration.core.PollableChannel;
+
+/**
+ *
+ * @author Will Schipp
+ *
+ */
+public class MessageChannelPartitionHandlerTest {
+
+ private MessageChannelPartitionHandler messageChannelPartitionHandler;
+
+ @Test
+ public void testHandleNoReply() throws Exception {
+ //execute with no default set
+ messageChannelPartitionHandler = new MessageChannelPartitionHandler();
+ //mock
+ StepExecution masterStepExecution = mock(StepExecution.class);
+ StepExecutionSplitter stepExecutionSplitter = mock(StepExecutionSplitter.class);
+ MessagingOperations operations = mock(MessagingOperations.class);
+ Message message = mock(Message.class);
+ //when
+ when(message.getPayload()).thenReturn(Collections.emptyList());
+ when(operations.receive((PollableChannel) anyObject())).thenReturn(message);
+ //set
+ messageChannelPartitionHandler.setMessagingOperations(operations);
+
+ //execute
+ Collection executions = messageChannelPartitionHandler.handle(stepExecutionSplitter, masterStepExecution);
+ //verify
+ assertNotNull(executions);
+ assertTrue(executions.isEmpty());
+ }
+
+ @Test
+ public void testHandleWithReplyChannel() throws Exception {
+ //execute with no default set
+ messageChannelPartitionHandler = new MessageChannelPartitionHandler();
+ //mock
+ StepExecution masterStepExecution = mock(StepExecution.class);
+ StepExecutionSplitter stepExecutionSplitter = mock(StepExecutionSplitter.class);
+ MessagingOperations operations = mock(MessagingOperations.class);
+ Message message = mock(Message.class);
+ PollableChannel replyChannel = mock(PollableChannel.class);
+ //when
+ when(message.getPayload()).thenReturn(Collections.emptyList());
+ when(operations.receive(replyChannel)).thenReturn(message);
+ //set
+ messageChannelPartitionHandler.setMessagingOperations(operations);
+ messageChannelPartitionHandler.setReplyChannel(replyChannel);
+
+ //execute
+ Collection executions = messageChannelPartitionHandler.handle(stepExecutionSplitter, masterStepExecution);
+ //verify
+ assertNotNull(executions);
+ assertTrue(executions.isEmpty());
+
+ }
+
+}