diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java index 7d29319ed..b653c62e5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java @@ -40,14 +40,15 @@ import org.springframework.util.ClassUtils; * Input resources are ordered using {@link #setComparator(Comparator)} to make * sure resource ordering is preserved between job runs in restart scenario. * - * Reset (rollback) capability is implemented by item buffering. - * * * @author Robert Kasanicky + * @author Lucas Ward */ public class MultiResourceItemReader implements ItemReader, ItemStream { private static final Log logger = LogFactory.getLog(MultiResourceItemReader.class); + + private static final String RESOURCE_KEY = "resourceIndex"; private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(); @@ -55,9 +56,9 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { private Resource[] resources; - private MultiResourceIndex index = new MultiResourceIndex(); - private boolean saveState = true; + + private int currentResource = -1; // signals there are no resources to read -> just return null on first read private boolean noInput; @@ -85,11 +86,17 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { if (noInput) { return null; } + + //If there is no resource, then this is the first item, set the current + //resource to 0 and open the first delegate. + if(currentResource == -1){ + currentResource = 0; + delegate.setResource(resources[currentResource]); + delegate.open(new ExecutionContext()); + } T item; item = readNextItem(); - index.incrementItemCount(); - return item; } @@ -105,14 +112,14 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { while (item == null) { - index.incrementResourceCount(); + currentResource++; - if (index.currentResource >= resources.length) { + if (currentResource >= resources.length) { return null; } delegate.close(); - delegate.setResource(resources[index.currentResource]); + delegate.setResource(resources[currentResource]); delegate.open(new ExecutionContext()); item = delegate.read(); @@ -126,7 +133,6 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { * and reset instance variable values. */ public void close() throws ItemStreamException { - index = new MultiResourceIndex(); delegate.close(); noInput = false; } @@ -148,19 +154,13 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { Arrays.sort(resources, comparator); - index.open(executionContext); - - delegate.setResource(resources[index.currentResource]); - - delegate.open(new ExecutionContext()); - - try { - for (int i = 0; i < index.currentItem; i++) { - delegate.read(); - } + if (executionContext.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) { + currentResource = executionContext.getInt(executionContextUserSupport.getKey(RESOURCE_KEY)); + delegate.setResource(resources[currentResource]); + delegate.open(executionContext); } - catch (Exception e) { - throw new ItemStreamException("Could not restore position on restart", e); + else{ + currentResource = -1; } } @@ -169,7 +169,8 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { */ public void update(ExecutionContext executionContext) throws ItemStreamException { if (saveState) { - index.update(executionContext); + executionContext.putInt(executionContextUserSupport.getKey(RESOURCE_KEY), currentResource); + delegate.update(executionContext); } } @@ -207,62 +208,10 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { } public Resource getCurrentResource() { - if (index.currentResource >= resources.length) { + if (currentResource >= resources.length) { return null; } - return resources[index.currentResource]; - } - - /** - * Facilitates keeping track of the position within multi-resource input. - */ - private class MultiResourceIndex { - - private static final String RESOURCE_KEY = "resourceIndex"; - - private static final String ITEM_KEY = "itemIndex"; - - private int currentResource = 0; - - private int markedResource = 0; - - private int currentItem = 0; - - private int markedItem = 0; - - public void incrementItemCount() { - currentItem++; - } - - public void incrementResourceCount() { - currentResource++; - currentItem = 0; - } - - public void mark() { - markedResource = currentResource; - markedItem = currentItem; - } - - public void reset() { - currentResource = markedResource; - currentItem = markedItem; - } - - public void open(ExecutionContext ctx) { - if (ctx.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) { - currentResource = ctx.getInt(executionContextUserSupport.getKey(RESOURCE_KEY)); - } - - if (ctx.containsKey(executionContextUserSupport.getKey(ITEM_KEY))) { - currentItem = ctx.getInt(executionContextUserSupport.getKey(ITEM_KEY)); - } - } - - public void update(ExecutionContext ctx) { - ctx.putInt(executionContextUserSupport.getKey(RESOURCE_KEY), index.currentResource); - ctx.putInt(executionContextUserSupport.getKey(ITEM_KEY), index.currentItem); - } + return resources[currentResource]; } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java index ff6eefebe..7c4857ad6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java @@ -3,13 +3,18 @@ package org.springframework.batch.item.file; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.fail; +import java.io.IOException; +import java.io.InputStream; import java.util.Comparator; import org.junit.Before; import org.junit.Test; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.file.mapping.PassThroughLineMapper; +import org.springframework.core.io.AbstractResource; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; @@ -79,7 +84,6 @@ public class MultiResourceItemReaderIntegrationTests { tested.open(ctx); - assertSame(r1, tested.getCurrentResource()); assertEquals("1", tested.read()); assertSame(r1, tested.getCurrentResource()); assertEquals("2", tested.read()); @@ -264,5 +268,60 @@ public class MultiResourceItemReaderIntegrationTests { tested.close(); } + + @Test + public void testMiddleResourceThrowsException() throws Exception{ + + Resource badResource = new AbstractResource() { + + public InputStream getInputStream() throws IOException { + throw new RuntimeException(); + } + + public String getDescription() {return null;} + }; + + tested.setResources(new Resource[] { r1, badResource, r3, r4, r5 }); + + tested.open(ctx); + + assertEquals("1", tested.read()); + assertEquals("2", tested.read()); + assertEquals("3", tested.read()); + try{ + assertEquals("4", tested.read()); + fail(); + } + catch(ItemStreamException ex){ + //a try/catch was used to ensure the exception was thrown when reading + //the 4th item, rather than on open + } + } + + @Test + public void testFirstResourceThrowsExceptionOnRead() throws Exception{ + + Resource badResource = new AbstractResource() { + + public InputStream getInputStream() throws IOException { + throw new RuntimeException(); + } + + public String getDescription() {return null;} + }; + + tested.setResources(new Resource[] { badResource, r2, r3, r4, r5 }); + + tested.open(ctx); + + try{ + assertEquals("1", tested.read()); + fail(); + } + catch(ItemStreamException ex){ + //a try/catch was used to ensure the exception was thrown when reading + //the 1st item, rather than on open + } + } }