BATCH-365: ItemStream#Update and ItemStream#close now accept an ExecutionContext as an argument. I removed the local ExecutionContexts in the readers and readers, and had them use the passed in ones instead.

This commit is contained in:
lucasward
2008-02-26 21:22:10 +00:00
parent 93445b9164
commit e952a55ab2
28 changed files with 172 additions and 163 deletions

View File

@@ -81,9 +81,9 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
private boolean initialized = false;
private ExecutionContext executionContext = new ExecutionContext();
private String name = HibernateCursorItemReader.class.getName();
private boolean saveState = false;
public Object read() {
if (!initialized) {
@@ -109,7 +109,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
* Closes the result set cursor and hibernate session.
*/
public void close() {
public void close(ExecutionContext executionContext) {
initialized = false;
cursor.close();
currentProcessedRow = 0;
@@ -127,8 +127,6 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
* Creates cursor for the query.
*/
public void open(ExecutionContext executionContext) {
this.executionContext = executionContext;
Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
if (useStatelessSession) {
@@ -187,10 +185,13 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
*/
public void update() {
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
String skipped = skippedRows.toString();
executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
public void update(ExecutionContext executionContext) {
if(saveState){
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
String skipped = skippedRows.toString();
executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
}
}
/**
@@ -238,4 +239,8 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
private String getKey(String key){
return name + "." + key;
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
}

View File

@@ -155,8 +155,6 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
private boolean initialized = false;
private ExecutionContext executionContext = new ExecutionContext();
private boolean saveState = false;
private String name = JdbcCursorItemReader.class.getName();
@@ -269,9 +267,9 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
* will be closed. This must be called or the connection and cursor will be
* held open indefinitely!
*
* @see org.springframework.batch.item.ResourceLifecycle#close()
* @see org.springframework.batch.item.ResourceLifecycle#close(ExecutionContext)
*/
public void close() {
public void close(ExecutionContext executionContext) {
initialized = false;
JdbcUtils.closeResultSet(this.rs);
JdbcUtils.closeStatement(this.stmt);
@@ -312,7 +310,7 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
handleWarnings(this.stmt.getWarnings());
}
catch (SQLException se) {
close();
close(null);
throw getExceptionTranslator().translate("Executing query", sql, se);
}
@@ -385,8 +383,9 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#getExecutionContext()
*/
public void update() {
public void update(ExecutionContext executionContext) {
if(saveState){
Assert.notNull(executionContext, "ExecutionContext must not be null");
String skipped = skippedRows.toString();
executionContext.putString(addName(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
executionContext.putLong(addName(CURRENT_PROCESSED_ROW), currentProcessedRow);
@@ -404,7 +403,6 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
Assert.notNull(context, "ExecutionContext must not be null");
executeQuery();
initialized = true;
this.executionContext = context;
// Properties restartProperties = data.getProperties();
if (!context.containsKey(addName(CURRENT_PROCESSED_ROW))) {

View File

@@ -61,8 +61,6 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
private KeyGenerator keyGenerator;
private ExecutionContext executionContext = new ExecutionContext();
private boolean saveState = false;
public DrivingQueryItemReader() {
@@ -119,7 +117,7 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
* Close the resource by setting the list of keys to null, allowing them to
* be garbage collected.
*/
public void close() {
public void close(ExecutionContext executionContext) {
initialized = false;
currentIndex = 0;
lastCommitIndex = 0;
@@ -143,11 +141,11 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
keys = keyGenerator.retrieveKeys(executionContext);
keysIterator = keys.listIterator();
initialized = true;
this.executionContext = executionContext;
}
public void update() {
public void update(ExecutionContext executionContext) {
if(saveState){
Assert.notNull(executionContext, "ExecutionContext must not be null");
if(getCurrentKey() != null){
keyGenerator.saveState(getCurrentKey(), executionContext);
}

View File

@@ -93,9 +93,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
private LineTokenizer tokenizer = new DelimitedLineTokenizer();
private FieldSetMapper fieldSetMapper;
private ExecutionContext executionContext = new ExecutionContext();
private String name = FlatFileItemReader.class.getName();
/**
@@ -112,8 +110,6 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
Assert.state(resource.exists(), "Resource must exist: [" + resource + "]");
this.executionContext = executionContext;
log.debug("Opening flat file for reading: " + resource);
if (this.reader == null) {
@@ -161,11 +157,11 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* Close and null out the reader.
* @throws Exception
*/
public void close() throws StreamException {
public void close(ExecutionContext executionContext) throws StreamException {
try {
if (reader != null) {
log.debug("Closing flat file for reading: " + resource);
reader.close();
reader.close(null);
}
}
finally {
@@ -202,10 +198,11 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* the current Line Count which can be used to reinitialise the batch job in
* case of restart.
*/
public void update() {
public void update(ExecutionContext executionContext) {
if (reader == null) {
throw new StreamException("ItemStream not open or already closed.");
}
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(getKey(READ_STATISTICS_NAME), reader.getPosition());
executionContext.putLong(getKey(SKIPPED_STATISTICS_NAME), skippedLines.size());
}

View File

@@ -79,8 +79,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
private Resource resource;
private ExecutionContext executionContext = new ExecutionContext();
private OutputState state = null;
private LineAggregator lineAggregator = new DelimitedLineAggregator();
@@ -149,9 +147,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
* @see ResourceLifecycle#close()
* @see ResourceLifecycle#close(ExecutionContext)
*/
public void close() {
public void close(ExecutionContext executionContext) {
if (state != null) {
getOutputState().close();
state = null;
@@ -184,7 +182,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
* @see ResourceLifecycle#open()
*/
public void open(ExecutionContext executionContext) {
this.executionContext = executionContext;
OutputState outputState = getOutputState();
if(executionContext.containsKey(getKey(RESTART_DATA_NAME))){
outputState.restoreFrom(executionContext);
@@ -192,12 +189,13 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
* @see ItemStream#update()
* @see ItemStream#update(ExecutionContext)
*/
public void update() {
public void update(ExecutionContext executionContext) {
if (state == null) {
throw new StreamException("ItemStream not open or already closed.");
}
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(getKey(RESTART_DATA_NAME), state.position());
executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
executionContext.putLong(getKey(RESTART_COUNT_STATISTICS_NAME), state.restartCount);

View File

@@ -26,6 +26,7 @@ import java.util.HashSet;
import java.util.Iterator;
import org.springframework.batch.io.exception.BatchEnvironmentException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.exception.MarkFailedException;
@@ -169,11 +170,11 @@ public class ResourceLineReader extends ItemStreamSupport implements LineReader,
/**
* Close the reader associated with this input source.
*
* @see org.springframework.batch.io.ItemReader#close()
* @see org.springframework.batch.io.ItemReader#close(ExecutionContext)
* @throws BatchEnvironmentException if there is an {@link IOException}
* during the close operation.
*/
public synchronized void close() {
public synchronized void close(ExecutionContext executionContext) {
if (state == null) {
return;
}

View File

@@ -61,8 +61,8 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
private List skipRecords = new ArrayList();
private ExecutionContext executionContext = new ExecutionContext();
private boolean saveState = false;
/**
* Read in the next root element from the file, and return it.
*
@@ -90,7 +90,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
return item;
}
public void close() {
public void close(ExecutionContext executionContext) {
initialized = false;
if (fragmentReader == null && inputStream == null) {
return;
@@ -113,8 +113,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
public void open(ExecutionContext executionContext) {
Assert.state(resource.exists(), "Input resource does not exist: [" + resource + "]");
this.executionContext = executionContext;
try {
inputStream = resource.getInputStream();
txReader = new DefaultTransactionalEventReader(XMLInputFactory.newInstance().createXMLEventReader(
@@ -193,26 +192,13 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
}
/**
* @see ItemStream#update()
* @see ItemStream#update(ExecutionContext)
*/
public void update() {
executionContext.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
}
/**
* Restores the input source for the given restart data by rereading and
* skipping the number of records stored in the {@link ExecutionContext}.
*
* @param ExecutionContext that holds the line count from the last
* commit.
* @throws IllegalStateException if the ItemReader has already been
* initialized or if the number of records to read and skip exceeds the
* available records.
*/
public void restoreFrom(ExecutionContext data) {
public void update(ExecutionContext executionContext) {
if(saveState){
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
}
}
/**
@@ -273,4 +259,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
fragmentReader.reset();
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
}

View File

@@ -101,8 +101,10 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
// current count of processed records
private long currentRecordCount = 0;
private ExecutionContext executionContext = new ExecutionContext();
private String name = StaxEventItemWriter.class.getName();
private boolean saveState = false;
/**
* Set output file.
*
@@ -223,13 +225,12 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
* @see org.springframework.batch.item.ResourceLifecycle#open()
*/
public void open(ExecutionContext executionContext) {
this.executionContext = executionContext;
long startAtPosition = 0;
// if restart data is provided, restart from provided offset
// otherwise start from beginning
if (executionContext.containsKey(RESTART_DATA_NAME)) {
startAtPosition = executionContext.getLong(RESTART_DATA_NAME);
if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) {
startAtPosition = executionContext.getLong(getKey(RESTART_DATA_NAME));
restarted = true;
}
@@ -330,9 +331,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Close the output source.
*
* @see org.springframework.batch.item.ResourceLifecycle#close()
* @see org.springframework.batch.item.ResourceLifecycle#close(ExecutionContext)
*/
public void close() {
public void close(ExecutionContext executionContext) {
initialized = false;
try {
endDocument(delegateEventWriter);
@@ -365,14 +366,18 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Get the restart data.
* @see org.springframework.batch.item.ItemStream#update()
* @see org.springframework.batch.item.ItemStream#update(ExecutionContext)
*/
public void update() {
public void update(ExecutionContext executionContext) {
if (!initialized) {
throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context.");
}
executionContext.putLong(RESTART_DATA_NAME, getPosition());
executionContext.putLong(WRITE_STATISTICS_NAME, currentRecordCount);
if(saveState){
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(getKey(RESTART_DATA_NAME), getPosition());
executionContext.putLong(getKey(WRITE_STATISTICS_NAME), currentRecordCount);
}
}
/**
@@ -432,12 +437,24 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
public void clear() throws ClearFailedException {
currentRecordCount = lastCommitPointRecordCount;
// close output
close();
close(null);
// and reopen it - we do this because we need to reopen stream
// reader at specified position - calling setPosition() is not
// enough!
restarted = true;
open(lastCommitPointPosition);
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
public void setName(String name) {
this.name = name;
}
private String getKey(String key){
return name + "." + key;
}
}

View File

@@ -35,20 +35,23 @@ public interface ItemStream {
*
* @throws IllegalArgumentException if context is null
*/
void open(ExecutionContext context) throws StreamException;
void open(ExecutionContext executionContext) throws StreamException;
/**
* Indicates that the execution context provided during open
* is about to be saved. If any state is remaining, but
* has not been put in the context, it should be added
* here.
* @param executionContext to be updated
* @throws IllegalArgumentException if executionContext is null.
*/
void update();
void update(ExecutionContext executionContext);
/**
* If any resources are needed for the stream to operate they need to be
* destroyed here. Once this method has been called all other methods
* (except open) may throw an exception.
* @param executionContext TODO
*/
void close() throws StreamException;
void close(ExecutionContext executionContext) throws StreamException;
}

View File

@@ -50,13 +50,13 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
}
/**
* @see ItemStream#update()
* @see ItemStream#update(ExecutionContext)
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public void update() {
public void update(ExecutionContext executionContext) {
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).update();
((ItemStream) itemReader).update(executionContext);
}
}
@@ -95,9 +95,9 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#open()
*/
public void close() throws StreamException {
public void close(ExecutionContext executionContext) throws StreamException {
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).close();
((ItemStream) itemReader).close(null);
}
}

View File

@@ -16,6 +16,7 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
@@ -40,7 +41,7 @@ public class ItemReaderAdapter extends AbstractMethodInvokingDelegator implement
/**
* Do nothing.
*
* @see org.springframework.batch.item.ItemReader#close()
* @see org.springframework.batch.item.ItemReader#close(ExecutionContext)
*/
public void close() throws StreamException {

View File

@@ -29,9 +29,9 @@ public class ItemStreamSupport implements ItemStream {
/**
* No-op.
* @see org.springframework.batch.item.ItemStream#close()
* @see org.springframework.batch.item.ItemStream#close(ExecutionContext)
*/
public void close() throws StreamException {
public void close(ExecutionContext executionContext) throws StreamException {
}
/**
@@ -43,9 +43,9 @@ public class ItemStreamSupport implements ItemStream {
/**
* Return empty {@link ExecutionContext}.
* @see org.springframework.batch.item.ExecutionContextProvider#update()
* @see org.springframework.batch.item.ExecutionContextProvider#update(ExecutionContext)
*/
public void update() {
public void update(ExecutionContext executionContext) {
}
}

View File

@@ -68,11 +68,11 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getExecutionContext(java.lang.Object)
*/
public void update() {
public void update(ExecutionContext executionContext) {
synchronized (streams) {
for (Iterator it = streams.iterator(); it.hasNext();) {
ItemStream itemStream = (ItemStream) it.next();
itemStream.update();
itemStream.update(executionContext);
}
}
}
@@ -96,11 +96,11 @@ public class SimpleStreamManager implements StreamManager {
* Broadcast the call to close from this {@link StreamManager}.
* @throws StreamException
*/
public void close() throws StreamException {
public void close(ExecutionContext executionContext) throws StreamException {
synchronized (streams) {
for (Iterator it = streams.iterator(); it.hasNext();) {
ItemStream itemStream = (ItemStream) it.next();
itemStream.close();
itemStream.close(executionContext);
}
}
}

View File

@@ -49,7 +49,7 @@ public interface StreamManager {
* @return {@link ExecutionContext} aggregating the contexts of all providers
* registered under this key, or empty otherwise.
*/
void update();
void update(ExecutionContext executionContext);
/**
* If any resources are needed for the stream to operate they need to be
@@ -58,7 +58,7 @@ public interface StreamManager {
* @param key the key under which {@link ItemStream} instances might have
* been registered.
*/
void close() throws StreamException;
void close(ExecutionContext executionContext) throws StreamException;
/**
* If any resources are needed for the stream to operate they need to be

View File

@@ -75,17 +75,17 @@ public class DelegatingItemWriter implements ItemWriter, ItemStream, Initializin
/**
* @throws StreamException
* @see org.springframework.batch.item.ItemStream#close()
* @see org.springframework.batch.item.ItemStream#close(ExecutionContext)
*/
public void close() throws StreamException {
stream.close();
public void close(ExecutionContext executionContext) throws StreamException {
stream.close(null);
}
/**
* @see org.springframework.batch.item.ExecutionContextProvider#update()
* @see org.springframework.batch.item.ExecutionContextProvider#update(ExecutionContext)
*/
public void update() {
stream.update();
public void update(ExecutionContext executionContext) {
stream.update(executionContext);
}
/**