BATCH-919: reuse ChunkProcessor in integration

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

@@ -6,8 +6,10 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.listener.StepExecutionListenerSupport;
+import org.springframework.batch.core.step.item.Chunk;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
@@ -19,9 +21,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
 
-	static final String ACTUAL = "ACTUAL";
+	static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName()+".ACTUAL";
 
-	static final String EXPECTED = "EXPECTED";
+	static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName()+".EXPECTED";
 
 	private static final long DEFAULT_THROTTLE_LIMIT = 6;
 	
@@ -54,7 +56,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		if (!items.isEmpty()) {
 
 			logger.debug("Dispatching chunk: " + items);
-			ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState.getSkipCount());
+			ChunkRequest<T> request = new ChunkRequest<T>(new Chunk<T>(items), localState.getJobId(), localState.createStepContribution());
 			messagingGateway.send(request);
 			localState.expected++;
 
@@ -156,8 +158,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 			return expected - actual;
 		}
 
-		public int getSkipCount() {
-			return stepExecution.getSkipCount();
+		public StepContribution createStepContribution() {
+			return stepExecution.createStepContribution();
 		}
 
 		public Long getJobId() {

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

@@ -1,32 +0,0 @@
-/*
- * Copyright 2006-2007 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.batch.integration.chunk;
-
-import java.util.Collection;
-
-import org.springframework.transaction.annotation.Propagation;
-import org.springframework.transaction.annotation.Transactional;
-
-/**
- * @author Dave Syer
- */
-public interface ChunkProcessor<S> {
-
-	// This is transactional with REQUIRES_NEW because we need to force rollback
-	@Transactional(propagation = Propagation.REQUIRES_NEW)
-	int process(Collection<? extends S> items, int skipCount) throws Exception;
-
-}

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

@@ -2,17 +2,17 @@ package org.springframework.batch.integration.chunk;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.step.item.ChunkProcessor;
 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;
 
@@ -29,37 +29,33 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
 	/**
 	 * 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 ChunkHandler#handleChunk(ChunkRequest)
 	 */
 	@ServiceActivator
 	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
 
 		logger.debug("Handling chunk: " + chunkRequest);
 
-		int skipCount = 0;
+		StepContribution stepContribution = chunkRequest.getStepContribution();
 		try {
-			skipCount = chunkProcessor.process(chunkRequest.getItems(),
-					chunkRequest.getSkipCount());
-		} catch (Exception e) {
+			chunkProcessor.process(stepContribution, chunkRequest.getChunk());
+		}
+		catch (Exception e) {
 			logger.debug("Failed chunk", e);
-			return new ChunkResponse(false, chunkRequest.getJobId(), skipCount,
-					e.getClass().getName() + ": " + e.getMessage());
+			return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": "
+					+ e.getMessage());
 		}
 
-		logger.debug("Completed chunk handling with " + skipCount + " skips");
-		return new ChunkResponse(true, chunkRequest.getJobId(), skipCount);
+		logger.debug("Completed chunk handling with " + stepContribution);
+		return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution);
 
 	}
 }

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

@@ -1,38 +1,43 @@
 package org.springframework.batch.integration.chunk;
 
 import java.io.Serializable;
