RESOLVED - issue BATCH-231: Chunk completion policy

http://jira.springframework.org/browse/BATCH-231

Added RepeatOperationsStepFactoryBean.

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java

@@ -106,6 +106,22 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
 		this.itemWriter = itemWriter;
 	}
 
+	/**
+	 * Protected getter for the {@link ItemReader} for subclasses to use.
+	 * @return the itemReader
+	 */
+	protected ItemReader getItemReader() {
+		return itemReader;
+	}
+
+	/**
+	 * Protected getter for the {@link ItemWriter} for subclasses to use
+	 * @return the itemWriter
+	 */
+	protected ItemWriter getItemWriter() {
+		return itemWriter;
+	}
+
 	/**
 	 * Public setter for {@link JobRepository}.
 	 * 
@@ -154,20 +170,4 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
 		return Step.class;
 	}
 
-	/**
-	 * Public getter for the ItemReader.
-	 * @return the itemReader
-	 */
-	public ItemReader getItemReader() {
-		return itemReader;
-	}
-
-	/**
-	 * Public getter for the ItemWriter.
-	 * @return the itemWriter
-	 */
-	public ItemWriter getItemWriter() {
-		return itemWriter;
-	}
-
 }

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/BatchListenerFactoryHelper.java

@@ -0,0 +1,173 @@
+/*
+ * 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.execution.step.support;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.batch.core.domain.BatchListener;
+import org.springframework.batch.core.domain.ChunkListener;
+import org.springframework.batch.core.domain.ItemReadListener;
+import org.springframework.batch.core.domain.ItemWriteListener;
+import org.springframework.batch.core.domain.StepListener;
+import org.springframework.batch.execution.listener.CompositeChunkListener;
+import org.springframework.batch.execution.listener.CompositeItemReadListener;
+import org.springframework.batch.execution.listener.CompositeItemWriteListener;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.reader.DelegatingItemReader;
+import org.springframework.batch.item.writer.DelegatingItemWriter;
+import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.repeat.listener.RepeatListenerSupport;
+import org.springframework.batch.repeat.support.RepeatTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * Package private helper for step factory beans.
+ * 
+ * @author Dave Syer
+ * 
+ */
+class BatchListenerFactoryHelper {
+
+	/**
+	 * @param itemReader2
+	 * @param listeners
+	 * @return
+	 */
+	public ItemReader getItemReader(ItemReader itemReader, BatchListener[] listeners) {
+
+		final CompositeItemReadListener multicaster = new CompositeItemReadListener();
+
+		for (int i = 0; i < listeners.length; i++) {
+			BatchListener listener = listeners[i];
+			if (listener instanceof ItemReadListener) {
+				multicaster.register((ItemReadListener) listener);
+			}
+		}
+
+		itemReader = new DelegatingItemReader(itemReader) {
+			public Object read() throws Exception {
+				try {
+					multicaster.beforeRead();
+					Object item = super.read();
+					multicaster.afterRead(item);
+					return item;
+				}
+				catch (Exception e) {
+					multicaster.onReadError(e);
+					throw e;
+				}
+			}
+		};
+
+		return itemReader;
+	}
+
+	/**
+	 * @param itemWriter2
+	 * @param listeners
+	 * @return
+	 */
+	public ItemWriter getItemWriter(ItemWriter itemWriter, BatchListener[] listeners) {
+		final CompositeItemWriteListener multicaster = new CompositeItemWriteListener();
+
+		for (int i = 0; i < listeners.length; i++) {
+			BatchListener listener = listeners[i];
+			if (listener instanceof ItemWriteListener) {
+				multicaster.register((ItemWriteListener) listener);
+			}
+		}
+
+		itemWriter = new DelegatingItemWriter(itemWriter) {
+			public void write(Object item) throws Exception {
+				try {
+					multicaster.beforeWrite(item);
+					super.write(item);
+					multicaster.afterWrite();
+				}
+				catch (Exception e) {
+					multicaster.onWriteError(e, item);
+					throw e;
+				}
+			}
+		};
+		
+		return itemWriter;
+
+	}
+
+	/**
+	 * @param stepOperations
+	 * @param listeners
+	 * @return
+	 */
+	public RepeatOperations getStepOperations(RepeatOperations stepOperations, BatchListener[] listeners) {
+
+		final CompositeChunkListener multicaster = new CompositeChunkListener();
+
+		boolean hasChunkListener = false;
+
+		for (int i = 0; i < listeners.length; i++) {
+			BatchListener listener = listeners[i];
+			if (listener instanceof ChunkListener) {
+				hasChunkListener = true;
+			}
+			if (listener instanceof ChunkListener) {
+				multicaster.register((ChunkListener) listener);
+			}
+		}
+
+		if (hasChunkListener) {
+
+			Assert.state(stepOperations instanceof RepeatTemplate,
+					"Step operations is injected but not a RepeatTemplate, so chunk listeners cannot also be registered. "
+							+ "Either inject a RepeatTemplate, or remove the ChunkListener.");
+
+			RepeatTemplate stepTemplate = (RepeatTemplate) stepOperations;
+			stepTemplate.registerListener(new RepeatListenerSupport() {
+				public void open(RepeatContext context) {
+					multicaster.beforeChunk();
+				}
+
+				public void close(RepeatContext context) {
+					multicaster.afterChunk();
+				}
+			});
+
+		}
+
+		return stepOperations;
+
+	}
+
+	/**
+	 * @param listeners
+	 * @return
+	 */
+	public StepListener[] getStepListeners(BatchListener[] listeners) {
+		List list = new ArrayList();
+		for (int i = 0; i < listeners.length; i++) {
+			BatchListener listener = listeners[i];
+			if (listener instanceof StepListener) {
+				list.add(listener);
+			}
+		}
+		return (StepListener[]) list.toArray(new StepListener[list.size()]);
+	}
+
+}

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java

