BATCH-378: Added ChunkListener, ItemReadListener, and ItemWriterListener. Also added them to the ListenerMulticaster.

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

@@ -273,8 +273,9 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 
 					try {
 						itemReader.mark();
+						listener.beforeChunk();
 						result = processChunk(contribution);
-
+						listener.afterChunk();
 						contribution.incrementCommitCount();
 
 						// If the step operations are asynchronous then we need
@@ -528,21 +529,25 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 		if (retryCallback == null) {
 			Object item = null;
 			try {
+				listener.beforeRead();
 				item = itemReader.read();
+				listener.afterRead(item);
 			}
 			catch (Exception ex) {
-				itemFailureHandler.handleReadFailure(ex);
+				listener.onReadError(ex);
 				throw ex;
 			}
 			if (item == null) {
 				return ExitStatus.FINISHED;
 			}
 			try {
+				listener.beforeWrite(item);
 				itemWriter.write(item);
+				listener.afterWrite();
 			}
 			catch (Exception e) {
 
-				itemFailureHandler.handleWriteFailure(item, e);
+				listener.onWriteError(e, item);
 				// Re-throw the exception so that the surrounding transaction
 				// rolls back if there is one
 				throw e;

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

@@ -15,8 +15,14 @@
  */
 package org.springframework.batch.execution.step.support;
 
+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.StepExecution;
 import org.springframework.batch.core.domain.StepListener;
+import org.springframework.batch.core.interceptor.CompositeChunkListener;
+import org.springframework.batch.core.interceptor.CompositeItemReadListener;
+import org.springframework.batch.core.interceptor.CompositeItemWriteListener;
 import org.springframework.batch.core.interceptor.CompositeStepListener;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStream;
@@ -28,11 +34,17 @@ import org.springframework.batch.repeat.ExitStatus;
  * @author Dave Syer
  * 
  */
-public class ListenerMulticaster implements ItemStream, StepListener {
+public class ListenerMulticaster implements ItemStream, StepListener, ChunkListener, ItemReadListener, ItemWriteListener {
 
 	private CompositeItemStream stream = new CompositeItemStream();
 
 	private CompositeStepListener stepListener = new CompositeStepListener();
+	
+	private CompositeChunkListener chunkListener = new CompositeChunkListener();
+	
+	private CompositeItemReadListener itemReadListener = new CompositeItemReadListener();
+	
+	private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener();
 
 	/**
 	 * Register each of the objects as listeners. Once registered, calls to the
@@ -58,6 +70,15 @@ public class ListenerMulticaster implements ItemStream, StepListener {
 		if (listener instanceof ItemStream) {
 			this.stream.register((ItemStream) listener);
 		}
+		if(listener instanceof ChunkListener){
+			this.chunkListener.register((ChunkListener)listener);
+		}
+		if(listener instanceof ItemReadListener){
+			this.itemReadListener.register((ItemReadListener)listener);
+		}
+		if(listener instanceof ItemWriteListener){
+			this.itemWriteListener.register((ItemWriteListener)listener);
+		}
 	}
 
 	/**
@@ -111,4 +132,36 @@ public class ListenerMulticaster implements ItemStream, StepListener {
 		stream.update(executionContext);
 	}
 
+	public void afterChunk() {
+		chunkListener.afterChunk();
+	}
+
+	public void beforeChunk() {
+		chunkListener.beforeChunk();
+	}
+
+	public void afterRead(Object item) {
+		itemReadListener.afterRead(item);
+	}
+
+	public void beforeRead() {
+		itemReadListener.beforeRead();
+	}
+
+	public void onReadError(Exception ex) {
+		itemReadListener.onReadError(ex);
+	}
+
+	public void afterWrite() {
+		itemWriteListener.afterWrite();
+	}
+
+	public void beforeWrite(Object item) {
+		itemWriteListener.beforeWrite(item);
+	}
+
+	public void onWriteError(Exception ex, Object item) {
+		itemWriteListener.onWriteError(ex, item);
+	}
+
 }