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

OPEN - issue BATCH-789: Remove mark/reset from ItemReader

Browse Source 
Removed all code from mark() and reset() implementations
This commit is contained in:
dsyer
2008-08-21 11:03:44 +00:00
parent 04db13bbc4
commit e1d1413cc3
24 changed files with 92 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java
View File

@@ -69,14 +69,12 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
private boolean initialized = false;
private List<T> keys;
private T currentKey = null;
private Iterator<T> keysIterator;
private int currentIndex = 0;
private int lastCommitIndex = 0;
private KeyCollector<T> keyCollector;
private boolean saveState = false;
@@ -91,7 +89,6 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
* @param keys
*/
public DrivingQueryItemReader(List<T> keys) {
this.keys = keys;
this.keysIterator = keys.iterator();
}
@@ -105,7 +102,8 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
if (keysIterator.hasNext()) {
currentIndex++;
return keysIterator.next();
currentKey = keysIterator.next();
return currentKey;
}
return null;
@@ -119,11 +117,7 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
* @return the current key.
*/
protected T getCurrentKey() {
if (initialized && currentIndex > 0) {
return keys.get(currentIndex - 1);
}
return null;
return currentKey;
}
/**
@@ -133,8 +127,6 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
public void close(ExecutionContext executionContext) {
initialized = false;
currentIndex = 0;
lastCommitIndex = 0;
keys = null;
keysIterator = null;
}
@@ -147,9 +139,9 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
*/
public void open(ExecutionContext executionContext) {
Assert.state(keys == null && !initialized, "Cannot open an already opened item reader"
Assert.state(keysIterator == null && !initialized, "Cannot open an already opened item reader"
+ ", call close() first.");
keys = keyCollector.retrieveKeys(executionContext);
List<T> keys = keyCollector.retrieveKeys(executionContext);
Assert.notNull(keys, "Keys must not be null");
keysIterator = keys.listIterator();
initialized = true;
@@ -177,26 +169,10 @@ public class DrivingQueryItemReader<T> implements ItemReader<T>, InitializingBea
this.keyCollector = keyCollector;
}
/**
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*/
public void mark() {
lastCommitIndex = currentIndex;
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.io.support.AbstractTransactionalIoSource#reset
* (org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
keysIterator = keys.listIterator(lastCommitIndex);
currentIndex = lastCommitIndex;
}
public void setSaveState(boolean saveState) {

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java
View File

@@ -23,7 +23,7 @@ import org.hibernate.SessionFactory;
import org.hibernate.StatelessSession;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.support.AbstractItemReaderItemStream;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -51,7 +51,7 @@ import org.springframework.util.ClassUtils;
* @author Robert Kasanicky
* @author Dave Syer
*/
public class HibernateCursorItemReader<T> extends AbstractBufferedItemReaderItemStream<T> implements ItemStream,
public class HibernateCursorItemReader<T> extends AbstractItemReaderItemStream<T> implements ItemStream,
InitializingBean {
private SessionFactory sessionFactory;

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
View File

@@ -29,7 +29,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.support.AbstractItemReaderItemStream;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.InvalidDataAccessResourceUsageException;
import org.springframework.jdbc.SQLWarningException;
@@ -97,7 +97,7 @@ import org.springframework.util.ClassUtils;
* @author Peter Zozom
* @author Robert Kasanicky
*/
public class JdbcCursorItemReader<T> extends AbstractBufferedItemReaderItemStream<T> implements InitializingBean {
public class JdbcCursorItemReader<T> extends AbstractItemReaderItemStream<T> implements InitializingBean {
private static Log log = LogFactory.getLog(JdbcCursorItemReader.class);

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java
View File

@@ -27,7 +27,7 @@ import javax.persistence.Query;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.support.AbstractItemReaderItemStream;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.util.Assert;
@@ -58,7 +58,7 @@ import org.springframework.util.ClassUtils;
*
* @author Thomas Risberg
*/
public class JpaPagingItemReader<T> extends AbstractBufferedItemReaderItemStream<T> implements InitializingBean {
public class JpaPagingItemReader<T> extends AbstractItemReaderItemStream<T> implements InitializingBean {
protected Log logger = LogFactory.getLog(getClass());

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

@@ -33,7 +33,7 @@ import org.springframework.batch.item.file.separator.RecordSeparatorPolicy;
import org.springframework.batch.item.file.transform.AbstractLineTokenizer;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.LineTokenizer;
import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.support.AbstractItemReaderItemStream;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
@@ -62,7 +62,7 @@ import org.springframework.util.ClassUtils;
* @author Robert Kasanicky
* @author Dave Syer
*/
public class FlatFileItemReader<T> extends AbstractBufferedItemReaderItemStream<T> implements
public class FlatFileItemReader<T> extends AbstractItemReaderItemStream<T> implements
ResourceAwareItemReaderItemStream<T>, InitializingBean {
private static Log log = LogFactory.getLog(FlatFileItemReader.class);

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

@@ -1,10 +1,7 @@
package org.springframework.batch.item.file;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.ListIterator;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
@@ -33,12 +30,9 @@ import org.springframework.util.ClassUtils;
*
* @author Robert Kasanicky
*/
public class MultiResourceItemReader<T> extends ExecutionContextUserSupport implements ItemReader<T>, ItemStream {
public class MultiResourceItemReader<T> implements ItemReader<T>, ItemStream {
/**
* Unique object instance that marks resource boundaries in the item buffer
*/
private static final Object END_OF_RESOURCE_MARKER = new Object();
private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport();
private ResourceAwareItemReaderItemStream<T> delegate;
@@ -46,13 +40,7 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
private MultiResourceIndex index = new MultiResourceIndex();
private List<Object> itemBuffer = new ArrayList<Object>();
private ListIterator<Object> itemBufferIterator = null;
private boolean shouldReadBuffer = false;
private boolean saveState = false;
private boolean saveState = true;
private Comparator<Resource> comparator = new Comparator<Resource>() {
@@ -66,7 +54,7 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
};
public MultiResourceItemReader() {
setName(ClassUtils.getShortName(MultiResourceItemReader.class));
executionContextUserSupport.setName(ClassUtils.getShortName(MultiResourceItemReader.class));
}
/**
@@ -75,13 +63,7 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
public T read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
T item;
if (shouldReadBuffer) {
item = readBufferedItem();
}
else {
item = readNextItem();
}
item = readNextItem();
index.incrementItemCount();
return item;
@@ -103,7 +85,6 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
if (index.currentResource >= resources.length) {
return null;
}
itemBuffer.add(END_OF_RESOURCE_MARKER);
delegate.close(new ExecutionContext());
delegate.setResource(resources[index.currentResource]);
@@ -112,68 +93,20 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
item = delegate.read();
}
itemBuffer.add(item);
return item;
}
/**
* Read next item from buffer while keeping track of the position within the
* input for possible restart.
* @return next item from buffer
*/
@SuppressWarnings("unchecked")
private T readBufferedItem() {
Object buffered = itemBufferIterator.next();
while (buffered == END_OF_RESOURCE_MARKER) {
index.incrementResourceCount();
buffered = itemBufferIterator.next();
}
if (!itemBufferIterator.hasNext()) {
// buffer is exhausted, continue reading from file
shouldReadBuffer = false;
itemBufferIterator = null;
}
return (T) buffered;
}
/**
* Remove the longer needed items from buffer, mark the index position and
* call mark() on delegate so that it clears its buffers.
*/
public void mark() throws MarkFailedException {
emptyBuffer();
index.mark();
delegate.mark();
}
/**
* Discard the buffered items that have already been read.
*/
private void emptyBuffer() {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
}
else {
itemBuffer = itemBuffer.subList(itemBufferIterator.nextIndex(), itemBuffer.size());
itemBufferIterator = itemBuffer.listIterator();
}
}
/**
* Switches to 'read from buffer' state.
*/
public void reset() throws ResetFailedException {
if (!itemBuffer.isEmpty()) {
shouldReadBuffer = true;
itemBufferIterator = itemBuffer.listIterator();
}
index.reset();
}
/**
@@ -181,10 +114,7 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
* and reset instance variable values.
*/
public void close(ExecutionContext executionContext) throws ItemStreamException {
shouldReadBuffer = false;
itemBufferIterator = null;
index = new MultiResourceIndex();
itemBuffer.clear();
delegate.close(new ExecutionContext());
}
@@ -207,7 +137,6 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
try {
for (int i = 0; i < index.currentItem; i++) {
delegate.read();
delegate.mark();
}
}
catch (Exception e) {
@@ -294,18 +223,18 @@ public class MultiResourceItemReader<T> extends ExecutionContextUserSupport impl
}
public void open(ExecutionContext ctx) {
if (ctx.containsKey(getKey(RESOURCE_KEY))) {
currentResource = Long.valueOf(ctx.getLong(getKey(RESOURCE_KEY))).intValue();
if (ctx.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) {
currentResource = Long.valueOf(ctx.getLong(executionContextUserSupport.getKey(RESOURCE_KEY))).intValue();
}
if (ctx.containsKey(getKey(ITEM_KEY))) {
currentItem = ctx.getLong(getKey(ITEM_KEY));
if (ctx.containsKey(executionContextUserSupport.getKey(ITEM_KEY))) {
currentItem = ctx.getLong(executionContextUserSupport.getKey(ITEM_KEY));
}
}
public void update(ExecutionContext ctx) {
ctx.putLong(getKey(RESOURCE_KEY), index.currentResource);
ctx.putLong(getKey(ITEM_KEY), index.currentItem);
ctx.putLong(executionContextUserSupport.getKey(RESOURCE_KEY), index.currentResource);
ctx.putLong(executionContextUserSupport.getKey(ITEM_KEY), index.currentItem);
}
}

63
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractBufferedItemReaderItemStream.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemReaderItemStream.java
View File

@@ -1,17 +1,11 @@
package org.springframework.batch.item.support;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.batch.item.NoWorkFoundException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.ResetFailedException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.util.Assert;
@@ -26,22 +20,12 @@ import org.springframework.util.Assert;
*
* @author Robert Kasanicky
*/
public abstract class AbstractBufferedItemReaderItemStream<T> implements ItemReader<T>, ItemStream {
public abstract class AbstractItemReaderItemStream<T> implements ItemReader<T>, ItemStream {
private static final String READ_COUNT = "read.count";
private int currentItemCount = 0;
private int lastMarkedItemCount = 0;
private boolean shouldReadBuffer = false;
private List<T> itemBuffer = new ArrayList<T>();
private ListIterator<T> itemBufferIterator = null;
private int lastMarkedBufferIndex = 0;
private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport();
private boolean saveState = true;
@@ -75,24 +59,8 @@ public abstract class AbstractBufferedItemReaderItemStream<T> implements ItemRea
}
public T read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
currentItemCount++;
if (shouldReadBuffer) {
if (itemBufferIterator.hasNext()) {
return itemBufferIterator.next();
}
else {
// buffer is exhausted, continue reading from file
shouldReadBuffer = false;
itemBufferIterator = null;
}
}
T item = doRead();
itemBuffer.add(item);
return item;
return doRead();
}
/**
@@ -100,28 +68,11 @@ public abstract class AbstractBufferedItemReaderItemStream<T> implements ItemRea
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.support.AbstractItemReader#mark()
*/
public void mark() throws MarkFailedException {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
lastMarkedBufferIndex = 0;
}
else {
lastMarkedBufferIndex = itemBufferIterator.nextIndex();
}
lastMarkedItemCount = currentItemCount;
public void mark() {
}
public void reset() throws ResetFailedException {
currentItemCount = lastMarkedItemCount;
shouldReadBuffer = true;
itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
public void reset() {
}
protected int getCurrentItemCount() {
@@ -134,12 +85,6 @@ public abstract class AbstractBufferedItemReaderItemStream<T> implements ItemRea
public void close(ExecutionContext executionContext) throws ItemStreamException {
currentItemCount = 0;
lastMarkedItemCount = 0;
lastMarkedBufferIndex = 0;
itemBufferIterator = null;
shouldReadBuffer = false;
itemBuffer.clear();
try {
doClose();
}

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java
View File

@@ -97,11 +97,9 @@ public class AggregateItemReader<T> implements ItemReader<List<T>> {
}
public void mark() throws MarkFailedException {
itemReader.mark();
}
public void reset() throws ResetFailedException {
itemReader.reset();
}
public void setItemReader(ItemReader<AggregateItem<T>> itemReader) {

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java
View File

@@ -74,7 +74,6 @@ public class DelegatingItemReader<T> extends AbstractItemReader<T> implements In
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
itemReader.mark();
}
/*
@@ -82,6 +81,5 @@ public class DelegatingItemReader<T> extends AbstractItemReader<T> implements In
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
itemReader.reset();
}
}

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

@@ -9,7 +9,7 @@ import javax.xml.stream.XMLStreamException;
import javax.xml.stream.events.StartElement;
import org.springframework.batch.item.file.ResourceAwareItemReaderItemStream;
import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.support.AbstractItemReaderItemStream;
import org.springframework.batch.item.xml.stax.DefaultFragmentEventReader;
import org.springframework.batch.item.xml.stax.FragmentEventReader;
import org.springframework.beans.factory.InitializingBean;
@@ -30,7 +30,7 @@ import org.springframework.util.ClassUtils;
*
* @author Robert Kasanicky
*/
public class StaxEventItemReader<T> extends AbstractBufferedItemReaderItemStream<T> implements
public class StaxEventItemReader<T> extends AbstractItemReaderItemStream<T> implements
ResourceAwareItemReaderItemStream<T>, InitializingBean {
private FragmentEventReader fragmentReader;

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java
View File

@@ -18,7 +18,6 @@ package org.springframework.batch.retry.policy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
import org.springframework.batch.retry.ExhaustedRetryException;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;