Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - BATCH-520: Add a Delegating File Reader For Multiple Files

Browse Source 
added MultiResourceReader (can be used both with FlatFileItemReader or StaxEventItemReader)
This commit is contained in:
robokaso
2008-05-21 15:30:02 +00:00
parent c914ce74fc
commit 6419588259
5 changed files with 360 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java Normal file
View File

@@ -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");
}
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ResourceAwareItemReaderItemStream.java Normal file
View File

@@ -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);
}

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
View File

@@ -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);

27
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java
View File

@@ -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.
*/

156
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/MultiResourceItemReaderIntegrationTests.java Normal file
View File

@@ -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);
}
}