BATCHADM-137 - added optional setting of the replyChannel externally in

the messagechannelparitionhandler

spring-batch-admin-parent/pom.xml

@@ -682,6 +682,12 @@
 				<version>${spring.integration.version}</version>
 				<scope>compile</scope>
 			</dependency>
+			<dependency>
+				<groupId>org.mockito</groupId>
+				<artifactId>mockito-all</artifactId>
+				<version>1.9.5</version>
+				<scope>test</scope>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 	<pluginRepositories>

spring-batch-integration/pom.xml

@@ -140,6 +140,11 @@
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-test</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-all</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 	<build>
 		<plugins>

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<StepExecution> 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<StepExecutionRequest> request = createMessage(count++, split.size(), new StepExecutionRequest(

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<StepExecution> 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<StepExecution> executions = messageChannelPartitionHandler.handle(stepExecutionSplitter, masterStepExecution);
+		//verify
+		assertNotNull(executions);
+		assertTrue(executions.isEmpty());
+		
+	}
+
+}