@@ -22,11 +22,7 @@ import org.springframework.batch.execution.step.ItemOrientedStep;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.item.reader.DelegatingItemReader;
-import org.springframework.batch.item.writer.DelegatingItemWriter;
-import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler;
-import org.springframework.batch.repeat.listener.RepeatListenerSupport;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
 import org.springframework.core.task.TaskExecutor;
@@ -48,7 +44,7 @@ public class DefaultStepFactoryBean extends SimpleStepFactoryBean {
 	private TaskExecutor taskExecutor;
 
 	/**
-	 * Public setter for the flag that determines skip policy. If this flag is
+	 * Public setter for a flag that determines skip policy. If this flag is
 	 * true then an exception in chunk processing will cause the item to be
 	 * skipped and no exceptions propagated. If it is false then all exceptions
 	 * will be propagated from the chunk and cause the step to abort.
@@ -116,51 +112,21 @@ public class DefaultStepFactoryBean extends SimpleStepFactoryBean {
 			step.registerStepListener((StepListener) itemWriter);
 		}
 
-		itemReader = new DelegatingItemReader(itemReader) {
-			public Object read() throws Exception {
-				try {
-					listener.beforeRead();
-					Object item = super.read();
-					listener.afterRead(item);
-					return item;
-				}
-				catch (Exception e) {
-					listener.onReadError(e);
-					throw e;
-				}
-			}
-		};
-		// In case it is used by subclasses:
-		setItemReader(itemReader);
-		step.setItemReader(itemReader);
-
-		itemWriter = new DelegatingItemWriter(itemWriter) {
-			public void write(Object item) throws Exception {
-				try {
-					listener.beforeWrite(item);
-					super.write(item);
-					listener.afterWrite();
-				}
-				catch (Exception e) {
-					listener.onWriteError(e, item);
-					throw e;
-				}
-			}
-		};
-		// In case it is used by subclasses:
-		setItemWriter(itemWriter);
-		step.setItemWriter(itemWriter);
+		BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
 
+		StepListener[] stepListeners = helper.getStepListeners(listeners);
+		itemReader = helper.getItemReader(itemReader, listeners);
+		itemWriter = helper.getItemWriter(itemWriter, listeners);
 		RepeatTemplate stepOperations = new RepeatTemplate();
-		stepOperations.setListener(new RepeatListenerSupport() {
-			public void open(RepeatContext context) {
-				listener.beforeChunk();
-			}
+		stepOperations = (RepeatTemplate) helper.getStepOperations(stepOperations, listeners);
 
-			public void close(RepeatContext context) {
-				listener.afterChunk();
-			}
-		});
+		// In case they are used by subclasses:
+		setItemReader(itemReader);
+		setItemWriter(itemWriter);
+
+		step.setStepListeners(stepListeners);
+		step.setItemReader(itemReader);
+		step.setItemWriter(itemWriter);
 
 		if (taskExecutor != null) {
 			TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java

@@ -0,0 +1,141 @@
+/*
+ * 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.execution.step.support;
+
+import org.springframework.batch.core.domain.BatchListener;
+import org.springframework.batch.core.domain.Step;
+import org.springframework.batch.core.domain.StepListener;
+import org.springframework.batch.execution.step.ItemOrientedStep;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.repeat.support.RepeatTemplate;
+
+/**
+ * Extends a {@link SimpleStepFactoryBean} allowing registration of listeners
+ * and also direct injection of the {@link RepeatOperations} needed at step and
+ * chunk level.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
+
+	private ItemStream[] streams = new ItemStream[0];
+
+	private BatchListener[] listeners = new BatchListener[0];
+
+	private RepeatOperations chunkOperations = new RepeatTemplate();
+
+	private RepeatOperations stepOperations = new RepeatTemplate();
+
+	/**
+	 * The streams to inject into the {@link Step}. Any instance of
+	 * {@link ItemStream} can be used, and will then receive callbacks at the
+	 * appropriate stage in the step.
+	 * 
+	 * @param streams an array of listeners
+	 */
+	public void setStreams(ItemStream[] streams) {
+		this.streams = streams;
+	}
+
+	/**
+	 * The listeners to inject into the {@link Step}. Any instance of
+	 * {@link BatchListener} can be used, and will then receive callbacks at the
+	 * appropriate stage in the step.
+	 * 
+	 * @param listeners an array of listeners
+	 */
+	public void setListeners(BatchListener[] listeners) {
+		this.listeners = listeners;
+	}
+
+	/**
+	 * The {@link RepeatOperations} to use for the outer loop of the batch
+	 * processing. Should be set up by the caller through a factory. Defaults to
+	 * a plain {@link RepeatTemplate}.
+	 * 
+	 * @param stepOperations a {@link RepeatOperations} instance.
+	 */
+	public void setStepOperations(RepeatOperations stepOperations) {
+		this.stepOperations = stepOperations;
+	}
+
+	/**
+	 * The {@link RepeatOperations} to use for the inner loop of the batch
+	 * processing. should be set up by the caller through a factory. defaults to
+	 * a plain {@link RepeatTemplate}.
+	 * 
+	 * @param chunkOperations a {@link RepeatOperations} instance.
+	 */
+	public void setChunkOperations(RepeatOperations chunkOperations) {
+		this.chunkOperations = chunkOperations;
+	}
+
+	/**
+	 * @param step
+	 * 
+	 */
+	protected void applyConfiguration(ItemOrientedStep step) {
+
+		super.applyConfiguration(step);
+
+		step.setStreams(streams);
+
+		ItemReader itemReader = getItemReader();
+		ItemWriter itemWriter = getItemWriter();
+
+		/*
+		 * Since we are going to wrap these things with listener callbacks we
+		 * need to register them here because the step will not know we did
+		 * that.
+		 */
+		if (itemReader instanceof ItemStream) {
+			step.registerStream((ItemStream) itemReader);
+		}
+		if (itemReader instanceof StepListener) {
+			step.registerStepListener((StepListener) itemReader);
+		}
+		if (itemWriter instanceof ItemStream) {
+			step.registerStream((ItemStream) itemWriter);
+		}
+		if (itemWriter instanceof StepListener) {
+			step.registerStepListener((StepListener) itemWriter);
+		}
+
+		BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
+
+		StepListener[] stepListeners = helper.getStepListeners(listeners);
+		itemReader = helper.getItemReader(itemReader, listeners);
+		itemWriter = helper.getItemWriter(itemWriter, listeners);
+		RepeatOperations stepOperations = helper.getStepOperations(this.stepOperations, listeners);
+
+		// In case they are used by subclasses:
+		setItemReader(itemReader);
+		setItemWriter(itemWriter);
+
+		step.setStepListeners(stepListeners);
+		step.setItemReader(itemReader);
+		step.setItemWriter(itemWriter);
+
+		step.setChunkOperations(chunkOperations);
+		step.setStepOperations(stepOperations);
+
+	}
+
+}

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java

