From 64506040ff2297f4cac93a85e1abd51a5bb461b0 Mon Sep 17 00:00:00 2001 From: lucasward Date: Tue, 26 Feb 2008 08:29:37 +0000 Subject: [PATCH] BATCH-365: There should now be one ExecutionContext per step. All itemStreams will be opened with an execution context, and will be notified before it is saved, to ensure they have all state in the context. Most ItemReader/Writers should now have the logic for whether or not to put their state in the context, but a few have likely been missed. --- .../io/cursor/HibernateCursorItemReader.java | 66 ++++---- .../batch/io/cursor/JdbcCursorItemReader.java | 65 ++++---- .../io/driving/DrivingQueryItemReader.java | 57 +++---- .../batch/io/driving/KeyGenerator.java | 14 +- .../ColumnMapExecutionContextRowMapper.java | 55 +------ .../support/ExecutionContextRowMapper.java | 2 +- .../driving/support/IbatisKeyGenerator.java | 43 ++--- .../MultipleColumnJdbcKeyGenerator.java | 23 +-- .../support/SingleColumnJdbcKeyGenerator.java | 51 ++---- .../batch/io/file/FlatFileItemReader.java | 63 ++++---- .../batch/io/file/FlatFileItemWriter.java | 47 +++--- .../batch/io/xml/StaxEventItemReader.java | 48 +++--- .../batch/io/xml/StaxEventItemWriter.java | 40 +++-- .../batch/item/ExecutionContext.java | 4 + .../batch/item/ExecutionContextProvider.java | 32 ---- .../batch/item/ItemReader.java | 2 +- .../batch/item/ItemStream.java | 32 ++-- .../item/reader/DelegatingItemReader.java | 22 +-- .../batch/item/stream/ItemStreamSupport.java | 33 +--- .../item/stream/SimpleStreamManager.java | 138 +++------------- .../batch/item/stream/StreamManager.java | 18 +-- .../item/writer/DelegatingItemWriter.java | 22 +-- ...rnateCursorItemReaderIntegrationTests.java | 3 +- .../JdbcCursorItemReaderIntegrationTests.java | 1 + .../driving/DrivingQueryItemReaderTests.java | 39 +++-- .../batch/io/driving/FooInputSource.java | 11 +- .../IbatisItemReaderIntegrationTests.java | 1 + ...rivingQueryItemReaderIntegrationTests.java | 1 + ...rivingQueryItemReaderIntegrationTests.java | 2 +- ...lumnMapExecutionContextRowMapperTests.java | 33 ++-- ...olumnJdbcKeyGeneratorIntegrationTests.java | 28 ++-- ...olumnJdbcKeyGeneratorIntegrationTests.java | 24 +-- .../file/FlatFileItemReaderAdvancedTests.java | 43 ++--- .../io/file/FlatFileItemReaderBasicTests.java | 29 ++-- .../io/file/FlatFileItemWriterTests.java | 44 +++--- ...bstractJdbcItemReaderIntegrationTests.java | 23 +-- ...tDataSourceItemReaderIntegrationTests.java | 24 +-- .../io/xml/StaxEventItemReaderTests.java | 41 +++-- .../io/xml/StaxEventItemWriterTests.java | 18 ++- .../reader/DelegatingItemReaderTests.java | 32 ++-- .../item/stream/SimpleStreamManagerTests.java | 149 ++---------------- .../support/AbstractTradeBatchTests.java | 3 +- 42 files changed, 512 insertions(+), 914 deletions(-) delete mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextProvider.java 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("