Use MessagingGateway for chunk writer

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java

@@ -12,10 +12,7 @@ import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.core.Message;
-import org.springframework.integration.core.MessageChannel;
-import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.endpoint.MessagingGateway;
 import org.springframework.util.Assert;
 
 public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>, ItemStream {
@@ -27,10 +24,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	static final String EXPECTED = "EXPECTED";
 
 	private static final long DEFAULT_THROTTLE_LIMIT = 6;
-
-	private MessageChannel target;
-
-	private PollableChannel source;
+	
+	private MessagingGateway messagingGateway;
 
 	// TODO: abstract the state or make a factory for this writer?
 	private LocalState localState = new LocalState();
@@ -46,36 +41,26 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		this.throttleLimit = throttleLimit;
 	}
 
-	// TODO: refactor to ChannelAdapter?
-	public void setInputChannel(PollableChannel source) {
-		this.source = source;
-	}
-
-	// TODO: re-evaluate the refactor to MessageChannel
-	public void setOutputChannel(MessageChannel target) {
-		this.target = target;
+	public void setMessagingGateway(MessagingGateway messagingGateway) {
+		this.messagingGateway = messagingGateway;
 	}
 
 	public void write(List<? extends T> items) throws Exception {
 		// Block until expecting <= throttle limit - can Spring
 		// Integration do that for me?
 		while (localState.getExpecting() > throttleLimit) {
-			getNextResult(100);
+			getNextResult();
 		}
 
 		if (!items.isEmpty()) {
 
 			logger.debug("Dispatching chunk: " + items);
 			ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState.getSkipCount());
-			GenericMessage<ChunkRequest<T>> message = new GenericMessage<ChunkRequest<T>>(request);
-			target.send(message);
+			messagingGateway.send(request);
 			localState.expected++;
 
 		}
 
-		// Short little timeout to look for an immediate reply.
-		getNextResult(1);
-
 	}
 
 	@Override
@@ -137,7 +122,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		int count = 0;
 		int maxCount = 40;
 		while (localState.getExpecting() > 0 && count++ < maxCount) {
-			getNextResult(100);
+			getNextResult();
 		}
 		return count < maxCount;
 	}
@@ -146,20 +131,17 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	 * Get the next result if it is available within the timeout specified,
 	 * otherwise return null.
 	 */
-	private void getNextResult(long timeout) {
-		@SuppressWarnings("unchecked")
-		Message<ChunkResponse> message = (Message<ChunkResponse>) source.receive(timeout);
-		if (message != null) {
-			ChunkResponse payload = message.getPayload();
+	private void getNextResult() {
+		ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
+		if (payload != null) {
 			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 ["
 					+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
 			localState.actual++;
 			// TODO: apply the skip count
-			BatchStatus result = payload.getStatus();
-			if (BatchStatus.COMPLETED!=result) {
-				throw new AsynchronousFailureException("Failure or early completion detected in handler: " + result);
+			if (! payload.isSuccessful()) {
+				throw new AsynchronousFailureException("Failure or interrupt detected in handler: "+payload.getMessage());
 			}
 		}
 	}

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java

@@ -2,37 +2,46 @@ package org.springframework.batch.integration.chunk;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.core.BatchStatus;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.util.Assert;
 
 @MessageEndpoint
-public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
+public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
+		InitializingBean {
 
-	private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
+	private static final Log logger = LogFactory
+			.getLog(ChunkProcessorChunkHandler.class);
 
 	private ChunkProcessor<S> chunkProcessor;
-	
-	/* (non-Javadoc)
-	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 */
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(chunkProcessor, "A ChunkProcessor must be provided");
 	}
-	
+
 	/**
 	 * Public setter for the {@link ChunkProcessor}.
-	 * @param chunkProcessor the chunkProcessor to set
+	 * 
+	 * @param chunkProcessor
+	 *            the chunkProcessor to set
 	 */
 	public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
 		this.chunkProcessor = chunkProcessor;
 	}
-	
+
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection)
+	 * 
+	 * @see
+	 * org.springframework.integration.batch.slave.ChunkHandler#handleChunk(
+	 * java.util.Collection)
 	 */
 	@ServiceActivator
 	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
@@ -41,15 +50,16 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
 
 		int skipCount = 0;
 		try {
-			skipCount = chunkProcessor.process(chunkRequest.getItems(), chunkRequest.getSkipCount());
-		}
-		catch (Exception e) {
+			skipCount = chunkProcessor.process(chunkRequest.getItems(),
+					chunkRequest.getSkipCount());
+		} catch (Exception e) {
 			logger.debug("Failed chunk", e);
-			return new ChunkResponse(BatchStatus.FAILED, chunkRequest.getJobId(), skipCount);
+			return new ChunkResponse(false, chunkRequest.getJobId(), skipCount,
+					e.getClass().getName() + ": " + e.getMessage());
 		}
 
 		logger.debug("Completed chunk handling with " + skipCount + " skips");
-		return new ChunkResponse(BatchStatus.COMPLETED, chunkRequest.getJobId(), skipCount);
+		return new ChunkResponse(true, chunkRequest.getJobId(), skipCount);
 
 	}
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java

@@ -2,18 +2,30 @@ package org.springframework.batch.integration.chunk;
 
 import java.io.Serializable;
 
-import org.springframework.batch.core.BatchStatus;
-
 public class ChunkResponse implements Serializable {
 
 	private final int skipCount;
 	private final Long jobId;
-	private final BatchStatus status;
+	private final boolean status;
+	private final String message;
 
-	public ChunkResponse(BatchStatus status, Long jobId, int skipCount) {
+	public ChunkResponse(Long jobId) {
+		this(true, jobId, 0, null);
+	}
+
+	public ChunkResponse(Long jobId, int skipCount) {
+		this(true, jobId, skipCount, null);
+	}
+
+	public ChunkResponse(boolean status, Long jobId, int skipCount) {
+		this(status, jobId, skipCount, null);
+	}
+
+	public ChunkResponse(boolean status, Long jobId, int skipCount, String message) {
 		this.status = status;
 		this.jobId = jobId;
 		this.skipCount = skipCount;
+		this.message = message;
 	}
 
 	public int getSkipCount() {
@@ -24,16 +36,20 @@ public class ChunkResponse implements Serializable {
 		return jobId;
 	}
 
-	public BatchStatus getStatus() {
+	public boolean isSuccessful() {
 		return status;
 	}
 
+	public String getMessage() {
+		return message;
+	}
+
 	/**
 	 * @see java.lang.Object#toString()
 	 */
 	@Override
 	public String toString() {
-		return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", status="+status;
+		return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", successful="+status;
 	}
 
 }