diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java index 7bd6b2fec..a37e3231a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java +++ b/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 + * @param type of object being read */ -public class SynchronizedItemStreamReader implements ItemStream, ItemReader { +public class SynchronizedItemStreamReader implements ItemStreamReader, InitializingBean { - ItemStreamReader itemStreamReader; + private ItemStreamReader delegate; - public void setItemStreamReader(ItemStreamReader itemStreamReader) { - this.itemStreamReader = itemStreamReader; + public void setDelegate(ItemStreamReader delegate) { + this.delegate = delegate; } /** - * This delegates to the read method of the itemStreamReader + * This delegates to the read method of the delegate */ 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"); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java index b10ba64f7..1aee9e64f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java +++ b/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 synchronizedItemStreamReader = new SynchronizedItemStreamReader(); - 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(); } } };