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

RESOLVED - BATCH-663: MultiResourceItemReader doesn't restart correctly after multi-resource rollback

Browse Source 
* added common test for the "rollback + restart" scenario
* MultiResourceItemReader now extends AbstractBufferedItemStreamItemReader
This commit is contained in:
robokaso
2008-06-11 13:04:44 +00:00
parent dcca8b74cb
commit 5b0d65c478
2 changed files with 65 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java
View File

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

41
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java
View File

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