From 5b0d65c4787979ed9a642f3f6dd9170f7b575670 Mon Sep 17 00:00:00 2001 From: robokaso Date: Wed, 11 Jun 2008 13:04:44 +0000 Subject: [PATCH] RESOLVED - BATCH-663: MultiResourceItemReader doesn't restart correctly after multi-resource rollback * added common test for the "rollback + restart" scenario * MultiResourceItemReader now extends AbstractBufferedItemStreamItemReader --- .../batch/item/MultiResourceItemReader.java | 157 ++++-------------- .../item/CommonItemStreamItemReaderTests.java | 41 ++++- 2 files changed, 65 insertions(+), 133 deletions(-) diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java index 979acd2c6..408086a8f 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java @@ -1,9 +1,5 @@ package org.springframework.batch.item; -import java.util.ArrayList; -import java.util.List; -import java.util.ListIterator; - import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.util.Assert; @@ -21,130 +17,18 @@ import org.springframework.util.ClassUtils; * * @author Robert Kasanicky */ -public class MultiResourceItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, - InitializingBean { - - private static final String RESOURCE_INDEX = "resourceIndex"; +public class MultiResourceItemReader extends AbstractBufferedItemReaderItemStream implements InitializingBean { private ResourceAwareItemReaderItemStream delegate; private Resource[] resources; - private int currentResourceIndex; - - private List itemBuffer = new ArrayList(); - - private ListIterator itemBufferIterator = null; - - private boolean shouldReadBuffer = false; - - private boolean saveState = false; - - private int lastMarkedBufferIndex = 0; + private int currentResourceIndex = 0; public MultiResourceItemReader() { setName(ClassUtils.getShortName(MultiResourceItemReader.class)); } - /** - * Reads the next item, jumping to next resource if necessary. - */ - public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { - - if (shouldReadBuffer) { - if (itemBufferIterator.hasNext()) { - return itemBufferIterator.next(); - } - else { - // buffer is exhausted, continue reading from file - shouldReadBuffer = false; - itemBufferIterator = null; - } - } - - Object item = delegate.read(); - - while (item == null) { - - if (++currentResourceIndex >= resources.length) { - return null; - } - delegate.close(new ExecutionContext()); - delegate.setResource(resources[currentResourceIndex]); - delegate.open(new ExecutionContext()); - item = delegate.read(); - - } - - itemBuffer.add(item); - - return item; - } - - /** - * Clears the item buffer and cancels reading from buffer if it applies. - * - * @see ItemReader#mark() - */ - public void mark() throws MarkFailedException { - if (!shouldReadBuffer) { - itemBuffer.clear(); - itemBufferIterator = null; - lastMarkedBufferIndex = 0; - } - else { - lastMarkedBufferIndex = itemBufferIterator.nextIndex(); - } - delegate.mark(); - } - - /** - * Switches to 'read from buffer' state. - * - * @see ItemReader#reset() - */ - public void reset() throws ResetFailedException { - shouldReadBuffer = true; - itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex); - } - - /** - * Close the {@link #setDelegate(ResourceAwareItemReaderItemStream)} reader - * and reset instance variable values. - */ - public void close(ExecutionContext executionContext) throws ItemStreamException { - shouldReadBuffer = false; - itemBufferIterator = null; - itemBuffer.clear(); - delegate.close(executionContext); - } - - /** - * Figure out which resource to start with in case of restart and open the - * delegate. - */ - public void open(ExecutionContext executionContext) throws ItemStreamException { - - if (executionContext.containsKey(getKey(RESOURCE_INDEX))) { - int index = Long.valueOf(executionContext.getLong(getKey(RESOURCE_INDEX))).intValue(); - currentResourceIndex = index; - } - - delegate.setResource(resources[currentResourceIndex]); - - delegate.open(executionContext); - } - - /** - * Store the current resource index and delegate's data. - */ - public void update(ExecutionContext executionContext) throws ItemStreamException { - if (saveState) { - executionContext.putLong(getKey(RESOURCE_INDEX), currentResourceIndex); - delegate.update(executionContext); - } - } - /** * @param delegate reads items from single {@link Resource}. */ @@ -162,16 +46,33 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impleme public void setResources(Resource[] resources) { this.resources = resources; } - - /** - * Set the boolean indicating whether or not state should be saved in the - * provided {@link ExecutionContext} during the {@link ItemStream} call to - * update. - * - * @param saveState - */ - public void setSaveState(boolean saveState) { - this.saveState = saveState; + + protected void doClose() throws Exception { + currentResourceIndex = 0; + delegate.close(new ExecutionContext()); + } + + protected void doOpen() throws Exception { + delegate.setResource(resources[0]); + delegate.open(new ExecutionContext()); + } + + protected Object doRead() throws Exception { + + Object item = delegate.read(); + + while (item == null) { + + if (++currentResourceIndex >= resources.length) { + return null; + } + delegate.close(new ExecutionContext()); + delegate.setResource(resources[currentResourceIndex]); + delegate.open(new ExecutionContext()); + item = delegate.read(); + } + + return item; } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java index c4c7a6c6b..2b21fe131 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java @@ -28,8 +28,6 @@ public abstract class CommonItemStreamItemReaderTests extends CommonItemReaderTe testedAsStream().close(executionContext); } - - /** * Restart scenario - read items, update execution context, create new * reader and restore from restart data - the new input source should @@ -38,7 +36,7 @@ public abstract class CommonItemStreamItemReaderTests extends CommonItemReaderTe public void testRestart() throws Exception { testedAsStream().update(executionContext); - + Foo foo1 = (Foo) tested.read(); assertEquals(1, foo1.getValue()); @@ -55,10 +53,43 @@ public abstract class CommonItemStreamItemReaderTests extends CommonItemReaderTe Foo fooAfterRestart = (Foo) tested.read(); assertEquals(3, fooAfterRestart.getValue()); } - + + /** + * Restart scenario - read items, rollback to last marked position, update + * execution context, create new reader and restore from restart data - the + * new input source should continue where the old one finished. + */ + public void testResetAndRestart() throws Exception { + + testedAsStream().update(executionContext); + + Foo foo1 = (Foo) tested.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) tested.read(); + assertEquals(2, foo2.getValue()); + + tested.mark(); + + Foo foo3 = (Foo) tested.read(); + assertEquals(3, foo3.getValue()); + + tested.reset(); + + testedAsStream().update(executionContext); + + // create new input source + tested = getItemReader(); + + testedAsStream().open(executionContext); + + Foo fooAfterRestart = (Foo) tested.read(); + assertEquals(3, fooAfterRestart.getValue()); + } + public void testReopen() throws Exception { testedAsStream().update(executionContext); - + Foo foo1 = (Foo) tested.read(); assertEquals(1, foo1.getValue());