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

RESOLVED - issue BATCH-1253: Ability to throttle the number of rows returned by a HibernateCursorItemReader

Browse Source
This commit is contained in:
dsyer
2009-07-07 08:45:10 +00:00
parent 456c16962b
commit e516cbd4a0
2 changed files with 140 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java
View File

@@ -34,7 +34,8 @@ import org.springframework.util.Assert;
*
* @author Robert Kasanicky
*/
public abstract class AbstractItemCountingItemStreamItemReader<T> implements ItemReader<T>, ItemStream {
public abstract class AbstractItemCountingItemStreamItemReader<T> implements
ItemReader<T>, ItemStream {
private static final String READ_COUNT = "read.count";
@@ -50,6 +51,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
/**
* Read next item from input.
*
* @return item
* @throws Exception
*/
@@ -76,8 +78,9 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
}
}
public final T read() throws Exception, UnexpectedInputException, ParseException {
if (currentItemCount >= maxItemCount-1) {
public final T read() throws Exception, UnexpectedInputException,
ParseException {
if (currentItemCount >= maxItemCount) {
return null;
}
currentItemCount++;
@@ -88,42 +91,69 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
return currentItemCount;
}
protected void setCurrentItemCount(int count) {
/**
* The index of the item to start reading from. If the
* {@link ExecutionContext} contains a key <code>[name].read.count</code>
* (where <code>[name]</code> is the name of this component) the value from
* the {@link ExecutionContext} will be used in preference.
*
* @see #setName(String)
*
* @param count
* the value of the current item count
*/
public void setCurrentItemCount(int count) {
this.currentItemCount = count;
}
/**
* The maximum index of the items to be read. If the
* {@link ExecutionContext} contains a key <code>[name].read.count.max</code>
* (where <code>[name]</code> is the name of this component) the value from
* the {@link ExecutionContext} will be used in preference.
*
* @see #setName(String)
*
* @param count
* the value of the maximum item count
*/
public void setMaxItemCount(int count) {
this.maxItemCount = count;
}
public void close() throws ItemStreamException {
currentItemCount = 0;
try {
doClose();
}
catch (Exception e) {
} catch (Exception e) {
throw new ItemStreamException("Error while closing item reader", e);
}
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
public void open(ExecutionContext executionContext)
throws ItemStreamException {
try {
doOpen();
}
catch (Exception e) {
} catch (Exception e) {
throw new ItemStreamException("Failed to initialize the reader", e);
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) {
maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX));
maxItemCount = executionContext.getInt(ecSupport
.getKey(READ_COUNT_MAX));
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) {
int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT));
int itemCount = executionContext.getInt(ecSupport
.getKey(READ_COUNT));
if (itemCount < maxItemCount) {
try {
jumpToItem(itemCount);
}
catch (Exception e) {
throw new ItemStreamException("Could not move to stored position on restart", e);
} catch (Exception e) {
throw new ItemStreamException(
"Could not move to stored position on restart", e);
}
}
currentItemCount = itemCount;
@@ -132,17 +162,29 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
public void update(ExecutionContext executionContext)
throws ItemStreamException {
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount);
Assert.notNull(executionContext,
"ExecutionContext must not be null");
executionContext.putInt(ecSupport.getKey(READ_COUNT),
currentItemCount);
if (maxItemCount < Integer.MAX_VALUE) {
executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount);
executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX),
maxItemCount);
}
}
}
/**
* The name of the component which will be used as a stem for keys in the
* {@link ExecutionContext}. Subclasses should provide a default value, e.g.
* the short form of the class name.
*
* @param name
* the name for the component
*/
public void setName(String name) {
ecSupport.setName(name);
}
@@ -152,7 +194,8 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
* {@link ExecutionContext}. Only switch this to false if you don't want to
* save any state from this stream, and you don't need it to be restartable.
*
* @param saveState flag value (default true).
* @param saveState
* flag value (default true).
*/
public void setSaveState(boolean saveState) {
this.saveState = saveState;