@@ -34,9 +34,6 @@ import org.springframework.batch.retry.support.RetryTemplate;
  * limit given by the {@link RetryPolicy}. When the retry is exhausted instead
  * of the item being skipped it is handled by an {@link ItemRecoverer}.<br/>
  * 
- * TODO: make sure listeners are called, and add item listener callbacks to the
- * recovery path.
- * 
  * TODO: checking for null retry callback is a sucky way of determining if a
  * stateful retry has been requested.
  * 

spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeJobListenerTests.java

@@ -57,7 +57,7 @@ public class CompositeJobListenerTests extends TestCase {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.execution.listener.CompositeJobListener#setListener(org.springframework.batch.core.domain.JobListener)}.
+	 * {@link org.springframework.batch.execution.listener.CompositeJobListener#registerListener(org.springframework.batch.core.domain.JobListener)}.
 	 */
 	public void testSetListener() {
 		listener.register(new JobListenerSupport() {

spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeStepListenerTests.java

@@ -58,7 +58,7 @@ public class CompositeStepListenerTests extends TestCase {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.execution.listener.CompositeStepListener#setListener(org.springframework.batch.core.domain.StepListener)}.
+	 * {@link org.springframework.batch.execution.listener.CompositeStepListener#registerListener(org.springframework.batch.core.domain.StepListener)}.
 	 */
 	public void testSetListener() {
 		listener.register(new StepListenerSupport() {

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBeanTests.java

@@ -0,0 +1,81 @@
+/*
+ * 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.execution.step.support;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.springframework.batch.core.domain.JobExecution;
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.JobParameters;
+import org.springframework.batch.core.domain.Step;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.execution.job.JobSupport;
+import org.springframework.batch.execution.launch.EmptyItemWriter;
+import org.springframework.batch.item.reader.ListItemReader;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.repeat.RepeatCallback;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class RepeatOperationsStepFactoryBeanTests extends TestCase {
+
+	private RepeatOperationsStepFactoryBean factory = new RepeatOperationsStepFactoryBean();
+
+	private List list;
+
+	private JobExecution jobExecution = new JobExecution(new JobInstance(new Long(0L), new JobParameters(),
+			new JobSupport("job")));;
+
+	public void testType() throws Exception {
+		assertEquals(Step.class, factory.getObjectType());
+	}
+
+	public void testDefaultValue() throws Exception {
+		assertTrue(factory.getObject() instanceof Step);
+	}
+
+	public void testStepOperationsWithoutChunkListener() throws Exception {
+
+		factory.setItemReader(new ListItemReader(new ArrayList()));
+		factory.setItemWriter(new EmptyItemWriter());
+		factory.setJobRepository(new JobRepositorySupport());
+		factory.setTransactionManager(new ResourcelessTransactionManager());
+
+		factory.setStepOperations(new RepeatOperations() {
+
+			public ExitStatus iterate(RepeatCallback callback) {
+				list = new ArrayList();
+				list.add("foo");
+				return ExitStatus.FINISHED;
+			}
+		});
+
+		factory.setSingleton(false);
+
+		Step step = (Step) factory.getObject();
+		step.execute(new StepExecution(step, jobExecution));
+
+		assertEquals(1, list.size());
+	}
+
+}