From 64195882597d961f41314f8b2d5ddc237ebf394d Mon Sep 17 00:00:00 2001 From: robokaso Date: Wed, 21 May 2008 15:30:02 +0000 Subject: [PATCH] IN PROGRESS - BATCH-520: Add a Delegating File Reader For Multiple Files added MultiResourceReader (can be used both with FlatFileItemReader or StaxEventItemReader) --- .../batch/item/MultiResourceItemReader.java | 174 ++++++++++++++++++ .../ResourceAwareItemReaderItemStream.java | 14 ++ .../batch/item/file/FlatFileItemReader.java | 3 +- .../batch/item/xml/StaxEventItemReader.java | 27 +-- ...ltiResourceItemReaderIntegrationTests.java | 156 ++++++++++++++++ 5 files changed, 360 insertions(+), 14 deletions(-) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ResourceAwareItemReaderItemStream.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/MultiResourceItemReaderIntegrationTests.java 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 new file mode 100644 index 000000000..0aef44662 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java @@ -0,0 +1,174 @@ +package org.springframework.batch.item; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.io.Resource; +import org.springframework.core.io.support.PathMatchingResourcePatternResolver; +import org.springframework.core.io.support.ResourcePatternResolver; +import org.springframework.util.Assert; + +/** + * Reads items from multiple resources sequentially - resource list is given by + * {@link #setResourcePatternResolver(ResourcePatternResolver)}, the actual + * reading is delegated to + * {@link #setDelegate(ResourceAwareItemReaderItemStream)}. + * + * Reset (rollback) capability is implemented by item buffering. + * + * @author Robert Kasanicky + */ +public class MultiResourceItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, + InitializingBean { + + private static final String RESOURCE_INDEX = "resourceIndex"; + + private ResourceAwareItemReaderItemStream delegate; + + private ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver(); + + private String resourceLocationPattern; + + private Resource[] resources; + + private int currentResourceIndex; + + private List itemBuffer = new ArrayList(); + + private Iterator itemBufferIterator = null; + + private boolean shouldReadBuffer = false; + + public MultiResourceItemReader() { + setName(MultiResourceItemReader.class.getSimpleName()); + } + + /** + * 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 { + delegate.mark(); + itemBuffer.clear(); + shouldReadBuffer = false; + } + + /** + * Switches to 'read from buffer' state. + * + * @see ItemReader#reset() + */ + public void reset() throws ResetFailedException { + shouldReadBuffer = true; + itemBufferIterator = itemBuffer.listIterator(); + } + + /** + * 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 { + try { + resources = resourcePatternResolver.getResources(resourceLocationPattern); + } + catch (Exception e) { + throw new ItemStreamException("Couldn't get resource", e); + } + + 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 { + executionContext.putLong(getKey(RESOURCE_INDEX), currentResourceIndex); + delegate.update(executionContext); + } + + /** + * @param delegate reads items from single {@link Resource}. + */ + public void setDelegate(ResourceAwareItemReaderItemStream delegate) { + this.delegate = delegate; + } + + /** + * @param resourcePatternResolver provides the list of input + * {@link Resource}s given {@link #setResourceLocationPattern(String)}. + * {@link PathMatchingResourcePatternResolver} is used by default. + */ + public void setResourcePatternResolver(ResourcePatternResolver resourcePatternResolver) { + this.resourcePatternResolver = resourcePatternResolver; + } + + /** + * @param resourceLocationPattern identifies the input {@link Resource}s, + * parsed by {@link #setResourcePatternResolver(ResourcePatternResolver)} + */ + public void setResourceLocationPattern(String resourceLocationPattern) { + this.resourceLocationPattern = resourceLocationPattern; + } + + public void afterPropertiesSet() throws Exception { + Assert.notNull(resourcePatternResolver, "resourcePatternResolver property must be set"); + Assert.hasLength(resourceLocationPattern, "resourceLocationPattern property must be set"); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ResourceAwareItemReaderItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ResourceAwareItemReaderItemStream.java new file mode 100644 index 000000000..a18211b44 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ResourceAwareItemReaderItemStream.java @@ -0,0 +1,14 @@ +package org.springframework.batch.item; + +import org.springframework.core.io.Resource; + +/** + * Interface for {@link ItemReader}s that implement {@link ItemStream} and read + * input from {@link Resource}. + * + * @author Robert Kasanicky + */ +public interface ResourceAwareItemReaderItemStream extends ItemReader, ItemStream { + + void setResource(Resource resource); +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java index a9d18fe00..0ce26bd8c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java @@ -25,6 +25,7 @@ import org.springframework.batch.item.ItemReaderException; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ReaderNotOpenException; +import org.springframework.batch.item.ResourceAwareItemReaderItemStream; import org.springframework.batch.item.file.mapping.FieldSet; import org.springframework.batch.item.file.mapping.FieldSetMapper; import org.springframework.batch.item.file.separator.LineReader; @@ -61,7 +62,7 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, InitializingBean { +public class FlatFileItemReader extends ExecutionContextUserSupport implements ResourceAwareItemReaderItemStream, InitializingBean { private static Log log = LogFactory.getLog(FlatFileItemReader.class); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java index 16e519d9f..c9279cb68 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java @@ -14,10 +14,10 @@ import javax.xml.stream.events.StartElement; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ExecutionContextUserSupport; -import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ReaderNotOpenException; +import org.springframework.batch.item.ResourceAwareItemReaderItemStream; import org.springframework.batch.item.xml.stax.DefaultFragmentEventReader; import org.springframework.batch.item.xml.stax.FragmentEventReader; import org.springframework.beans.factory.InitializingBean; @@ -36,7 +36,7 @@ import org.springframework.util.ClassUtils; * * @author Robert Kasanicky */ -public class StaxEventItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, +public class StaxEventItemReader extends ExecutionContextUserSupport implements ResourceAwareItemReaderItemStream, InitializingBean { private static final String READ_COUNT_STATISTICS_NAME = "read.count"; @@ -60,13 +60,14 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements private long currentRecordCount = 0; private boolean saveState = false; - + private List buffer = new ArrayList(); - + private Iterator bufferIterator = null; - + /** - * indicates the reader has been shouldReadBuffer and should read items from buffer + * indicates the reader has been shouldReadBuffer and should read items from + * buffer */ private boolean shouldReadBuffer = false; @@ -86,18 +87,19 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements } currentRecordCount++; - + // read from buffer after rollback if (shouldReadBuffer) { if (bufferIterator.hasNext()) { return bufferIterator.next(); - } else { + } + else { // buffer is exhausted, continue reading from file shouldReadBuffer = false; bufferIterator = null; } } - + Object item = null; if (moveCursorToNextFragment(fragmentReader)) { @@ -142,8 +144,7 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements try { inputStream = resource.getInputStream(); - eventReader = XMLInputFactory.newInstance().createXMLEventReader( - inputStream); + eventReader = XMLInputFactory.newInstance().createXMLEventReader(inputStream); fragmentReader = new DefaultFragmentEventReader(eventReader); } catch (XMLStreamException xse) { @@ -159,7 +160,7 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements int REASONABLE_ADHOC_COMMIT_FREQUENCY = 100; while (currentRecordCount <= restoredRecordCount) { currentRecordCount++; - + if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) { mark(); // clear the history buffer } @@ -288,7 +289,7 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements public void setSaveState(boolean saveState) { this.saveState = saveState; } - + /** * Clear the buffer and release the iterator. */ diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/MultiResourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/MultiResourceItemReaderIntegrationTests.java new file mode 100644 index 000000000..82349030c --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/MultiResourceItemReaderIntegrationTests.java @@ -0,0 +1,156 @@ +package org.springframework.batch.item; + +import junit.framework.TestCase; + +import org.easymock.MockControl; +import org.springframework.batch.item.file.FlatFileItemReader; +import org.springframework.batch.item.file.mapping.FieldSet; +import org.springframework.batch.item.file.mapping.PassThroughFieldSetMapper; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.core.io.support.ResourcePatternResolver; + +/** + * Tests for {@link MultiResourceItemReader}. + */ +public class MultiResourceItemReaderIntegrationTests extends TestCase { + + private static final String PATTERN = "resource location pattern"; + + private MultiResourceItemReader tested = new MultiResourceItemReader(); + + private FlatFileItemReader itemReader = new FlatFileItemReader(); + + private ExecutionContext ctx = new ExecutionContext(); + + // test input spans several resources + private Resource r1 = new ByteArrayResource("1\n2\n3\n".getBytes()); + + private Resource r2 = new ByteArrayResource("4\n5\n".getBytes()); + + private Resource r3 = new ByteArrayResource("".getBytes()); + + private Resource r4 = new ByteArrayResource("6\n".getBytes()); + + private Resource r5 = new ByteArrayResource("7\n8\n".getBytes()); + + /** + * Setup the tested reader to read from the test resources. + */ + protected void setUp() throws Exception { + + MockControl control = MockControl.createStrictControl(ResourcePatternResolver.class); + ResourcePatternResolver resolver = (ResourcePatternResolver) control.getMock(); + resolver.getResources(PATTERN); + control.setReturnValue(new Resource[] { r1, r2, r3, r4, r5 }, 2); + control.replay(); + + itemReader.setFieldSetMapper(new PassThroughFieldSetMapper()); + + tested.setResourcePatternResolver(resolver); + tested.setDelegate(itemReader); + tested.setResourceLocationPattern(PATTERN); + tested.afterPropertiesSet(); + } + + /** + * Read input from start to end. + */ + public void testRead() throws Exception { + + tested.open(ctx); + + assertEquals("1", readItem()); + assertEquals("2", readItem()); + assertEquals("3", readItem()); + assertEquals("4", readItem()); + assertEquals("5", readItem()); + assertEquals("6", readItem()); + assertEquals("7", readItem()); + assertEquals("8", readItem()); + assertEquals(null, readItem()); + + tested.close(ctx); + } + + /** + * Read items with a couple of rollbacks, requiring to jump back to items + * from previous resources. + */ + public void testReset() throws Exception { + + tested.open(ctx); + + assertEquals("1", readItem()); + + tested.mark(); + + assertEquals("2", readItem()); + assertEquals("3", readItem()); + + tested.reset(); + + assertEquals("2", readItem()); + assertEquals("3", readItem()); + assertEquals("4", readItem()); + + tested.reset(); + + assertEquals("2", readItem()); + assertEquals("3", readItem()); + assertEquals("4", readItem()); + assertEquals("5", readItem()); + + tested.mark(); + + assertEquals("6", readItem()); + assertEquals("7", readItem()); + + tested.reset(); + + assertEquals("6", readItem()); + assertEquals("7", readItem()); + + assertEquals("8", readItem()); + assertEquals(null, readItem()); + + tested.close(ctx); + } + + /** + * Restore from saved state. + */ + public void testRestart() throws Exception { + + itemReader.setSaveState(true); + + tested.open(ctx); + + assertEquals("1", readItem()); + assertEquals("2", readItem()); + assertEquals("3", readItem()); + assertEquals("4", readItem()); + + tested.update(ctx); + + assertEquals("5", readItem()); + assertEquals("6", readItem()); + + tested.close(ctx); + + tested.open(ctx); + + assertEquals("5", readItem()); + assertEquals("6", readItem()); + assertEquals("7", readItem()); + assertEquals("8", readItem()); + assertEquals(null, readItem()); + } + + private String readItem() throws Exception { + Object result = tested.read(); + return result == null ? null : ((FieldSet) result).readString(0); + + } + +}