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 06af85373..181f269cb 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,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 extends StepExecutionListenerSupport implements ItemWriter, ItemStream { @@ -27,10 +24,8 @@ public class ChunkMessageChannelItemWriter 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 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 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 request = new ChunkRequest(items, localState.getJobId(), localState.getSkipCount()); - GenericMessage> message = new GenericMessage>(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 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 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 message = (Message) 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()); } } } 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 c66cbff2f..ee5772c29 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 @@ -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 implements ChunkHandler, InitializingBean { +public class ChunkProcessorChunkHandler implements ChunkHandler, + InitializingBean { - private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class); + private static final Log logger = LogFactory + .getLog(ChunkProcessorChunkHandler.class); private ChunkProcessor 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 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 chunkRequest) { @@ -41,15 +50,16 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, 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); } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java index 7f6659a8c..f6729fbfc 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java +++ b/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; } } 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 8d9dc73e3..4617f1f4b 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 @@ -34,6 +34,7 @@ import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.gateway.SimpleMessagingGateway; import org.springframework.integration.message.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -72,9 +73,13 @@ public class ChunkMessageItemWriterIntegrationTests { factory.setBeanName("step"); factory.setItemWriter(writer); factory.setCommitInterval(4); + + SimpleMessagingGateway gateway = new SimpleMessagingGateway(); + writer.setMessagingGateway(gateway); - writer.setInputChannel(replies); - writer.setOutputChannel(requests); + gateway.setRequestChannel(requests); + gateway.setReplyChannel(replies); + gateway.setReplyTimeout(100); TestItemWriter.count = 0; diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java index ace410876..f3c70cdfc 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java @@ -1,11 +1,11 @@ package org.springframework.batch.integration.chunk; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import java.util.Collection; import org.junit.Test; -import org.springframework.batch.core.BatchStatus; import org.springframework.util.StringUtils; public class ChunkProcessorChunkHandlerTests { @@ -14,7 +14,6 @@ public class ChunkProcessorChunkHandlerTests { protected int count = 0; - @SuppressWarnings("unchecked") @Test public void testVanillaHandleChunk() { handler.setChunkProcessor(new ChunkProcessor() { @@ -23,11 +22,11 @@ public class ChunkProcessorChunkHandlerTests { return 0; } }); - ChunkResponse response = handler.handleChunk(new ChunkRequest(StringUtils.commaDelimitedListToSet("foo,bar"), + ChunkResponse response = handler.handleChunk(new ChunkRequest(StringUtils.commaDelimitedListToSet("foo,bar"), 12L, 10)); assertEquals(0, response.getSkipCount()); assertEquals(new Long(12L), response.getJobId()); - assertEquals(BatchStatus.COMPLETED, response.getStatus()); + assertTrue(response.isSuccessful()); assertEquals(2, count); }