Polishing BATCH-2387

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java

@@ -16,55 +16,59 @@
 package org.springframework.batch.item.support;
 
 import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.item.ItemReader;
-import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamReader;
 import org.springframework.batch.item.NonTransientResourceException;
 import org.springframework.batch.item.ParseException;
 import org.springframework.batch.item.UnexpectedInputException;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
 
 /**
  * 
  * This is a simple ItemStreamReader decorator with a synchronized ItemReader.read() 
  * method - which makes a non-thread-safe ItemReader thread-safe.
  * 
- * However, if reprocessing an item is problematic then using this will make a job not 
- * restartable.  If a restartable job is desired in that case, then further co-ordination 
- * between the read and close methods needs to be implemented.
+ * However, if reprocessing an item is problematic then using this will make a job not
+ * restartable.
  * 
  * Here are some links about the motivation behind this class:
  * - http://projects.spring.io/spring-batch/faq.html#threading-reader}
  * - http://stackoverflow.com/a/20002493/2910265}
  * 
  * @author Matthew Ouyang
- * @since 3.0
+ * @since 3.0.4
  *
- * @param <T>
+ * @param <T> type of object being read
  */
-public class SynchronizedItemStreamReader<T> implements ItemStream, ItemReader<T> {
+public class SynchronizedItemStreamReader<T> implements ItemStreamReader<T>, InitializingBean {
 
-	ItemStreamReader<T> itemStreamReader;
+	private ItemStreamReader<T> delegate;
 
-	public void setItemStreamReader(ItemStreamReader<T> itemStreamReader) {
-		this.itemStreamReader = itemStreamReader;
+	public void setDelegate(ItemStreamReader<T> delegate) {
+		this.delegate = delegate;
 	}
 
 	/**
-	 * This delegates to the read method of the <code>itemStreamReader</code>
+	 * This delegates to the read method of the <code>delegate</code>
 	 */
 	public synchronized T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
-		return this.itemStreamReader.read();
+		return this.delegate.read();
 	}
 
 	public void close() {
-		this.itemStreamReader.close();
+		this.delegate.close();
 	}
 
 	public void open(ExecutionContext executionContext) {
-		this.itemStreamReader.open(executionContext);
+		this.delegate.open(executionContext);
 	}
 
 	public void update(ExecutionContext executionContext) {
-		this.itemStreamReader.update(executionContext);
+		this.delegate.update(executionContext);
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.delegate, "A delegate item reader is required");
 	}
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java

@@ -23,11 +23,11 @@ import java.util.HashSet;
 import java.util.Set;
 
 import org.junit.Test;
+
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStreamReader;
 import org.springframework.batch.item.NonTransientResourceException;
 import org.springframework.batch.item.ParseException;
-import org.springframework.batch.item.UnexpectedInputException;
 
 /**
  * 
@@ -52,8 +52,7 @@ public class SynchronizedItemStreamReaderTests {
 		public static final String HAS_BEEN_OPENED = "hasBeenOpened";
 		public static final String UPDATE_COUNT_KEY = "updateCount";
 
-		public Integer read() throws Exception, UnexpectedInputException, ParseException, 
-			NonTransientResourceException {
+		public Integer read() throws Exception, ParseException, NonTransientResourceException {
 			cursor = cursor + 1;
 			return cursor;
 		}
@@ -92,7 +91,7 @@ public class SynchronizedItemStreamReaderTests {
 
 		final TestItemReader testItemReader = new TestItemReader();
 		final SynchronizedItemStreamReader<Integer> synchronizedItemStreamReader = new SynchronizedItemStreamReader<Integer>();
-		synchronizedItemStreamReader.setItemStreamReader(testItemReader);
+		synchronizedItemStreamReader.setDelegate(testItemReader);
 
 		// Open the ItemReader and make sure it's initialized properly.
 		synchronizedItemStreamReader.open(executionContext);
@@ -111,8 +110,8 @@ public class SynchronizedItemStreamReaderTests {
 					try {
 						ecSet.add(synchronizedItemStreamReader.read());
 						synchronizedItemStreamReader.update(executionContext);
-					} catch (Exception e) {
-					} finally {
+					} catch (Exception ignore) {
+						ignore.printStackTrace();
 					}
 				}
 			};