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

BATCH-365: There should now be one ExecutionContext per step. All itemStreams will be opened with an execution context, and will be notified before it is saved, to ensure they have all state in the context. Most ItemReader/Writers should now have the logic for whether or not to put their state in the context, but a few have likely been missed.

Browse Source
This commit is contained in:
lucasward
2008-02-26 08:29:37 +00:00
parent 3b48ece7e5
commit 64506040ff
42 changed files with 512 additions and 914 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java
View File

@@ -17,7 +17,6 @@ package org.springframework.batch.io.cursor;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.hibernate.ScrollableResults;
import org.hibernate.Session;
@@ -57,7 +56,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
+ ".rowNumber";
private static final String SKIPPED_ROWS = ClassUtils.getShortName(HibernateCursorItemReader.class)
+ ".skippedRows";;
+ ".skippedRows";
private SessionFactory sessionFactory;
@@ -81,10 +80,14 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
private int currentProcessedRow = 0;
private boolean initialized = false;
private ExecutionContext executionContext = new ExecutionContext();
private String name = HibernateCursorItemReader.class.getName();
public Object read() {
if (!initialized) {
open();
open(new ExecutionContext());
}
if (cursor.next()) {
currentProcessedRow++;
@@ -123,7 +126,11 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
* Creates cursor for the query.
*/
public void open() {
public void open(ExecutionContext executionContext) {
this.executionContext = executionContext;
Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
if (useStatelessSession) {
statelessSession = sessionFactory.openStatelessSession();
cursor = statelessSession.createQuery(queryString).scroll();
@@ -133,6 +140,18 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
cursor = statefulSession.createQuery(queryString).scroll();
}
initialized = true;
if (executionContext.containsKey(getKey(RESTART_DATA_ROW_NUMBER_KEY))) {
currentProcessedRow = Integer.parseInt(executionContext.getString(getKey(RESTART_DATA_ROW_NUMBER_KEY)));
cursor.setRowNumber(currentProcessedRow - 1);
}
if (executionContext.containsKey(getKey(SKIPPED_ROWS))) {
String[] skipped = StringUtils.commaDelimitedListToStringArray(executionContext.getString(getKey(SKIPPED_ROWS)));
for (int i = 0; i < skipped.length; i++) {
this.skippedRows.add(new Integer(skipped[i]));
}
}
}
/**
@@ -167,41 +186,11 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
}
/**
* @return the current row number wrapped as {@link ExecutionContext}
*/
public ExecutionContext getExecutionContext() {
Properties props = new Properties();
props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, "" + currentProcessedRow);
ExecutionContext executionContext = new ExecutionContext();
executionContext.putString(RESTART_DATA_ROW_NUMBER_KEY, "" + currentProcessedRow);
public void beforeSave() {
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
String skipped = skippedRows.toString();
executionContext.putString(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1));
return executionContext;
}
/**
* Sets the cursor to the received row number.
*/
public void restoreFrom(ExecutionContext data) {
Assert.state(!initialized, "Cannot restore when already intialized. Call close() first before restore()");
Properties props = data.getProperties();
if (props.getProperty(RESTART_DATA_ROW_NUMBER_KEY) == null) {
return;
}
currentProcessedRow = Integer.parseInt(props.getProperty(RESTART_DATA_ROW_NUMBER_KEY));
open();
cursor.setRowNumber(currentProcessedRow - 1);
if (!props.containsKey(SKIPPED_ROWS)) {
return;
}
String[] skipped = StringUtils.commaDelimitedListToStringArray(props.getProperty(SKIPPED_ROWS));
for (int i = 0; i < skipped.length; i++) {
this.skippedRows.add(new Integer(skipped[i]));
}
executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
}
/**
@@ -254,4 +243,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
}
}
private String getKey(String key){
return name + "." + key;
}
}

65
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java
View File

@@ -29,7 +29,6 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
@@ -108,7 +107,7 @@ import org.springframework.util.StringUtils;
* @author Lucas Ward
* @author Peter Zozom
*/
public class JdbcCursorItemReader extends AbstractTransactionalIoSource implements KeyedItemReader, InitializingBean,
public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
ItemStream, Skippable {
private static Log log = LogFactory.getLog(JdbcCursorItemReader.class);
@@ -155,6 +154,12 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
private RowMapper mapper;
private boolean initialized = false;
private ExecutionContext executionContext = new ExecutionContext();
private boolean saveState = false;
private String name = JdbcCursorItemReader.class.getName();
/**
* Assert that mandatory properties are set.
@@ -190,7 +195,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
public Object read() {
if (!initialized) {
open();
open(new ExecutionContext());
}
Assert.state(mapper != null, "Mapper must not be null.");
@@ -380,45 +385,45 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#getExecutionContext()
*/
public ExecutionContext getExecutionContext() {
String skipped = skippedRows.toString();
ExecutionContext context = new ExecutionContext();
context.putString(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1));
context.putLong(CURRENT_PROCESSED_ROW, currentProcessedRow);
context.putLong(SKIP_COUNT, skipCount);
return context;
public void beforeSave() {
if(saveState){
String skipped = skippedRows.toString();
executionContext.putString(addName(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
executionContext.putLong(addName(CURRENT_PROCESSED_ROW), currentProcessedRow);
executionContext.putLong(addName(SKIP_COUNT), skipCount);
}
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#restoreFrom(org.springframework.batch.item.ExecutionContext)
*/
public void restoreFrom(ExecutionContext data) {
public void open(ExecutionContext context) {
Assert.state(!initialized);
if (data == null)
return;
open();
Assert.isNull(rs);
Assert.notNull(context, "ExecutionContext must not be null");
executeQuery();
initialized = true;
this.executionContext = context;
// Properties restartProperties = data.getProperties();
if (!data.containsKey(CURRENT_PROCESSED_ROW)) {
if (!context.containsKey(addName(CURRENT_PROCESSED_ROW))) {
return;
}
try {
this.currentProcessedRow = data.getLong(CURRENT_PROCESSED_ROW);
this.currentProcessedRow = context.getLong(addName(CURRENT_PROCESSED_ROW));
rs.absolute((int) currentProcessedRow);
}
catch (SQLException se) {
throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se);
}
if (!data.containsKey(SKIPPED_ROWS)) {
if (!context.containsKey(addName(SKIPPED_ROWS))) {
return;
}
String[] skipped = StringUtils.commaDelimitedListToStringArray(data.getString(SKIPPED_ROWS));
String[] skipped = StringUtils.commaDelimitedListToStringArray(context.getString(addName(SKIPPED_ROWS)));
for (int i = 0; i < skipped.length; i++) {
this.skippedRows.add(new Long(skipped[i]));
}
@@ -513,13 +518,6 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
this.sql = sql;
}
public void open() {
Assert.isNull(rs);
executeQuery();
initialized = true;
}
/**
* Return the item itself (which is already a key).
* @see org.springframework.batch.item.ItemReader#getKey(java.lang.Object)
@@ -527,5 +525,16 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
public Object getKey(Object item) {
return item;
}
public void setName(String name) {
this.name = name;
}
public String addName(String key){
return name + "." + key;
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
}

57
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -18,7 +18,6 @@ package org.springframework.batch.io.driving;
import java.util.Iterator;
import java.util.List;
import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
@@ -47,7 +46,7 @@ import org.springframework.util.Assert;
* @author Lucas Ward
* @since 1.0
*/
public class DrivingQueryItemReader extends AbstractTransactionalIoSource implements KeyedItemReader, InitializingBean,
public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean,
ItemStream {
private boolean initialized = false;
@@ -61,6 +60,10 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource implem
private int lastCommitIndex = 0;
private KeyGenerator keyGenerator;
private ExecutionContext executionContext = new ExecutionContext();
private boolean saveState = false;
public DrivingQueryItemReader() {
@@ -86,7 +89,7 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource implem
*/
public Object read() {
if (!initialized) {
open();
open(new ExecutionContext());
}
if (keysIterator.hasNext()) {
@@ -133,49 +136,22 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource implem
* @throws IllegalStateException if the keys list is null or initialized is
* true.
*/
public void open() {
public void open(ExecutionContext executionContext) {
Assert.state(keys == null || initialized, "Cannot open an already opened input source"
Assert.state(keys == null && !initialized, "Cannot open an already opened input source"
+ ", call close() first.");
keys = keyGenerator.retrieveKeys();
keys = keyGenerator.retrieveKeys(executionContext);
keysIterator = keys.listIterator();
initialized = true;
this.executionContext = executionContext;
}
/**
* Restore input source to previous state. If the input source has already
* been initialized before calling restore (meaning, read has been called)
* then an IllegalStateException will be thrown, since all input sources
* should be restored before being read from, otherwise already processed
* data could be returned. The {@link ExecutionContext} attempting to be
* restored from must have been obtained from the <strong>same input source
* as the one being restored from</strong> otherwise it is invalid.
*
* @throws IllegalArgumentException if restart data or it's properties is
* null.
* @throws IllegalStateException if the input source has already been
* initialized.
*/
public final void restoreFrom(ExecutionContext data) {
Assert.notNull(data, "ExecutionContext must not be null.");
Assert.notNull(data.getProperties(), "ExecutionContext properties must not be null.");
Assert.state(!initialized, "Cannot restore when already intialized. Call" + " close() first before restore()");
if (data.getProperties().size() == 0) {
return;
public void beforeSave() {
if(saveState){
if(getCurrentKey() != null){
keyGenerator.saveState(getCurrentKey(), executionContext);
}
}
keys = keyGenerator.restoreKeys(data);
if (keys != null && keys.size() > 0) {
keysIterator = keys.listIterator();
initialized = true;
}
}
public ExecutionContext getExecutionContext() {
return keyGenerator.getKeyAsExecutionContext(getCurrentKey());
}
public void afterPropertiesSet() throws Exception {
@@ -238,4 +214,7 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource implem
keysIterator = keys.listIterator(lastCommitIndex);
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java
View File

@@ -13,18 +13,10 @@ import org.springframework.batch.item.ExecutionContext;
public interface KeyGenerator {
/**
* @param executionContext TODO
* @return list of keys returned by the driving query
*/
List retrieveKeys();
/**
* Restore the keys list based on provided restart data.
*
* @param executionContext, the restart data to restore the keys list from.
* @return a list of keys.
* @throws IllegalArgumentException if executionContext is null.
*/
List restoreKeys(ExecutionContext executionContext);
List retrieveKeys(ExecutionContext executionContext);
/**
* Return the provided key as restart data.
@@ -34,5 +26,5 @@ public interface KeyGenerator {
* @throws IllegalArgumentException if key is null.
* @throws IllegalArgumentException if key is an incompatible type.
*/
ExecutionContext getKeyAsExecutionContext(Object key);
void saveState(Object key, ExecutionContext executionContext);
}

55
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java
View File

@@ -9,11 +9,9 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.CollectionFactory;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.SqlParameterValue;
@@ -39,11 +37,8 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
public static final String KEY_PREFIX = ClassUtils.getQualifiedName(ColumnMapExecutionContextRowMapper.class) + ".KEY.";
public PreparedStatementSetter createSetter(ExecutionContext executionContext) {
ColumnMapExecutionContext columnData = new ColumnMapExecutionContext(executionContext.getProperties());
List columns = new ArrayList();
for (Iterator iterator = columnData.keys.entrySet().iterator(); iterator.hasNext();) {
for (Iterator iterator = executionContext.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
Object column = entry.getValue();
columns.add(column);
@@ -52,52 +47,13 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
return new ArgPreparedStatementSetter(columns.toArray());
}
public ExecutionContext createExecutionContext(Object key) {
public void mapKeys(Object key, ExecutionContext executionContext) {
Assert.isInstanceOf(Map.class, key, "Input to create ExecutionContext must be of type Map.");
Map keys = (Map) key;
return new ColumnMapExecutionContext(keys);
}
private static class ColumnMapExecutionContext extends ExecutionContext {
private final Map keys;
public ColumnMapExecutionContext(Map keys) {
this.keys = keys;
for (Iterator it = keys.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry)it.next();
executionContext.put(entry.getKey().toString(), entry.getValue());
}
public ColumnMapExecutionContext(Properties props) {
keys = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(props.size());
for (int counter = 0; counter < props.size(); counter++) {
String key = KEY_PREFIX + counter;
String column = props.getProperty(key);
if (column != null) {
keys.put(key, column);
}
else {
break;
}
}
}
public Properties getProperties() {
Properties props = new Properties();
int counter = 0;
for (Iterator iterator = keys.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
props.setProperty(KEY_PREFIX + counter, entry.getValue().toString());
counter++;
}
return props;
}
}
/*
@@ -131,4 +87,5 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
}
}
}
}

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ExecutionContextRowMapper.java
View File

@@ -40,7 +40,7 @@ public interface ExecutionContextRowMapper extends RowMapper {
* @return ExecutionContext representing the composite key.
* @throws IllegalArgumentException if key is null or of an unsupported type.
*/
public ExecutionContext createExecutionContext(Object key);
public void mapKeys(Object key, ExecutionContext executionContext);
/**
* Given the provided restart data, return a PreparedStatementSeter that can

43
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java
View File

@@ -1,7 +1,6 @@
package org.springframework.batch.io.driving.support;
import java.util.List;
import java.util.Properties;
import org.springframework.batch.io.driving.DrivingQueryItemReader;
import org.springframework.batch.io.driving.KeyGenerator;
@@ -29,37 +28,31 @@ public class IbatisKeyGenerator implements KeyGenerator {
private String drivingQuery;
private String restartQueryId;
private String name = IbatisKeyGenerator.class.getName();
/*
* Retrieve the keys using the provided driving query id.
*
* @see org.springframework.batch.io.support.AbstractDrivingQueryItemReader#retrieveKeys()
*/
public List retrieveKeys() {
return sqlMapClientTemplate.queryForList(drivingQuery);
public List retrieveKeys(ExecutionContext executionContext) {
if(executionContext.containsKey(getKey(RESTART_KEY))){
Object key = executionContext.getString(getKey(RESTART_KEY));
return sqlMapClientTemplate.queryForList(restartQueryId, key);
}
else{
return sqlMapClientTemplate.queryForList(drivingQuery);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext(java.lang.Object)
*/
public ExecutionContext getKeyAsExecutionContext(Object key) {
Properties props = new Properties();
props.setProperty(RESTART_KEY, key.toString());
ExecutionContext executionContext = new ExecutionContext();
executionContext.putString(RESTART_KEY, key.toString());
return executionContext;
}
/**
* Restore the keys list given the provided restart data.
*
* @see org.springframework.batch.io.driving.DrivingQueryItemReader#restoreKeys(org.springframework.batch.item.ExecutionContext)
*/
public List restoreKeys(ExecutionContext data) {
Properties props = data.getProperties();
Object key = props.getProperty(RESTART_KEY);
return sqlMapClientTemplate.queryForList(restartQueryId, key);
public void saveState(Object key, ExecutionContext executionContext) {
Assert.notNull(key, "Key must not be null");
Assert.notNull(executionContext, "ExecutionContext must be null");
executionContext.putString(getKey(RESTART_KEY), key.toString());
}
/* (non-Javadoc)
@@ -99,4 +92,12 @@ public class IbatisKeyGenerator implements KeyGenerator {
public final SqlMapClientTemplate getSqlMapClientTemplate() {
return sqlMapClientTemplate;
}
public void setName(String name) {
this.name = name;
}
private String getKey(String key){
return name + "." + key;
}
}

23
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java
View File

@@ -15,7 +15,6 @@
*/
package org.springframework.batch.io.driving.support;
import java.util.ArrayList;
import java.util.List;
import org.springframework.batch.io.driving.DrivingQueryItemReader;
@@ -73,32 +72,26 @@ public class MultipleColumnJdbcKeyGenerator implements
/* (non-Javadoc)
* @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryItemReader#retrieveKeys()
*/
public List retrieveKeys() {
return jdbcTemplate.query(sql, keyMapper);
}
/* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerator#restoreKeys(org.springframework.batch.item.ExecutionContext)
*/
public List restoreKeys(ExecutionContext executionContext) {
public List retrieveKeys(ExecutionContext executionContext) {
Assert.state(keyMapper != null, "KeyMapper must not be null.");
Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty" +
" in order to restart.");
if (executionContext.getProperties() != null) {
if (executionContext.size() > 0) {
return jdbcTemplate.query(restartSql, keyMapper.createSetter(executionContext), keyMapper);
}
return new ArrayList();
else{
return jdbcTemplate.query(sql, keyMapper);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext(java.lang.Object)
*/
public ExecutionContext getKeyAsExecutionContext(Object key) {
public void saveState(Object key, ExecutionContext executionContext) {
Assert.state(keyMapper != null, "Kye mapper must not be null.");
return keyMapper.createExecutionContext(key);
keyMapper.mapKeys(key, executionContext);
}
/**

51
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java
View File

@@ -15,7 +15,6 @@
*/
package org.springframework.batch.io.driving.support;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.ClassUtils;
@@ -88,49 +87,29 @@ public class SingleColumnJdbcKeyGenerator implements KeyGenerator {
* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerationStrategy#retrieveKeys()
*/
public List retrieveKeys() {
return jdbcTemplate.query(sql, keyMapper);
public List retrieveKeys(ExecutionContext executionContext) {
Assert.notNull(executionContext, "The restart data must not be null.");
if (executionContext.containsKey(RESTART_KEY)) {
Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty"
+ " in order to restart.");
return jdbcTemplate.query(restartSql, new Object[] { executionContext.getString(RESTART_KEY) }, keyMapper);
}
else{
return jdbcTemplate.query(sql, keyMapper);
}
}
/**
* Get the restart data representing the last processed key.
*
* @see KeyGenerator#getKeyAsExecutionContext(Object)
* @see KeyGenerator#saveState(Object)
* @throws IllegalArgumentException if key is null.
*/
public ExecutionContext getKeyAsExecutionContext(Object key) {
public void saveState(Object key, ExecutionContext executionContext) {
Assert.notNull(key, "The key must not be null.");
ExecutionContext context = new ExecutionContext();
context.putString(RESTART_KEY, key.toString());
return context;
}
/**
* Return the remaining to be processed for the provided
* {@link ExecutionContext}. The {@link ExecutionContext} attempting to be
* restored from must have been obtained from the <strong>same
* KeyGenerationStrategy as the one being restored from</strong> otherwise
* it is invalid.
*
* @param executionContext {@link ExecutionContext} obtained by calling
* {@link #getKeyAsExecutionContext(Object)} during a previous run.
* @throws IllegalStateException if restart sql statement is null.
* @throws IllegalArgumentException if restart data is null.
* @see KeyGenerator#restoreKeys(org.springframework.batch.item.ExecutionContext)
*/
public List restoreKeys(ExecutionContext executionContext) {
Assert.notNull(executionContext, "The restart data must not be null.");
Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty"
+ " in order to restart.");
String lastProcessedKey = executionContext.getProperties().getProperty(RESTART_KEY);
if (lastProcessedKey != null) {
return jdbcTemplate.query(restartSql, new Object[] { lastProcessedKey }, keyMapper);
}
return new ArrayList();
executionContext.putString(RESTART_KEY, key.toString());
}
/*

63
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java
View File

@@ -93,6 +93,10 @@ 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();
/**
* Encapsulates the state of the input source. If it is null then we are
@@ -104,10 +108,12 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* Initialize the reader if necessary.
* @throws IllegalStateException if the resource cannot be opened
*/
public void open() throws StreamException {
public void open(ExecutionContext executionContext) throws StreamException {
Assert.state(resource.exists(), "Resource must exist: [" + resource + "]");
this.executionContext = executionContext;
log.debug("Opening flat file for reading: " + resource);
if (this.reader == null) {
@@ -135,7 +141,19 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
((AbstractLineTokenizer) tokenizer).setNames(names);
}
}
if (executionContext.containsKey(getKey(READ_STATISTICS_NAME))) {
log.debug("Initializing for restart. Restart data is: " + executionContext);
long lineCount = executionContext.getLong(getKey(READ_STATISTICS_NAME));
LineReader reader = getReader();
Object record = "";
while (reader.getPosition() < lineCount && record != null) {
record = readLine();
}
}
mark();
}
@@ -179,48 +197,17 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
return null;
}
/**
* This method initialises the reader for Restart. It opens the input file
* and position the buffer reader according to information provided by the
* restart data
*
* @param data {@link ExecutionContext} information
*/
public void restoreFrom(ExecutionContext data) {
if (data == null || data.getProperties() == null
|| data.getProperties().getProperty(READ_STATISTICS_NAME) == null || getReader() == null) {
// do nothing
return;
}
log.debug("Initializing for restart. Restart data is: " + data);
int lineCount = Integer.parseInt(data.getProperties().getProperty(READ_STATISTICS_NAME));
LineReader reader = getReader();
Object record = "";
while (reader.getPosition() < lineCount && record != null) {
record = readLine();
}
mark();
}
/**
* This method returns the execution attributes for the reader. It returns
* the current Line Count which can be used to reinitialise the batch job in
* case of restart.
*/
public ExecutionContext getExecutionContext() {
public void beforeSave() {
if (reader == null) {
throw new StreamException("ItemStream not open or already closed.");
}
ExecutionContext executionContext = new ExecutionContext();
executionContext.putLong(READ_STATISTICS_NAME, reader.getPosition());
executionContext.putLong(SKIPPED_STATISTICS_NAME, skippedLines.size());
return executionContext;
executionContext.putLong(getKey(READ_STATISTICS_NAME), reader.getPosition());
executionContext.putLong(getKey(SKIPPED_STATISTICS_NAME), skippedLines.size());
}
/**
@@ -390,5 +377,9 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
Assert.notNull(resource, "Input resource must not be null");
Assert.notNull(fieldSetMapper, "FieldSetMapper must not be null.");
}
private String getKey(String key){
return name + "." + key;
}
}

47
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java
View File

@@ -24,7 +24,6 @@ import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.charset.UnsupportedCharsetException;
import java.util.Properties;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.io.exception.BatchEnvironmentException;
@@ -74,7 +73,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
public static final String RESTART_COUNT_STATISTICS_NAME = "restart.count";
public static final String RESTART_DATA_NAME = "flatfileoutput.current.line";
public static final String RESTART_DATA_NAME = "current.count";
private Resource resource;
@@ -85,7 +84,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
private LineAggregator lineAggregator = new DelimitedLineAggregator();
private FieldSetCreator fieldSetCreator;
private String name = FlatFileItemWriter.class.getName();
/**
* Assert that mandatory properties (resource) are set.
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
@@ -180,30 +181,24 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
* Initialize the Output Template.
* @see ResourceLifecycle#open()
*/
public void open() {
getOutputState();
public void open(ExecutionContext executionContext) {
this.executionContext = executionContext;
OutputState outputState = getOutputState();
if(executionContext.containsKey(getKey(RESTART_DATA_NAME))){
outputState.restoreFrom(executionContext);
}
}
/**
* @see ItemStream#getExecutionContext()
* @see ItemStream#beforeSave()
*/
public ExecutionContext getExecutionContext() {
public void beforeSave() {
if (state == null) {
throw new StreamException("ItemStream not open or already closed.");
}
executionContext.putLong(RESTART_DATA_NAME, state.position());
executionContext.putLong(WRITTEN_STATISTICS_NAME, state.linesWritten);
executionContext.putLong(RESTART_COUNT_STATISTICS_NAME, state.restartCount);
return executionContext;
}
/**
* @see ItemStream#restoreFrom(ExecutionContext)
*/
public void restoreFrom(ExecutionContext data) {
if (data == null)
return;
getOutputState().restoreFrom(data.getProperties());
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);
}
// Returns object representing state.
@@ -274,8 +269,8 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
/**
* @param properties
*/
public void restoreFrom(Properties properties) {
lastMarkedByteOffsetPosition = Long.parseLong(properties.getProperty(RESTART_DATA_NAME));
public void restoreFrom(ExecutionContext executionContext) {
lastMarkedByteOffsetPosition = executionContext.getLong(getKey(RESTART_DATA_NAME));
restarted = true;
}
@@ -516,4 +511,12 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
public void flush() throws Exception {
getOutputState().mark();
}
public void setName(String name) {
this.name = name;
}
private String getKey(String key){
return name + "." + key;
}
}

48
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java
View File

@@ -60,6 +60,8 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
private long currentRecordCount = 0;
private List skipRecords = new ArrayList();
private ExecutionContext executionContext = new ExecutionContext();
/**
* Read in the next root element from the file, and return it.
@@ -109,8 +111,10 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
}
}
public void open() {
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(
@@ -124,7 +128,23 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
throw new DataAccessResourceFailureException("Unable to get input stream", ioe);
}
initialized = true;
mark();
if (executionContext.containsKey(READ_COUNT_STATISTICS_NAME)) {
long restoredRecordCount = executionContext.getLong(READ_COUNT_STATISTICS_NAME);
int REASONABLE_ADHOC_COMMIT_FREQUENCY = 100;
while (currentRecordCount <= restoredRecordCount) {
currentRecordCount++;
if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) {
txReader.onCommit(); // reset the history buffer
}
if (!fragmentReader.hasNext()) {
throw new StreamException("Restore point must be before end of input");
}
fragmentReader.next();
moveCursorToNextFragment(fragmentReader);
}
mark(); // reset the history buffer
}
}
public void setResource(Resource resource) {
@@ -173,13 +193,10 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
}
/**
* @return wrapped count of records read so far.
* @see ItemStream#getExecutionContext()
* @see ItemStream#beforeSave()
*/
public ExecutionContext getExecutionContext() {
ExecutionContext executionContext = new ExecutionContext();
public void beforeSave() {
executionContext.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
return executionContext;
}
/**
@@ -194,24 +211,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
*/
public void restoreFrom(ExecutionContext data) {
if (data == null || data.getProperties() == null || !data.containsKey(READ_COUNT_STATISTICS_NAME)) {
return;
}
long restoredRecordCount = data.getLong(READ_COUNT_STATISTICS_NAME);
int REASONABLE_ADHOC_COMMIT_FREQUENCY = 100;
while (currentRecordCount <= restoredRecordCount) {
currentRecordCount++;
if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) {
txReader.onCommit(); // reset the history buffer
}
if (!fragmentReader.hasNext()) {
throw new StreamException("Restore point must be before end of input");
}
fragmentReader.next();
moveCursorToNextFragment(fragmentReader);
}
mark(); // reset the history buffer
}

40
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java
View File

@@ -98,6 +98,8 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
// current count of processed records
private long currentRecordCount = 0;
private ExecutionContext executionContext = new ExecutionContext();
/**
* Set output file.
@@ -218,8 +220,18 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
*
* @see org.springframework.batch.item.ResourceLifecycle#open()
*/
public void open() {
open(0);
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);
restarted = true;
}
open(startAtPosition);
}
/*
@@ -351,17 +363,14 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Get the restart data.
* @return the restart data
* @see org.springframework.batch.item.ItemStream#getExecutionContext()
* @see org.springframework.batch.item.ItemStream#beforeSave()
*/
public ExecutionContext getExecutionContext() {
public void beforeSave() {
if (!initialized) {
throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context.");
}
ExecutionContext context = new ExecutionContext();
context.putLong(RESTART_DATA_NAME, getPosition());
context.putLong(WRITE_STATISTICS_NAME, currentRecordCount);
return context;
executionContext.putLong(RESTART_DATA_NAME, getPosition());
executionContext.putLong(WRITE_STATISTICS_NAME, currentRecordCount);
}
/**
@@ -371,19 +380,6 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
*/
public void restoreFrom(ExecutionContext data) {
long startAtPosition = 0;
// if restart data is provided, restart from provided offset
// otherwise start from beginning
if (data != null && data.getProperties() != null && data.containsKey(RESTART_DATA_NAME)) {
startAtPosition = data.getLong(RESTART_DATA_NAME);
restarted = true;
}
if (!initialized) {
open(startAtPosition);
}
}
/*

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

@@ -154,5 +154,9 @@ public class ExecutionContext {
public String toString() {
return map.toString();
}
public int size(){
return map.size();
}
}

32
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextProvider.java
View File

@@ -1,32 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.item;
/**
* @author Dave Syer
*
*/
public interface ExecutionContextProvider {
/**
* Get {@link ExecutionContext} representing this object's current state.
* Should not return null even if there is no state.
*
* @return {@link ExecutionContext} representing current state.
*/
ExecutionContext getExecutionContext();
}

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

@@ -56,7 +56,7 @@ public interface ItemReader {
/**
* Mark the stream so that it can be reset later and the items backed out.
* After this method is called the result will be reflected in subsequent
* calls to {@link ExecutionContextProvider#getExecutionContext()}.<br/>
* calls to {@link ExecutionContextProvider#beforeSave()}.<br/>
*
* In a multi-threaded setting implementations have to ensure that only the
* state from the current thread is saved.

32
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java
View File

@@ -24,35 +24,27 @@ import org.springframework.batch.item.exception.StreamException;
* restoring from that state should an error occur.
* <p>
*
* <p>
* The state that is stored is represented as {@link ExecutionContext} which
* enforces a requirement that any restart data can be represented by a
* Properties object. In general, the contract is that
* {@link ExecutionContext} that is returned via the
* {@link #getExecutionContext()} method will be given back to the
* {@link #restoreFrom(ExecutionContext)} method, exactly as it was provided.
* </p>
*
* @author Dave Syer
* @author Lucas Ward
*
*/
public interface ItemStream extends ExecutionContextProvider {
public interface ItemStream {
/**
* Restore to the state given the provided {@link ExecutionContext}.
* This can be used to restart after a failure - hence not normally used
* more than once per call to {@link #open()}.
* Open the stream for the provided {@link ExecutionContext}.
*
* @param context
* @throws IllegalArgumentException if context is null
*/
void restoreFrom(ExecutionContext context);
void open(ExecutionContext context) throws StreamException;
/**
* If any resources are needed for the stream to operate they need to be
* initialised here.
* 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.
*/
void open() throws StreamException;
void beforeSave();
/**
* 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

22
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java
View File

@@ -50,25 +50,13 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
}
/**
* @see ItemStream#getExecutionContext()
* @see ItemStream#beforeSave()
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public ExecutionContext getExecutionContext() {
public void beforeSave() {
if (itemReader instanceof ItemStream) {
return ((ItemStream) itemReader).getExecutionContext();
}
return new ExecutionContext();
}
/**
* @see ItemStream#restoreFrom(ExecutionContext)
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public void restoreFrom(ExecutionContext data) {
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).restoreFrom(data);
((ItemStream) itemReader).beforeSave();
}
}
@@ -101,9 +89,9 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#open()
*/
public void open() throws StreamException {
public void open(ExecutionContext executionContext) throws StreamException {
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).open();
((ItemStream) itemReader).open(executionContext);
}
}

33
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamSupport.java
View File

@@ -38,41 +38,14 @@ public class ItemStreamSupport implements ItemStream {
* No-op.
* @see org.springframework.batch.item.ItemStream#open()
*/
public void open() throws StreamException {
}
/**
* No-op.
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionContext)
*/
public void restoreFrom(ExecutionContext context) {
public void open(ExecutionContext executionContext) throws StreamException {
}
/**
* Return empty {@link ExecutionContext}.
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
* @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
*/
public ExecutionContext getExecutionContext() {
return new ExecutionContext();
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return false;
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
public void beforeSave() {
}
}

138
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java
View File

@@ -15,13 +15,9 @@
*/
package org.springframework.batch.item.stream;
import java.util.Collection;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.List;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
@@ -29,7 +25,6 @@ import org.springframework.batch.item.exception.StreamException;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.ClassUtils;
/**
* Simple {@link StreamManager} that tries to resolve conflicts between key
@@ -40,12 +35,10 @@ import org.springframework.util.ClassUtils;
*/
public class SimpleStreamManager implements StreamManager {
private Map registry = new HashMap();
private List streams = new ArrayList();
private PlatformTransactionManager transactionManager;
private boolean useClassNameAsPrefix = true;
/**
* @param transactionManager a {@link PlatformTransactionManager}
*/
@@ -61,18 +54,6 @@ public class SimpleStreamManager implements StreamManager {
super();
}
/**
* Public setter for the flag. If this is true then the class name of the
* streams will be used as a prefix in the {@link ExecutionContext} in
* {@link #getExecutionContext(Object)}. The default value is true, which
* gives the best chance of unique key names in the context.
*
* @param useClassNameAsPrefix the flag to set (default true).
*/
public void setUseClassNameAsPrefix(boolean useClassNameAsPrefix) {
this.useClassNameAsPrefix = useClassNameAsPrefix;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
* @param transactionManager the {@link PlatformTransactionManager} to set
@@ -87,23 +68,13 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getExecutionContext(java.lang.Object)
*/
public ExecutionContext getExecutionContext(Object key) {
final ExecutionContext result = new ExecutionContext();
iterate(key, new Callback() {
public void execute(ItemStream stream) {
ExecutionContext context = stream.getExecutionContext();
String prefix = ClassUtils.getQualifiedName(stream.getClass()) + ".";
if (!useClassNameAsPrefix) {
prefix = "";
}
for (Iterator iterator = context.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
String contextKey = prefix + entry.getKey();
result.put(contextKey, entry.getValue());
}
public void beforeSave() {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
itemStream.beforeSave();
}
});
return result;
}
}
/**
@@ -113,73 +84,36 @@ public class SimpleStreamManager implements StreamManager {
* @see org.springframework.batch.item.stream.StreamManager#register(java.lang.Object,
* org.springframework.batch.item.ItemStream, ExecutionContext)
*/
public void register(Object key, ItemStream stream) {
synchronized (registry) {
Set set = (Set) registry.get(key);
if (set == null) {
set = new LinkedHashSet();
registry.put(key, set);
}
set.add(stream);
public void register(ItemStream stream) {
synchronized (streams) {
streams.add(stream);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.StreamManager#restoreFrom(java.lang.Object, org.springframework.batch.item.ExecutionAttributes)
*/
public void restoreFrom(Object key, final ExecutionContext executionContext) {
if (executionContext != null) {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
stream.restoreFrom(extract(stream, executionContext));
}
});
}
}
/**
* @param stream
* @param executionContext
* @return
*/
private ExecutionContext extract(ItemStream stream, ExecutionContext executionContext) {
ExecutionContext result = new ExecutionContext();
String prefix = ClassUtils.getQualifiedName(stream.getClass()) + ".";
if (!useClassNameAsPrefix) {
prefix = "";
}
for (Iterator iterator = executionContext.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
String contextKey = (String) entry.getKey();
if (contextKey.startsWith(prefix)) {
result.putString(contextKey.substring(prefix.length()), entry.getValue().toString());
}
}
return result;
}
/**
* Broadcast the call to close from this {@link StreamManager}.
* @throws StreamException
*/
public void close(Object key) throws StreamException {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
stream.close();
public void close() throws StreamException {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
itemStream.close();
}
});
}
}
/**
* Broadcast the call to open from this {@link StreamManager}.
* @throws StreamException
*/
public void open(Object key) throws StreamException {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
stream.open();
public void open(ExecutionContext executionContext) throws StreamException {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
itemStream.open(executionContext);
}
});
}
}
/**
@@ -188,28 +122,11 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getTransaction()
*/
public TransactionStatus getTransaction(final Object key) {
public TransactionStatus getTransaction() {
TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition());
return transaction;
}
/**
* @param key
*/
private void iterate(Object key, Callback callback) {
Set set = new LinkedHashSet();
synchronized (registry) {
Collection collection = (Collection) registry.get(key);
if (collection != null) {
set.addAll(collection);
}
}
for (Iterator iterator = set.iterator(); iterator.hasNext();) {
ItemStream stream = (ItemStream) iterator.next();
callback.execute(stream);
}
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.stream.StreamManager#commit(java.lang.Object)
@@ -225,9 +142,4 @@ public class SimpleStreamManager implements StreamManager {
public void rollback(TransactionStatus status) {
transactionManager.rollback(status);
}
private interface Callback {
void execute(ItemStream stream);
}
}

18
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java
View File

@@ -38,7 +38,7 @@ public interface StreamManager {
* @param key the key under which to add the provider
* @param stream an {@link ItemStream}
*/
void register(Object key, ItemStream stream);
void register(ItemStream stream);
/**
* Extract and aggregate the {@link ExecutionContext} from all streams under
@@ -49,7 +49,7 @@ public interface StreamManager {
* @return {@link ExecutionContext} aggregating the contexts of all providers
* registered under this key, or empty otherwise.
*/
ExecutionContext getExecutionContext(Object key);
void beforeSave();
/**
* 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(Object key) throws StreamException;
void close() throws StreamException;
/**
* If any resources are needed for the stream to operate they need to be
@@ -67,20 +67,12 @@ public interface StreamManager {
* @param key the key under which {@link ItemStream} instances might have
* been registered.
*/
void open(Object key) throws StreamException;
void open(ExecutionContext executionContext) throws StreamException;
TransactionStatus getTransaction(Object key);
TransactionStatus getTransaction();
void commit(TransactionStatus transaction);
void rollback(TransactionStatus transaction);
/**
* Restore the streams registered under a goven key.
*
* @param key the key under which the streams are stored
* @param executionContext the context (may be null) to restore from
*/
void restoreFrom(Object key, ExecutionContext executionContext);
}

22
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java
View File

@@ -80,29 +80,17 @@ public class DelegatingItemWriter implements ItemWriter, ItemStream, Initializin
}
/**
* @return
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
* @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
*/
public ExecutionContext getExecutionContext() {
return stream.getExecutionContext();
public void beforeSave() {
stream.beforeSave();
}
/**
* @throws StreamException
* @see org.springframework.batch.item.ItemStream#open()
*/
public void open() throws StreamException {
stream.open();
public void open(ExecutionContext executionContext) throws StreamException {
stream.open(executionContext);
}
/**
* @param context
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionContext)
*/
public void restoreFrom(ExecutionContext context) {
stream.restoreFrom(context);
}
}