-import java.util.Collection;
+
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.step.item.Chunk;
 
 public class ChunkRequest<T> implements Serializable {
 
-	private final int skipCount;
 	private final Long jobId;
-	private final Collection<? extends T> items;
+	private final Chunk<T> items;
+	private final StepContribution stepContribution;
 
-	public ChunkRequest(Collection<? extends T> items, Long jobId, int skipCount) {
+	public ChunkRequest(Chunk<T> items, Long jobId, StepContribution stepContribution) {
 		this.items = items;
 		this.jobId = jobId;
-		this.skipCount = skipCount;
-	}
-
-	public int getSkipCount() {
-		return skipCount;
+		this.stepContribution = stepContribution;
 	}
 
 	public Long getJobId() {
 		return jobId;
 	}
 
-	public Collection<? extends T> getItems() {
+	public Chunk<T> getChunk() {
 		return items;
 	}
 	
+	/**
+	 * @return the {@link StepContribution} for this chunk
+	 */
+	public StepContribution getStepContribution() {
+		return stepContribution;
+	}
+	
 	/**
 	 * @see java.lang.Object#toString()
 	 */
 	@Override
 	public String toString() {
-		return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", item count="+items.size();
+		return getClass().getSimpleName()+": jobId="+jobId+", contribution="+stepContribution+", item count="+items.size();
 	}
 
 }

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

@@ -2,34 +2,32 @@ package org.springframework.batch.integration.chunk;
 
 import java.io.Serializable;
 
+import org.springframework.batch.core.StepContribution;
+
 public class ChunkResponse implements Serializable {
 
-	private final int skipCount;
+	private final StepContribution stepContribution;
 	private final Long jobId;
 	private final boolean status;
 	private final String message;
 
-	public ChunkResponse(Long jobId) {
-		this(true, jobId, 0, null);
+	public ChunkResponse(Long jobId, StepContribution stepContribution) {
+		this(true, jobId, stepContribution, null);
 	}
 
-	public ChunkResponse(Long jobId, int skipCount) {
-		this(true, jobId, skipCount, null);
+	public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution) {
+		this(status, jobId, stepContribution, null);
 	}
 
-	public ChunkResponse(boolean status, Long jobId, int skipCount) {
-		this(status, jobId, skipCount, null);
-	}
-
-	public ChunkResponse(boolean status, Long jobId, int skipCount, String message) {
+	public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) {
 		this.status = status;
 		this.jobId = jobId;
-		this.skipCount = skipCount;
+		this.stepContribution = stepContribution;
 		this.message = message;
 	}
 
-	public int getSkipCount() {
-		return skipCount;
+	public StepContribution getStepContribution() {
+		return stepContribution;
 	}
 
 	public Long getJobId() {
@@ -49,7 +47,7 @@ public class ChunkResponse implements Serializable {
 	 */
 	@Override
 	public String toString() {
-		return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", successful="+status;
+		return getClass().getSimpleName()+": jobId="+jobId+", stepContribution="+stepContribution+", successful="+status;
 	}
 
 }

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

@@ -1,78 +0,0 @@
-/*
- * Copyright 2006-2007 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.batch.integration.chunk;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.util.Assert;
-
-/**
- * @author Dave Syer
- * 
- */
-public class SimpleChunkProcessor<S, T> implements ChunkProcessor<S>, InitializingBean {
-
-	private ItemProcessor<? super S, ? extends T> itemProcessor;
-
-	private ItemWriter<? super T> itemWriter;
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
-	 */
-	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(itemProcessor, "An ItemProcessor must be provided");
-		Assert.notNull(itemWriter, "An ItemWriter must be provided");
-	}
-
-	/**
-	 * @param itemWriter
-	 */
-	public void setItemWriter(ItemWriter<? super T> itemWriter) {
-		this.itemWriter = itemWriter;
-	}
-	
-	/**
-	 * Public setter for the {@link ItemProcessor}.
-	 * @param itemProcessor the {@link ItemProcessor} to set
-	 */
-	public void setItemProcessor(ItemProcessor<? super S, ? extends T> itemProcessor) {
-		this.itemProcessor = itemProcessor;
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.batch.integration.chunk.ChunkProcessor#process(java.util.Collection,
-	 * int)
-	 */
-	public int process(Collection<? extends S> items, int parentSkipCount) throws Exception {
-
-		List<T> processed = new ArrayList<T>();
-		for (S item : items) {
-			processed.add(itemProcessor.process(item));
-		}
-		itemWriter.write(processed);
-
-		return 0;
-
-	}
-
-}