diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java index 3fc7b0a70..34aae62f5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java @@ -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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java index fe14c643b..32f55fc81 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java @@ -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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java index 5629c9bd0..aaf199d40 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java @@ -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 same input source - * as the one being restored from 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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java index bf4e8c06a..145ed4403 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java @@ -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); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java index d9c22b521..d27e0601f 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java @@ -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 } } } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ExecutionContextRowMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ExecutionContextRowMapper.java index 4170302c5..991c39e34 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ExecutionContextRowMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ExecutionContextRowMapper.java @@ -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 diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java index b86aaed86..d276215ee 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java @@ -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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java index eec7406f0..482b2bc0a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java @@ -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); } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java index 66ccc90a3..dafa11577 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java @@ -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 same - * KeyGenerationStrategy as the one being restored from 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()); } /* diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java index f07beb5ce..7a92021f1 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java @@ -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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java index cc2e04ec8..20faa0adc 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java @@ -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; + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java index 3143dc092..15ac9f287 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java @@ -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 } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index 3a59cf6a1..3cf5e5dfd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -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); - } - } /* diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java index d27eda727..e8e6c1c3e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java @@ -154,5 +154,9 @@ public class ExecutionContext { public String toString() { return map.toString(); } + + public int size(){ + return map.size(); + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextProvider.java deleted file mode 100644 index 483021495..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextProvider.java +++ /dev/null @@ -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(); - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java index 39aa80af1..a4acebcbe 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java @@ -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()}.
+ * calls to {@link ExecutionContextProvider#beforeSave()}.
* * In a multi-threaded setting implementations have to ensure that only the * state from the current thread is saved. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java index 82e362453..ac91d3ee7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java @@ -24,35 +24,27 @@ import org.springframework.batch.item.exception.StreamException; * restoring from that state should an error occur. *

* - *

- * 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. - *

- * * @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 diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java index f2c5b6bc3..37c448768 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java @@ -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); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamSupport.java index ca97f8171..5be64c8b3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamSupport.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamSupport.java @@ -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() { } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java index d1266871a..e2f778783 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java @@ -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); - } - } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java index 6665ebd1b..e8dafd9fb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java @@ -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); - } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java index 4c03c912b..48a836788 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java @@ -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); - } - - - } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorItemReaderIntegrationTests.java index 1eecf27e2..c6868263f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorItemReaderIntegrationTests.java @@ -3,6 +3,7 @@ package org.springframework.batch.io.cursor; import org.hibernate.SessionFactory; import org.hibernate.StatelessSession; import org.springframework.batch.io.support.AbstractDataSourceItemReaderIntegrationTests; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; @@ -48,7 +49,7 @@ public class HibernateCursorItemReaderIntegrationTests extends AbstractDataSourc HibernateCursorItemReader inputSource = ((HibernateCursorItemReader) reader); // initialize and call setter => error - inputSource.open(); + inputSource.open(new ExecutionContext()); try { inputSource.setUseStatelessSession(false); fail(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorItemReaderIntegrationTests.java index 70bc4bc23..ae12e1120 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorItemReaderIntegrationTests.java @@ -22,6 +22,7 @@ public class JdbcCursorItemReaderIntegrationTests extends AbstractDataSourceItem result.setFetchSize(10); result.setMaxRows(100); result.setQueryTimeout(1000); + result.setSaveState(true); return result; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java index c56f34e36..0fb2a0e7e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java @@ -32,6 +32,7 @@ public class DrivingQueryItemReaderTests extends TestCase { DrivingQueryItemReader inputSource = new DrivingQueryItemReader(); inputSource.setKeyGenerator(new MockKeyGenerator()); + inputSource.setSaveState(true); return inputSource; } @@ -67,18 +68,22 @@ public class DrivingQueryItemReaderTests extends TestCase { */ public void testRestart() throws Exception { + ExecutionContext executionContext = new ExecutionContext(); + + getAsItemStream(itemReader).open(executionContext); + Foo foo1 = (Foo) itemReader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) itemReader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext(); + getAsItemStream(itemReader).beforeSave(); // create new input source itemReader = createItemReader(); - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(executionContext); Foo fooAfterRestart = (Foo) itemReader.read(); assertEquals(3, fooAfterRestart.getValue()); @@ -89,13 +94,17 @@ public class DrivingQueryItemReaderTests extends TestCase { */ public void testInvalidRestore() throws Exception { + ExecutionContext executionContext = new ExecutionContext(); + + getAsItemStream(itemReader).open(executionContext); + Foo foo1 = (Foo) itemReader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) itemReader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext(); + getAsItemStream(itemReader).beforeSave(); // create new input source itemReader = createItemReader(); @@ -104,7 +113,7 @@ public class DrivingQueryItemReaderTests extends TestCase { assertEquals(1, foo.getValue()); try { - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(executionContext); fail(); } catch (IllegalStateException ex) { @@ -119,7 +128,7 @@ public class DrivingQueryItemReaderTests extends TestCase { public void testRestoreFromEmptyData() throws Exception { ExecutionContext streamContext = new ExecutionContext(new Properties()); - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(streamContext); Foo foo = (Foo) itemReader.read(); assertEquals(1, foo.getValue()); @@ -158,7 +167,7 @@ public class DrivingQueryItemReaderTests extends TestCase { return (InitializingBean) source; } - private ItemStream getAsRestartable(ItemReader source) { + private ItemStream getAsItemStream(ItemReader source) { return (ItemStream) source; } @@ -167,6 +176,7 @@ public class DrivingQueryItemReaderTests extends TestCase { static ExecutionContext streamContext; List keys; List restartKeys; + static final String RESTART_KEY = "restart.keys"; static{ Properties props = new Properties(); @@ -191,18 +201,21 @@ public class DrivingQueryItemReaderTests extends TestCase { restartKeys.add(new Foo(5, "5", 5)); } - public ExecutionContext getKeyAsExecutionContext(Object key) { + public ExecutionContext saveState(Object key) { return streamContext; } - public List restoreKeys(ExecutionContext streamContext) { - - assertEquals(MockKeyGenerator.streamContext, streamContext); - return restartKeys; + public List retrieveKeys(ExecutionContext executionContext) { + if(executionContext.containsKey(RESTART_KEY)){ + return restartKeys; + } + else{ + return keys; + } } - public List retrieveKeys() { - return keys; + public void saveState(Object key, ExecutionContext executionContext) { + executionContext.put(RESTART_KEY, restartKeys); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java index de86c52cf..94a70316f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java @@ -29,12 +29,8 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader } } - public ExecutionContext getExecutionContext() { - return inputSource.getExecutionContext(); - } - - public void restoreFrom(ExecutionContext data) { - inputSource.restoreFrom(data); + public void beforeSave() { + inputSource.beforeSave(); } public void destroy() throws Exception { @@ -48,7 +44,8 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader public void afterPropertiesSet() throws Exception { } - public void open() { + public void open(ExecutionContext executionContext) { + inputSource.open(executionContext); }; public void close() { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/IbatisItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/IbatisItemReaderIntegrationTests.java index f016ee1db..45a857e0f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/IbatisItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/IbatisItemReaderIntegrationTests.java @@ -31,6 +31,7 @@ public class IbatisItemReaderIntegrationTests extends AbstractDataSourceItemRead keyGenerator.setSqlMapClient(sqlMapClient); inputSource.setSqlMapClient(sqlMapClient); inputSource.setKeyGenerator(keyGenerator); + inputSource.setSaveState(true); return inputSource; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryItemReaderIntegrationTests.java index dee5e8724..5074529f3 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryItemReaderIntegrationTests.java @@ -34,6 +34,7 @@ public class MultipleColumnJdbcDrivingQueryItemReaderIntegrationTests extends keyGenerator.setRestartSql("SELECT ID, VALUE from T_FOOS where ID > ? and VALUE > ? order by ID"); DrivingQueryItemReader inputSource = new DrivingQueryItemReader(); + inputSource.setSaveState(true); inputSource.setKeyGenerator(keyGenerator); FooItemReader fooItemReader = new FooItemReader(inputSource, getJdbcTemplate()); fooItemReader.setFooDao(new CompositeKeyFooDao(getJdbcTemplate())); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryItemReaderIntegrationTests.java index 5f0d9d3a4..de47c1dcf 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryItemReaderIntegrationTests.java @@ -19,7 +19,7 @@ public class SingleColumnJdbcDrivingQueryItemReaderIntegrationTests extends Abst keyStrategy.setRestartSql("SELECT ID from T_FOOS where ID > ? order by ID"); DrivingQueryItemReader inputSource = new DrivingQueryItemReader(); inputSource.setKeyGenerator(keyStrategy); - + inputSource.setSaveState(true); return new FooItemReader(inputSource, getJdbcTemplate()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapperTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapperTests.java index ebbc2208c..42942c469 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapperTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapperTests.java @@ -20,8 +20,6 @@ import org.springframework.jdbc.core.PreparedStatementSetter; */ public class ColumnMapExecutionContextRowMapperTests extends TestCase { - private static final String KEY = ColumnMapExecutionContextRowMapper.KEY_PREFIX; - private ColumnMapExecutionContextRowMapper mapper; private Map key; @@ -29,6 +27,8 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase { private MockControl psControl = MockControl.createControl(PreparedStatement.class); private PreparedStatement ps; + private ExecutionContext executionContext; + protected void setUp() throws Exception { super.setUp(); @@ -37,12 +37,14 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase { key = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(2); key.put("1", new Integer(1)); key.put("2", new Integer(2)); + + executionContext = new ExecutionContext(); } public void testCreateExecutionContextWithInvalidType() throws Exception { try{ - mapper.createExecutionContext(new Object()); + mapper.mapKeys(new Object(), executionContext); fail(); }catch(IllegalArgumentException ex){ //expected @@ -52,7 +54,7 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase { public void testCreateExecutionContextWithNull(){ try{ - mapper.createExecutionContext(null); + mapper.mapKeys(null, null); fail(); }catch(IllegalArgumentException ex){ //expected @@ -60,31 +62,28 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase { } public void testCreateExecutionContext() throws Exception { - ExecutionContext streamContext = mapper.createExecutionContext(key); - Properties props = streamContext.getProperties(); - assertEquals("1", props.getProperty(KEY + "0")); - assertEquals("2", props.getProperty(KEY + "1")); + mapper.mapKeys(key, executionContext); + Properties props = executionContext.getProperties(); + assertEquals("1", props.getProperty("1")); + assertEquals("2", props.getProperty("2")); } public void testCreateExecutionContextFromEmptyKeys() throws Exception { - ExecutionContext streamContext = mapper.createExecutionContext(new HashMap()); - assertEquals(0, streamContext.getProperties().size()); + mapper.mapKeys(new HashMap(), executionContext); + assertEquals(0, executionContext.size()); } public void testCreateSetter() throws Exception { - Properties props = new Properties(); - props.setProperty(KEY + "0", "1"); - props.setProperty(KEY + "1", "2"); ExecutionContext streamContext = new ExecutionContext(); - streamContext.putString(KEY + "0", "1"); - streamContext.putString(KEY + "1", "2"); + streamContext.putString("0", "1"); + streamContext.putString("1", "2"); PreparedStatementSetter setter = mapper.createSetter(streamContext); ps = (PreparedStatement)psControl.getMock(); - ps.setString(1, "1"); - ps.setString(2, "2"); + ps.setString(1, "2"); + ps.setString(2, "1"); psControl.replay(); setter.setValues(ps); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java index 8997b2f42..768460e0e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java @@ -19,6 +19,8 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran MultipleColumnJdbcKeyGenerator keyStrategy; + ExecutionContext executionContext; + protected String[] getConfigLocations(){ return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; } @@ -30,11 +32,13 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran "SELECT ID, VALUE from T_FOOS order by ID"); keyStrategy.setRestartSql("SELECT ID, VALUE from T_FOOS where ID > ? and VALUE > ? order by ID"); + + executionContext = new ExecutionContext(); } public void testRetrieveKeys(){ - List keys = keyStrategy.retrieveKeys(); + List keys = keyStrategy.retrieveKeys(executionContext); for (int i = 0; i < keys.size(); i++) { Map id = (Map)keys.get(i); @@ -45,11 +49,10 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran public void testRestoreKeys(){ - ExecutionContext streamContext = new ExecutionContext(); - streamContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "0", "3"); - streamContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "1", "3"); + executionContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "0", "3"); + executionContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "1", "3"); - List keys = keyStrategy.restoreKeys(streamContext); + List keys = keyStrategy.retrieveKeys(executionContext); assertEquals(2, keys.size()); Map key = (Map)keys.get(0); @@ -66,8 +69,8 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran key.put("ID", new Long(3)); key.put("VALUE", new Integer(3)); - ExecutionContext streamContext = keyStrategy.getKeyAsExecutionContext(key); - Properties props = streamContext.getProperties(); + keyStrategy.saveState(key, executionContext); + Properties props = executionContext.getProperties(); assertEquals(2, props.size()); assertEquals("3", props.get(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "0")); @@ -77,19 +80,10 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran public void testGetNullKeyAsStreamContext(){ try{ - keyStrategy.getKeyAsExecutionContext(null); + keyStrategy.saveState(null, null); fail(); }catch(IllegalArgumentException ex){ //expected } } - - public void testRestoreKeysFromNull(){ - - try{ - keyStrategy.getKeyAsExecutionContext(null); - }catch(IllegalArgumentException ex){ - //expected - } - } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java index 899b9c31b..1fca690c1 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java @@ -1,7 +1,6 @@ package org.springframework.batch.io.driving.support; import java.util.List; -import java.util.Properties; import org.springframework.batch.item.ExecutionContext; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -15,6 +14,8 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa SingleColumnJdbcKeyGenerator keyStrategy; + ExecutionContext executionContext; + protected String[] getConfigLocations(){ return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; } @@ -27,11 +28,13 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa "SELECT ID from T_FOOS order by ID"); keyStrategy.setRestartSql("SELECT ID from T_FOOS where ID > ? order by ID"); + + executionContext = new ExecutionContext(); } public void testRetrieveKeys(){ - List keys = keyStrategy.retrieveKeys(); + List keys = keyStrategy.retrieveKeys(new ExecutionContext()); for (int i = 0; i < keys.size(); i++) { Long id = (Long)keys.get(i); @@ -41,11 +44,9 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa public void testRestoreKeys(){ - Properties props = new Properties(); - props.setProperty(SingleColumnJdbcKeyGenerator.RESTART_KEY, "3"); - ExecutionContext streamContext = new ExecutionContext(props); + executionContext.putString(SingleColumnJdbcKeyGenerator.RESTART_KEY, "3"); - List keys = keyStrategy.restoreKeys(streamContext); + List keys = keyStrategy.retrieveKeys(executionContext); assertEquals(2, keys.size()); assertEquals(new Long(4), keys.get(0)); @@ -54,17 +55,16 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa public void testGetKeyAsStreamContext(){ - ExecutionContext streamContext = keyStrategy.getKeyAsExecutionContext(new Long(3)); - Properties props = streamContext.getProperties(); + keyStrategy.saveState(new Long(3), executionContext); - assertEquals(1, props.size()); - assertEquals("3", props.get(SingleColumnJdbcKeyGenerator.RESTART_KEY)); + assertEquals(1, executionContext.size()); + assertEquals("3", executionContext.getString(SingleColumnJdbcKeyGenerator.RESTART_KEY)); } public void testGetNullKeyAsStreamContext(){ try{ - keyStrategy.getKeyAsExecutionContext(null); + keyStrategy.saveState(null, null); fail(); }catch(IllegalArgumentException ex){ //expected @@ -74,7 +74,7 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa public void testRestoreKeysFromNull(){ try{ - keyStrategy.getKeyAsExecutionContext(null); + keyStrategy.saveState(null, null); }catch(IllegalArgumentException ex){ //expected } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java index af1738d96..3a45561f6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java @@ -25,7 +25,6 @@ import org.springframework.batch.io.file.mapping.FieldSet; import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.exception.StreamException; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; @@ -41,6 +40,8 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { // common value used for writing to a file private String TEST_STRING = "FlatFileInputTemplate-TestData"; + + private ExecutionContext executionContext; // simple stub instead of a realistic tokenizer private LineTokenizer tokenizer = new LineTokenizer() { @@ -66,7 +67,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.setFieldSetMapper(fieldSetMapper); // context argument is necessary only for the FileLocator, which // is mocked - reader.open(); + executionContext = new ExecutionContext(); } /** @@ -88,7 +89,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.close(); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - reader.open(); + reader.open(executionContext); // read some records reader.read(); // #1 @@ -123,7 +124,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.close(); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - reader.open(); + reader.open(executionContext); // read some records reader.read(); // #1 @@ -142,30 +143,11 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { } - public void testRestartFromNullData() throws Exception { - reader.restoreFrom(null); - assertEquals("[FlatFileInputTemplate-TestData]", reader.read().toString()); - } - - public void testRestartBeforeOpen() throws Exception { - reader = new FlatFileItemReader(); - reader.setResource(getInputResource(TEST_STRING)); - reader.setFieldSetMapper(fieldSetMapper); - // do not open the template... - try { - reader.restoreFrom(reader.getExecutionContext()); - } catch (StreamException e) { - assertTrue("Message does not contain open: "+e.getMessage(), e.getMessage().contains("open")); - } - reader.open(); - assertEquals("[FlatFileInputTemplate-TestData]", reader.read().toString()); - } - public void testRestart() throws Exception { reader.close(); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - reader.open(); + reader.open(executionContext); // read some records reader.read(); @@ -177,24 +159,23 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.read(); // get restart data - ExecutionContext streamContext = reader.getExecutionContext(); - assertEquals("4", (String) streamContext.getProperties().getProperty( - FlatFileItemReader.READ_STATISTICS_NAME)); + reader.beforeSave(); + assertEquals(4, executionContext.getLong( + FlatFileItemReader.class.getName() + "." + FlatFileItemReader.READ_STATISTICS_NAME)); // close input reader.close(); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); // init for restart - reader.open(); - reader.restoreFrom(streamContext); + reader.open(executionContext); // read remaining records assertEquals("[testLine5]", reader.read().toString()); assertEquals("[testLine6]", reader.read().toString()); - ExecutionContext statistics = reader.getExecutionContext(); - assertEquals(6, statistics.getLong(FlatFileItemReader.READ_STATISTICS_NAME)); + reader.beforeSave(); + assertEquals(6, executionContext.getLong(FlatFileItemReader.class.getName() + "." + FlatFileItemReader.READ_STATISTICS_NAME)); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java index 31e3dcaa9..6555903f0 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java @@ -29,6 +29,7 @@ import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.separator.DefaultRecordSeparatorPolicy; import org.springframework.batch.io.file.transform.DelimitedLineTokenizer; import org.springframework.batch.io.file.transform.LineTokenizer; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.exception.StreamException; import org.springframework.core.io.AbstractResource; import org.springframework.core.io.ByteArrayResource; @@ -48,6 +49,8 @@ public class FlatFileItemReaderBasicTests extends TestCase { // common value used for writing to a file private String TEST_STRING = "FlatFileInputTemplate-TestData"; private String TEST_OUTPUT = "[FlatFileInputTemplate-TestData]"; + + private ExecutionContext executionContext; // simple stub instead of a realistic tokenizer private LineTokenizer tokenizer = new LineTokenizer() { @@ -72,8 +75,8 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setLineTokenizer(tokenizer); itemReader.setFieldSetMapper(fieldSetMapper); itemReader.afterPropertiesSet(); - - itemReader.open(); + + executionContext = new ExecutionContext(); } /** @@ -91,6 +94,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { * Regular usage of read method */ public void testRead() throws Exception { + itemReader.open(executionContext); assertEquals("[FlatFileInputTemplate-TestData]", itemReader.read().toString()); } @@ -98,6 +102,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { * Regular usage of read method */ public void testReadExhausted() throws Exception { + itemReader.open(executionContext); assertEquals("[FlatFileInputTemplate-TestData]", itemReader.read().toString()); assertEquals(null, itemReader.read()); } @@ -112,6 +117,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { } }); try { + itemReader.open(executionContext); itemReader.read(); fail("Expected ParsingException"); } catch (FlatFileParsingException e) { @@ -128,6 +134,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { }); try { + itemReader.open(executionContext); itemReader.read(); fail("Expected ParsingException"); } catch (FlatFileParsingException e) { @@ -154,7 +161,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setFieldSetMapper(fieldSetMapper); itemReader.close(); // The open does not happen automatically on a read... - itemReader.open(); + itemReader.open(executionContext); assertEquals("[FlatFileInputTemplate-TestData]", itemReader.read().toString()); } @@ -170,7 +177,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { } public void testOpenTwiceHasNoEffect() throws Exception { - itemReader.open(); + itemReader.open(executionContext); testRead(); } @@ -179,7 +186,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setEncoding("UTF-8"); itemReader.setResource(getInputResource(TEST_STRING)); itemReader.setFieldSetMapper(fieldSetMapper); - itemReader.open(); + itemReader.open(executionContext); testRead(); } @@ -188,7 +195,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setEncoding(null); itemReader.setResource(getInputResource(TEST_STRING)); try { - itemReader.open(); + itemReader.open(executionContext); fail("Expected IllegalArgumentException"); } catch (IllegalArgumentException e) { @@ -201,7 +208,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setEncoding("foo"); itemReader.setResource(getInputResource(TEST_STRING)); try { - itemReader.open(); + itemReader.open(executionContext); fail("Expected BatchEnvironmentException"); } catch (BatchEnvironmentException e) { @@ -238,7 +245,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setFieldSetMapper(fieldSetMapper); itemReader.setFirstLineIsHeader(true); itemReader.afterPropertiesSet(); - itemReader.open(); + itemReader.open(executionContext); FieldSet fs = (FieldSet) itemReader.read(); assertEquals("value1", fs.readString("name1")); @@ -261,7 +268,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setFieldSetMapper(fieldSetMapper); itemReader.setLinesToSkip(1); itemReader.afterPropertiesSet(); - itemReader.open(); + itemReader.open(executionContext); FieldSet fs = (FieldSet) itemReader.read(); assertEquals("one", fs.readString(0)); @@ -286,7 +293,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { testReader.afterPropertiesSet(); try{ - testReader.open(); + testReader.open(executionContext); fail(); }catch(IllegalStateException ex){ //expected @@ -309,7 +316,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { //replace the resource to simulate runtime resource creation testReader.setResource(getInputResource(TEST_STRING)); - testReader.open(); + testReader.open(executionContext); assertEquals(TEST_OUTPUT, testReader.read().toString()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java index 7f0fc0594..e26362344 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java @@ -28,6 +28,8 @@ import org.springframework.batch.io.file.mapping.FieldSet; import org.springframework.batch.io.file.mapping.FieldSetCreator; import org.springframework.batch.io.file.mapping.PassThroughFieldSetMapper; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; import org.springframework.core.io.FileSystemResource; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -53,6 +55,8 @@ public class FlatFileItemWriterTests extends TestCase { // reads the output file to check the result private BufferedReader reader; + + private ExecutionContext executionContext; /** * Create temporary output file, define mock behaviour, set dependencies and @@ -70,9 +74,7 @@ public class FlatFileItemWriterTests extends TestCase { inputSource.setResource(new FileSystemResource(outputFile)); inputSource.setFieldSetUnmapper(new PassThroughFieldSetMapper()); inputSource.afterPropertiesSet(); - - inputSource.open(); - + executionContext = new ExecutionContext(); } /** @@ -105,6 +107,7 @@ public class FlatFileItemWriterTests extends TestCase { * @throws Exception */ public void testWriteString() throws Exception { + inputSource.open(executionContext); inputSource.write(TEST_STRING); inputSource.close(); String lineFromFile = readLine(); @@ -198,6 +201,7 @@ public class FlatFileItemWriterTests extends TestCase { public void testRestart() throws Exception { + inputSource.open(executionContext); // write some lines inputSource.write("testLine1"); inputSource.write("testLine2"); @@ -220,25 +224,12 @@ public class FlatFileItemWriterTests extends TestCase { commit(); // get restart data - ExecutionContext streamContext = inputSource.getExecutionContext(); + inputSource.beforeSave(); // close template inputSource.close(); - // init for restart - inputSource.setBufferSize(0); - inputSource.open(); - - // try empty restart data... - try { - inputSource.restoreFrom(null); - assertTrue(true); - } - catch (IllegalArgumentException iae) { - fail("null restart data should be handled gracefully"); - } - // init with correct data - inputSource.restoreFrom(streamContext); + inputSource.open(executionContext); // write more lines inputSource.write("testLine6"); @@ -246,7 +237,7 @@ public class FlatFileItemWriterTests extends TestCase { inputSource.write("testLine8"); // get statistics - ExecutionContext statistics = inputSource.getExecutionContext(); + inputSource.beforeSave(); // close template inputSource.close(); @@ -256,7 +247,7 @@ public class FlatFileItemWriterTests extends TestCase { } // 3 lines were written to the file after restart - assertEquals(3, statistics.getLong(FlatFileItemWriter.WRITTEN_STATISTICS_NAME)); + assertEquals(3, executionContext.getLong(FlatFileItemWriter.class.getName() + "." + FlatFileItemWriter.WRITTEN_STATISTICS_NAME)); } @@ -276,11 +267,11 @@ public class FlatFileItemWriterTests extends TestCase { inputSource.setResource(new FileSystemResource(outputFile)); inputSource.setFieldSetUnmapper(new PassThroughFieldSetMapper()); inputSource.afterPropertiesSet(); - inputSource.open(); - ExecutionContext streamContext = inputSource.getExecutionContext(); - assertNotNull(streamContext); - assertEquals(3, streamContext.getProperties().size()); - assertEquals(0, streamContext.getLong(FlatFileItemWriter.RESTART_DATA_NAME)); + inputSource.open(executionContext); + inputSource.beforeSave(); + assertNotNull(executionContext); + assertEquals(3, executionContext.entrySet().size()); + assertEquals(0, executionContext.getLong(FlatFileItemWriter.class.getName() + "." + FlatFileItemWriter.RESTART_DATA_NAME)); } private void commit() throws Exception { @@ -291,4 +282,7 @@ public class FlatFileItemWriterTests extends TestCase { inputSource.clear(); } + private ItemStream getAsItemStream(ItemWriter itemWriter){ + return (ItemStream)itemWriter; + } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java index e4d2f8697..43a31e5ee 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java @@ -1,9 +1,9 @@ package org.springframework.batch.io.sql; import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -19,7 +19,8 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra protected ItemReader itemReader; - + protected ExecutionContext executionContext; + /** * @return input source with all necessary dependencies set */ @@ -33,6 +34,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra super.onSetUp(); itemReader = createItemReader(); getAsInitializingBean(itemReader).afterPropertiesSet(); + executionContext = new ExecutionContext(); } protected void onTearDown()throws Exception { @@ -45,6 +47,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra */ public void testNormalProcessing() throws Exception { getAsInitializingBean(itemReader).afterPropertiesSet(); + getAsItemStream(itemReader).open(executionContext); Foo foo1 = (Foo) itemReader.read(); assertEquals(1, foo1.getValue()); @@ -69,19 +72,18 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra * @throws Exception */ public void testRestart() throws Exception { - + getAsItemStream(itemReader).open(executionContext); Foo foo1 = (Foo) itemReader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) itemReader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext(); + getAsItemStream(itemReader).beforeSave(); // create new input source itemReader = createItemReader(); - - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(executionContext); Foo fooAfterRestart = (Foo) itemReader.read(); assertEquals(3, fooAfterRestart.getValue()); @@ -92,13 +94,14 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra */ public void testInvalidRestore() throws Exception { + getAsItemStream(itemReader).open(executionContext); Foo foo1 = (Foo) itemReader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) itemReader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext(); + getAsItemStream(itemReader).beforeSave(); // create new input source itemReader = createItemReader(); @@ -107,7 +110,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra assertEquals(1, foo.getValue()); try { - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(executionContext); fail(); } catch (IllegalStateException ex) { @@ -122,7 +125,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra public void testRestoreFromEmptyData() throws Exception { ExecutionContext streamContext = new ExecutionContext(); - getAsRestartable(itemReader).restoreFrom(streamContext); + getAsItemStream(itemReader).open(executionContext); Foo foo = (Foo) itemReader.read(); assertEquals(1, foo.getValue()); @@ -157,7 +160,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra itemReader.reset(); } - private ItemStream getAsRestartable(ItemReader source) { + private ItemStream getAsItemStream(ItemReader source) { return (ItemStream) source; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java index f687781f6..74bc7781a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java @@ -20,6 +20,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { protected ItemReader reader; + protected ExecutionContext executionContext; /** * @return configured input source ready for use @@ -37,6 +38,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends protected void onSetUpInTransaction() throws Exception { super.onSetUpInTransaction(); reader = createItemReader(); + executionContext = new ExecutionContext(); } /* @@ -79,18 +81,20 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends */ public void testRestart() throws Exception { + getAsItemStream(reader).open(executionContext); + Foo foo1 = (Foo) reader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) reader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsItemStream(reader).getExecutionContext(); + getAsItemStream(reader).beforeSave(); // create new input source reader = createItemReader(); - getAsItemStream(reader).restoreFrom(streamContext); + getAsItemStream(reader).open(executionContext); Foo fooAfterRestart = (Foo) reader.read(); assertEquals(3, fooAfterRestart.getValue()); @@ -101,13 +105,15 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends */ public void testInvalidRestore() throws Exception { + getAsItemStream(reader).open(executionContext); + Foo foo1 = (Foo) reader.read(); assertEquals(1, foo1.getValue()); Foo foo2 = (Foo) reader.read(); assertEquals(2, foo2.getValue()); - ExecutionContext streamContext = getAsItemStream(reader).getExecutionContext(); + getAsItemStream(reader).beforeSave(); // create new input source reader = createItemReader(); @@ -116,7 +122,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends assertEquals(1, foo.getValue()); try { - getAsItemStream(reader).restoreFrom(streamContext); + getAsItemStream(reader).open(executionContext); fail(); } catch (IllegalStateException ex) { @@ -129,9 +135,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends * @throws Exception */ public void testRestoreFromEmptyData() throws Exception { - ExecutionContext streamContext = new ExecutionContext(); - - getAsItemStream(reader).restoreFrom(streamContext); + getAsItemStream(reader).beforeSave(); Foo foo = (Foo) reader.read(); assertEquals(1, foo.getValue()); @@ -198,6 +202,8 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends return; } + getAsItemStream(reader).open(executionContext); + Foo foo1 = (Foo) reader.read(); commit(); @@ -212,12 +218,12 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends rollback(); - ExecutionContext streamContext = getAsItemStream(reader).getExecutionContext(); + getAsItemStream(reader).beforeSave(); // create new input source reader = createItemReader(); - getAsItemStream(reader).restoreFrom(streamContext); + getAsItemStream(reader).open(executionContext); assertEquals(foo2, reader.read()); Foo foo4 = (Foo) reader.read(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java index 1f0d127a4..cb9eb1fd3 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java @@ -37,8 +37,11 @@ public class StaxEventItemReaderTests extends TestCase { private EventReaderDeserializer deserializer = new MockFragmentDeserializer(); private static final String FRAGMENT_ROOT_ELEMENT = "fragment"; + + private ExecutionContext executionContext; protected void setUp() throws Exception { + this.executionContext = new ExecutionContext(); source = createNewInputSouce(); } @@ -83,7 +86,7 @@ public class StaxEventItemReaderTests extends TestCase { */ public void testFragmentWrapping() throws Exception { source.afterPropertiesSet(); - + source.open(executionContext); // see asserts in the mock deserializer assertNotNull(source.read()); assertNotNull(source.read()); @@ -112,13 +115,14 @@ public class StaxEventItemReaderTests extends TestCase { * Save restart data and restore from it. */ public void testRestart() { + source.open(executionContext); source.read(); - ExecutionContext streamContext = source.getExecutionContext(); - assertEquals(1, streamContext.getLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME)); + source.beforeSave(); + assertEquals(1, executionContext.getLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME)); List expectedAfterRestart = (List) source.read(); source = createNewInputSouce(); - source.restoreFrom(streamContext); + source.open(executionContext); List afterRestart = (List) source.read(); assertEquals(expectedAfterRestart.size(), afterRestart.size()); } @@ -131,7 +135,7 @@ public class StaxEventItemReaderTests extends TestCase { ExecutionContext context = new ExecutionContext(); context.putLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME, 100000); try { - source.restoreFrom(context); + source.open(context); fail("Expected StreamException"); } catch (StreamException e) { @@ -143,12 +147,13 @@ public class StaxEventItemReaderTests extends TestCase { public void testRestoreWorksFromClosedStream() throws Exception { source.close(); - source.restoreFrom(new ExecutionContext()); + source.beforeSave(); } /** * Skipping marked records after rollback. */ public void testSkip() { + source.open(executionContext); List first = (List) source.read(); source.skip(); List second = (List) source.read(); @@ -162,7 +167,7 @@ public class StaxEventItemReaderTests extends TestCase { * Rollback to last commited record. */ public void testRollback() { - + source.open(executionContext); // rollback between deserializing records List first = (List) source.read(); source.mark(); @@ -190,13 +195,16 @@ public class StaxEventItemReaderTests extends TestCase { * Statistics return the current record count. Calling read after end of * input does not increase the counter. */ - public void testStreamContext() { + public void testExecutionContext() { final int NUMBER_OF_RECORDS = 2; - + source.open(executionContext); + source.beforeSave(); + for (int i = 0; i < NUMBER_OF_RECORDS; i++) { long recordCount = extractRecordCount(); assertEquals(i, recordCount); source.read(); + source.beforeSave(); } assertEquals(NUMBER_OF_RECORDS, extractRecordCount()); @@ -205,7 +213,7 @@ public class StaxEventItemReaderTests extends TestCase { } private long extractRecordCount() { - return source.getExecutionContext().getLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME); + return executionContext.getLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME); } public void testCloseWithoutOpen() throws Exception { @@ -221,7 +229,7 @@ public class StaxEventItemReaderTests extends TestCase { newSource.setFragmentRootElementName(FRAGMENT_ROOT_ELEMENT); newSource.setFragmentDeserializer(deserializer); - newSource.open(); + newSource.open(executionContext); Object item = newSource.read(); assertNotNull(item); @@ -256,7 +264,7 @@ public class StaxEventItemReaderTests extends TestCase { }); try { - source.open(); + source.open(executionContext); } catch (DataAccessResourceFailureException ex) { assertTrue(ex.getCause() instanceof IOException); @@ -270,7 +278,7 @@ public class StaxEventItemReaderTests extends TestCase { source.afterPropertiesSet(); try { - source.open(); + source.open(executionContext); fail(); } catch (IllegalStateException ex) { @@ -284,6 +292,7 @@ public class StaxEventItemReaderTests extends TestCase { source.afterPropertiesSet(); source.setResource(new ByteArrayResource(xml.getBytes())); + source.open(executionContext); source.read(); } @@ -296,8 +305,6 @@ public class StaxEventItemReaderTests extends TestCase { newSource.setFragmentRootElementName(FRAGMENT_ROOT_ELEMENT); newSource.setFragmentDeserializer(deserializer); - newSource.open(); - return newSource; } @@ -381,8 +388,8 @@ public class StaxEventItemReaderTests extends TestCase { private boolean openCalled = false; - public void open() { - super.open(); + public void open(ExecutionContext executionContext) { + super.open(executionContext); openCalled = true; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java index 780fc69ea..086dfcbbe 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java @@ -30,6 +30,8 @@ public class StaxEventItemWriterTests extends TestCase { // output file private Resource resource; + + private ExecutionContext executionContext; // test record for writing to output private Object record = new Object() { @@ -45,13 +47,14 @@ public class StaxEventItemWriterTests extends TestCase { protected void setUp() throws Exception { resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", "xml")); writer = createItemWriter(); - writer.open(); + executionContext = new ExecutionContext(); } /** * Write should pass its argument and StaxResult object to Serializer */ public void testWrite() throws Exception { + writer.open(executionContext); Marshaller marshaller = new InputCheckMarshaller(); MarshallingEventWriterSerializer serializer = new MarshallingEventWriterSerializer(marshaller); writer.setSerializer(serializer); @@ -65,6 +68,7 @@ public class StaxEventItemWriterTests extends TestCase { * Rolled back records should not be written to output file. */ public void testRollback() throws Exception { + writer.open(executionContext); writer.write(record); // rollback writer.clear(); @@ -75,6 +79,7 @@ public class StaxEventItemWriterTests extends TestCase { * Commited output is written to the output file. */ public void testCommit() throws Exception { + writer.open(executionContext); writer.write(record); // commit writer.flush(); @@ -85,15 +90,16 @@ public class StaxEventItemWriterTests extends TestCase { * Restart scenario - content is appended to the output file after restart. */ public void testRestart() throws Exception { + writer.open(executionContext); // write record writer.write(record); // writer.mark(); - ExecutionContext streamContext = writer.getExecutionContext(); + writer.beforeSave(); writer.close(); // create new writer from saved restart data and continue writing writer = createItemWriter(); - writer.restoreFrom(streamContext); + writer.open(executionContext); writer.write(record); writer.close(); @@ -114,10 +120,12 @@ public class StaxEventItemWriterTests extends TestCase { * Count of 'records written so far' is returned as statistics. */ public void testStreamContext() throws Exception { + writer.open(executionContext); final int NUMBER_OF_RECORDS = 10; for (int i = 1; i <= NUMBER_OF_RECORDS; i++) { writer.write(record); - long writeStatistics = writer.getExecutionContext().getLong(StaxEventItemWriter.WRITE_STATISTICS_NAME); + writer.beforeSave(); + long writeStatistics = executionContext.getLong(StaxEventItemWriter.WRITE_STATISTICS_NAME); assertEquals(i, writeStatistics); } @@ -133,7 +141,7 @@ public class StaxEventItemWriterTests extends TestCase { put("attribute", "value"); } }); - writer.open(); + writer.open(executionContext); writer.flush(); assertTrue(outputFileContent().indexOf("