diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
index ef3150c18..701590ec9 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
@@ -98,9 +98,9 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
private RetryTemplate template = new RetryTemplate();
private ItemReaderRetryCallback retryCallback;
-
+
private int commitInterval = 0;
-
+
private boolean saveExecutionContext = false;
/**
@@ -154,7 +154,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
-
+
public void setCommitInterval(int commitInterval) {
this.commitInterval = commitInterval;
}
@@ -176,20 +176,20 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
"ItemReader must be instance of KeyedItemReader to use the retry policy");
retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter);
}
-
- if(streamManager == null && transactionManager != null){
+
+ if (streamManager == null && transactionManager != null) {
streamManager = new SimpleStreamManager(transactionManager);
}
- else if(streamManager == null && transactionManager == null){
+ else if (streamManager == null && transactionManager == null) {
throw new IllegalArgumentException("Either StreamManager or TransactionManager must be set");
}
-
- if(commitInterval > 0){
- ((RepeatTemplate)chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval));
+
+ if (commitInterval > 0) {
+ ((RepeatTemplate) chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval));
}
-
- if(exceptionHandler != null){
- ((RepeatTemplate)chunkOperations).setExceptionHandler(exceptionHandler);
+
+ if (exceptionHandler != null) {
+ ((RepeatTemplate) chunkOperations).setExceptionHandler(exceptionHandler);
}
}
@@ -256,17 +256,19 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
// the conversation in StepScope
stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId());
- streamManager.open(stepExecution.getExecutionContext());
-
if (saveExecutionContext && isRestart && lastStepExecution != null) {
stepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
}
- else{
+ else {
stepExecution.setExecutionContext(new ExecutionContext());
}
+ // Open the stream manager *after* the execution context is fixed in
+ // the step, otherwise it will not be the same reference that is
+ // updated by the streams. TODO: this is a little fragile - maybe
+ // StreamManager.update() should accept the context as a parameter.
streamManager.open(stepExecution.getExecutionContext());
-
+
status = stepOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
@@ -278,7 +280,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
interruptionPolicy.checkInterrupted(context);
ExitStatus result;
-
+
TransactionStatus transaction = streamManager.getTransaction();
try {
@@ -296,7 +298,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
// only if chunk was successful
stepExecution.apply(contribution);
- streamManager.beforeSave();
+ streamManager.update();
jobRepository.saveOrUpdate(stepExecution);
}
@@ -456,10 +458,10 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
}
catch (Exception e) {
- if(getItemSkipPolicy().shouldSkip(e, contribution)){
+ if (getItemSkipPolicy().shouldSkip(e, contribution)) {
skip();
}
- else{
+ else {
// Rethrow so that outer transaction is rolled back properly
throw e;
}
@@ -485,11 +487,11 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
private ExitStatus execute() throws Exception {
if (retryCallback == null) {
- Object item = null;
- try{
+ Object item = null;
+ try {
item = itemReader.read();
}
- catch(Exception ex){
+ catch (Exception ex) {
getItemFailureHandler().handleReadFailure(ex);
throw ex;
}
@@ -500,7 +502,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
itemWriter.write(item);
}
catch (Exception e) {
-
+
getItemFailureHandler().handleWriteFailure(item, e);
// Re-throw the exception so that the surrounding transaction
// rolls back if there is one
@@ -534,7 +536,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
((Skippable) this.itemWriter).skip();
}
}
-
/**
* Convenience method to update the status in all relevant places.
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
index 9a56e4bd3..b3aff87ee 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
@@ -18,9 +18,7 @@ package org.springframework.batch.execution.step;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import junit.framework.TestCase;
@@ -410,10 +408,9 @@ public class ItemOrientedStepTests extends TestCase {
assertEquals(false, stepExecution.getExecutionContext().containsKey("foo"));
- final Map map = new HashMap();
itemOrientedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) {
ExecutionContext executionContext;
- public void beforeSave() {
+ public void update() {
// TODO Auto-generated method stub
executionContext.putString("foo", "bar");
}
@@ -429,8 +426,6 @@ public class ItemOrientedStepTests extends TestCase {
// At least once in that process the statistics service was asked for
// statistics...
assertEquals("bar", stepExecution.getExecutionContext().getString("foo"));
- // ...but nothing was registered because nothing with step scoped.
- assertEquals(0, map.size());
}
private class MockRestartableItemReader extends ItemStreamSupport implements ItemReader {
@@ -452,7 +447,7 @@ public class ItemOrientedStepTests extends TestCase {
return restoreFromCalledWithSomeContext;
}
- public void beforeSave() {
+ public void update() {
getExecutionAttributesCalled = true;
executionContext.putString("spam", "bucket");
}
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 34aae62f5..90f1e85b7 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
@@ -187,7 +187,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
*/
- public void beforeSave() {
+ public void update() {
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
String skipped = skippedRows.toString();
executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
@@ -210,15 +210,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * (non-Javadoc)
- * @see org.springframework.batch.item.stream.ItemStreamAdapter#mark(org.springframework.batch.item.ExecutionContext)
+ * @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
lastCommitRowNumber = currentProcessedRow;
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 ea1cff423..1fe35b02d 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
@@ -385,7 +385,7 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#getExecutionContext()
*/
- public void beforeSave() {
+ public void update() {
if(saveState){
String skipped = skippedRows.toString();
executionContext.putString(addName(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
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 aaf199d40..466fbfe6f 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
@@ -146,7 +146,7 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
this.executionContext = executionContext;
}
- public void beforeSave() {
+ public void update() {
if(saveState){
if(getCurrentKey() != null){
keyGenerator.saveState(getCurrentKey(), executionContext);
@@ -190,16 +190,7 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext)
+ * @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
lastCommitIndex = currentIndex;
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 d27e0601f..fc186ab8c 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
@@ -38,10 +38,11 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
public PreparedStatementSetter createSetter(ExecutionContext executionContext) {
List columns = new ArrayList();
- for (Iterator iterator = executionContext.entrySet().iterator(); iterator.hasNext();) {
- Entry entry = (Entry) iterator.next();
- Object column = entry.getValue();
+ int count=0;
+ while(executionContext.containsKey(KEY_PREFIX+count)) {
+ Object column = executionContext.get(KEY_PREFIX+count);
columns.add(column);
+ count++;
}
return new ArgPreparedStatementSetter(columns.toArray());
@@ -50,9 +51,11 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
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;
+ int count = 0;
for (Iterator it = keys.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry)it.next();
- executionContext.put(entry.getKey().toString(), entry.getValue());
+ executionContext.put(KEY_PREFIX+count, entry.getValue());
+ count++;
}
}
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 482b2bc0a..a1bfd9550 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
@@ -90,7 +90,7 @@ public class MultipleColumnJdbcKeyGenerator implements
* @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext(java.lang.Object)
*/
public void saveState(Object key, ExecutionContext executionContext) {
- Assert.state(keyMapper != null, "Kye mapper must not be null.");
+ Assert.state(keyMapper != null, "Key mapper must not be null.");
keyMapper.mapKeys(key, executionContext);
}
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 7a92021f1..990eaacea 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
@@ -202,7 +202,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* the current Line Count which can be used to reinitialise the batch job in
* case of restart.
*/
- public void beforeSave() {
+ public void update() {
if (reader == null) {
throw new StreamException("ItemStream not open or already closed.");
}
@@ -216,15 +216,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * (non-Javadoc)
- * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
+ * @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
getReader().mark();
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 20faa0adc..2e80dea9c 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
@@ -190,9 +190,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
- * @see ItemStream#beforeSave()
+ * @see ItemStream#update()
*/
- public void beforeSave() {
+ public void update() {
if (state == null) {
throw new StreamException("ItemStream not open or already closed.");
}
@@ -471,34 +471,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
- /**
- * Mark is supported as long as this {@link ItemStream} is used in a
- * single-threaded environment. The state backing the mark is a single
- * counter, keeping track of the current position, so multiple threads
- * cannot be accommodated.
- *
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * To be deleted once interface changes are complete (non-Javadoc)
- * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext)
- */
- public void mark() {
-
- }
-
- /*
- * To be deleted once interface changes are complete (non-Javadoc)
- * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext)
- */
- public void reset() throws ResetFailedException {
-
- }
-
public void clear() throws Exception {
try {
getOutputState().reset();
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java
index 93ab15bda..467a10c88 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java
@@ -195,22 +195,15 @@ public class ResourceLineReader extends ItemStreamSupport implements LineReader,
}
/**
+ * Mark the state for return later with reset. Uses the read-ahead limit
+ * from an underlying {@link BufferedReader}, which means that there is a
+ * limit to how much data can be recovered if the mark needs to be reset.
+ *
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /**
- * Mark the state for return later with reset. Uses the read-ahead limit
- * from an underlying {@link BufferedReader}, which means that there is a
- * limit to how much data can be recovered if the mark needs to be reset.
- *
* @see #reset()
*
* @throws MarkFailedException if the mark could not be set.
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
index 05e465460..1224441e3 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
@@ -32,26 +32,5 @@ import org.springframework.batch.item.stream.ItemStreamSupport;
* @since 1.0
*/
public abstract class AbstractTransactionalIoSource extends ItemStreamSupport {
-
- /*
- * Called when a transaction has been committed.
- *
- * @see TransactionSynchronization#afterCompletion
- */
- public abstract void mark();
-
- /*
- * Called when a transaction has been rolled back.
- *
- * @see TransactionSynchronization#afterCompletion
- */
- public abstract void reset();
-
- /* (non-Javadoc)
- * @see org.springframework.batch.item.stream.ItemStreamAdapter#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
}
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 15ac9f287..0d8aa8ac0 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
@@ -193,9 +193,9 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
}
/**
- * @see ItemStream#beforeSave()
+ * @see ItemStream#update()
*/
- public void beforeSave() {
+ public void update() {
executionContext.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
}
@@ -254,16 +254,8 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
- *
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * (non-Javadoc)
- * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
+ *
+ * @see org.springframework.batch.item.reader.AbstractItemReader#mark()
*/
public void mark() {
lastCommitPointRecordCount = currentRecordCount;
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 3cf5e5dfd..d61fc86b4 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
@@ -363,9 +363,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Get the restart data.
- * @see org.springframework.batch.item.ItemStream#beforeSave()
+ * @see org.springframework.batch.item.ItemStream#update()
*/
- public void beforeSave() {
+ public void update() {
if (!initialized) {
throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context.");
}
@@ -421,38 +421,11 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
}
}
-
- /**
- * Mark is supported as long as this {@link ItemStream} is used in a
- * single-threaded environment. The state backing the mark is a single
- * counter, keeping track of the current position, so multiple threads
- * cannot be accommodated.
- *
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /* TODO remove once ItemStream interface is modified.
- * (non-Javadoc)
- * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
- */
- public void mark() {
- }
public void flush() throws Exception {
lastCommitPointPosition = getPosition();
lastCommitPointRecordCount = currentRecordCount;
}
-
- /* TODO remove once ItemStream interface is modified.
- * (non-Javadoc)
- * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
- */
- public void reset() {
-
- }
public void clear() throws Exception {
currentRecordCount = lastCommitPointRecordCount;
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 a4acebcbe..82d6d115d 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
@@ -52,22 +52,19 @@ public interface ItemReader {
* @throws Exception if an underlying resource is unavailable.
*/
Object read() throws Exception;
-
+
/**
- * 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#beforeSave()}.
+ * Mark the stream so that it can be reset later and the items backed out.
*
* In a multi-threaded setting implementations have to ensure that only the
* state from the current thread is saved.
*
- * @throws UnsupportedOperationException if the operation is not supported
* @throws MarkFailedException if there is a problem with the mark. If a
* mark fails inside a transaction, it would be worrying, but not normally
* fatal.
*/
void mark() throws MarkFailedException;
-
+
/**
* Reset the stream to the last mark. After a reset the stream state will be
* such that changes (items read or written) since the last call to mark
@@ -76,7 +73,6 @@ public interface ItemReader {
* In a multi-threaded setting implementations have to ensure that only the
* state from the current thread is reset.
*
- * @throws UnsupportedOperationException if the operation is not supported
* @throws ResetFailedException if there is a problem with the reset. If a
* reset fails inside a transaction, it would normally be fatal, and would
* leave the stream in an inconsistent state. So while this is an unchecked
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 ac91d3ee7..88224573a 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
@@ -43,7 +43,7 @@ public interface ItemStream {
* has not been put in the context, it should be added
* here.
*/
- void beforeSave();
+ void update();
/**
* If any resources are needed for the stream to operate they need to be
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 37c448768..ac9dfb47b 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,13 +50,13 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
}
/**
- * @see ItemStream#beforeSave()
+ * @see ItemStream#update()
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
- public void beforeSave() {
+ public void update() {
if (itemReader instanceof ItemStream) {
- ((ItemStream) itemReader).beforeSave();
+ ((ItemStream) itemReader).update();
}
}
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 5be64c8b3..98487da8f 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
@@ -43,9 +43,9 @@ public class ItemStreamSupport implements ItemStream {
/**
* Return empty {@link ExecutionContext}.
- * @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
+ * @see org.springframework.batch.item.ExecutionContextProvider#update()
*/
- public void beforeSave() {
+ public void update() {
}
}
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 e2f778783..c32f57cb6 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
@@ -68,11 +68,11 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getExecutionContext(java.lang.Object)
*/
- public void beforeSave() {
- synchronized(streams){
- for(Iterator it = streams.iterator(); it.hasNext();){
- ItemStream itemStream = (ItemStream)it.next();
- itemStream.beforeSave();
+ public void update() {
+ synchronized (streams) {
+ for (Iterator it = streams.iterator(); it.hasNext();) {
+ ItemStream itemStream = (ItemStream) it.next();
+ itemStream.update();
}
}
}
@@ -86,7 +86,9 @@ public class SimpleStreamManager implements StreamManager {
*/
public void register(ItemStream stream) {
synchronized (streams) {
- streams.add(stream);
+ if (!streams.contains(stream)) {
+ streams.add(stream);
+ }
}
}
@@ -95,9 +97,9 @@ public class SimpleStreamManager implements StreamManager {
* @throws StreamException
*/
public void close() throws StreamException {
- synchronized(streams){
- for(Iterator it = streams.iterator(); it.hasNext();){
- ItemStream itemStream = (ItemStream)it.next();
+ synchronized (streams) {
+ for (Iterator it = streams.iterator(); it.hasNext();) {
+ ItemStream itemStream = (ItemStream) it.next();
itemStream.close();
}
}
@@ -108,9 +110,9 @@ public class SimpleStreamManager implements StreamManager {
* @throws StreamException
*/
public void open(ExecutionContext executionContext) throws StreamException {
- synchronized(streams){
- for(Iterator it = streams.iterator(); it.hasNext();){
- ItemStream itemStream = (ItemStream)it.next();
+ synchronized (streams) {
+ for (Iterator it = streams.iterator(); it.hasNext();) {
+ ItemStream itemStream = (ItemStream) it.next();
itemStream.open(executionContext);
}
}
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 e8dafd9fb..ba66d437e 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
@@ -49,7 +49,7 @@ public interface StreamManager {
* @return {@link ExecutionContext} aggregating the contexts of all providers
* registered under this key, or empty otherwise.
*/
- void beforeSave();
+ void update();
/**
* If any resources are needed for the stream to operate they need to be
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 48a836788..2525afcf0 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,10 +80,10 @@ public class DelegatingItemWriter implements ItemWriter, ItemStream, Initializin
}
/**
- * @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
+ * @see org.springframework.batch.item.ExecutionContextProvider#update()
*/
- public void beforeSave() {
- stream.beforeSave();
+ public void update() {
+ stream.update();
}
/**
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 0fb2a0e7e..c1536a1a4 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
@@ -78,7 +78,7 @@ public class DrivingQueryItemReaderTests extends TestCase {
Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(itemReader).beforeSave();
+ getAsItemStream(itemReader).update();
// create new input source
itemReader = createItemReader();
@@ -104,7 +104,7 @@ public class DrivingQueryItemReaderTests extends TestCase {
Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(itemReader).beforeSave();
+ getAsItemStream(itemReader).update();
// create new input source
itemReader = createItemReader();
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 94a70316f..cb9e835be 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,8 +29,8 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader
}
}
- public void beforeSave() {
- inputSource.beforeSave();
+ public void update() {
+ inputSource.update();
}
public void destroy() throws Exception {
@@ -51,14 +51,6 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader
public void close() {
}
- /**
- * True.
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
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 42942c469..c967979a2 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
@@ -64,8 +64,8 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase {
public void testCreateExecutionContext() throws Exception {
mapper.mapKeys(key, executionContext);
Properties props = executionContext.getProperties();
- assertEquals("1", props.getProperty("1"));
- assertEquals("2", props.getProperty("2"));
+ assertEquals("1", props.getProperty(ColumnMapExecutionContextRowMapper.KEY_PREFIX+"0"));
+ assertEquals("2", props.getProperty(ColumnMapExecutionContextRowMapper.KEY_PREFIX+"1"));
}
public void testCreateExecutionContextFromEmptyKeys() throws Exception {
@@ -77,8 +77,8 @@ public class ColumnMapExecutionContextRowMapperTests extends TestCase {
public void testCreateSetter() throws Exception {
ExecutionContext streamContext = new ExecutionContext();
- streamContext.putString("0", "1");
- streamContext.putString("1", "2");
+ streamContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX+"1", "1");
+ streamContext.putString(ColumnMapExecutionContextRowMapper.KEY_PREFIX+"0", "2");
PreparedStatementSetter setter = mapper.createSetter(streamContext);
ps = (PreparedStatement)psControl.getMock();
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 768460e0e..539cdb3d1 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
@@ -3,12 +3,17 @@
*/
package org.springframework.batch.io.driving.support;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+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.PreparedStatementSetter;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
/**
@@ -67,14 +72,31 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran
Map key = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(1);
key.put("ID", new Long(3));
- key.put("VALUE", new Integer(3));
+ key.put("VALUE", new Integer(4));
+ keyStrategy.setKeyMapper(new ExecutionContextRowMapper() {
+ public PreparedStatementSetter createSetter(ExecutionContext executionContext) {
+ return null;
+ }
+ public void mapKeys(Object key, ExecutionContext executionContext) {
+ // Just slap the key as a map into the context
+ Map keys = (Map) key;
+ for (Iterator it = keys.entrySet().iterator(); it.hasNext();) {
+ Entry entry = (Entry)it.next();
+ executionContext.put(entry.getKey().toString(), entry.getValue());
+ }
+ }
+ public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+ return null;
+ }
+ });
keyStrategy.saveState(key, executionContext);
Properties props = executionContext.getProperties();
assertEquals(2, props.size());
- assertEquals("3", props.get(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "0"));
- assertEquals("3", props.get(ColumnMapExecutionContextRowMapper.KEY_PREFIX + "1"));
+ System.err.println(props);
+ assertEquals("3", props.get("ID"));
+ assertEquals("4", props.get("VALUE"));
}
public void testGetNullKeyAsStreamContext(){
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 3a45561f6..4aad27a6e 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
@@ -159,7 +159,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase {
reader.read();
// get restart data
- reader.beforeSave();
+ reader.update();
assertEquals(4, executionContext.getLong(
FlatFileItemReader.class.getName() + "." + FlatFileItemReader.READ_STATISTICS_NAME));
// close input
@@ -174,7 +174,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase {
assertEquals("[testLine5]", reader.read().toString());
assertEquals("[testLine6]", reader.read().toString());
- reader.beforeSave();
+ reader.update();
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/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java
index e26362344..c35f16eb4 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
@@ -224,7 +224,7 @@ public class FlatFileItemWriterTests extends TestCase {
commit();
// get restart data
- inputSource.beforeSave();
+ inputSource.update();
// close template
inputSource.close();
@@ -237,7 +237,7 @@ public class FlatFileItemWriterTests extends TestCase {
inputSource.write("testLine8");
// get statistics
- inputSource.beforeSave();
+ inputSource.update();
// close template
inputSource.close();
@@ -268,7 +268,7 @@ public class FlatFileItemWriterTests extends TestCase {
inputSource.setFieldSetUnmapper(new PassThroughFieldSetMapper());
inputSource.afterPropertiesSet();
inputSource.open(executionContext);
- inputSource.beforeSave();
+ inputSource.update();
assertNotNull(executionContext);
assertEquals(3, executionContext.entrySet().size());
assertEquals(0, executionContext.getLong(FlatFileItemWriter.class.getName() + "." + FlatFileItemWriter.RESTART_DATA_NAME));
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 43a31e5ee..d83ed5a9e 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
@@ -79,7 +79,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(itemReader).beforeSave();
+ getAsItemStream(itemReader).update();
// create new input source
itemReader = createItemReader();
@@ -101,7 +101,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(itemReader).beforeSave();
+ getAsItemStream(itemReader).update();
// create new input source
itemReader = createItemReader();
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 74bc7781a..9e95ff175 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
@@ -89,7 +89,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
Foo foo2 = (Foo) reader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(reader).beforeSave();
+ getAsItemStream(reader).update();
// create new input source
reader = createItemReader();
@@ -113,7 +113,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
Foo foo2 = (Foo) reader.read();
assertEquals(2, foo2.getValue());
- getAsItemStream(reader).beforeSave();
+ getAsItemStream(reader).update();
// create new input source
reader = createItemReader();
@@ -135,7 +135,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
* @throws Exception
*/
public void testRestoreFromEmptyData() throws Exception {
- getAsItemStream(reader).beforeSave();
+ getAsItemStream(reader).update();
Foo foo = (Foo) reader.read();
assertEquals(1, foo.getValue());
@@ -218,7 +218,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
rollback();
- getAsItemStream(reader).beforeSave();
+ getAsItemStream(reader).update();
// create new input source
reader = createItemReader();
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 cb9eb1fd3..3f16097a5 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
@@ -117,7 +117,7 @@ public class StaxEventItemReaderTests extends TestCase {
public void testRestart() {
source.open(executionContext);
source.read();
- source.beforeSave();
+ source.update();
assertEquals(1, executionContext.getLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME));
List expectedAfterRestart = (List) source.read();
@@ -147,7 +147,7 @@ public class StaxEventItemReaderTests extends TestCase {
public void testRestoreWorksFromClosedStream() throws Exception {
source.close();
- source.beforeSave();
+ source.update();
}
/**
* Skipping marked records after rollback.
@@ -198,13 +198,13 @@ public class StaxEventItemReaderTests extends TestCase {
public void testExecutionContext() {
final int NUMBER_OF_RECORDS = 2;
source.open(executionContext);
- source.beforeSave();
+ source.update();
for (int i = 0; i < NUMBER_OF_RECORDS; i++) {
long recordCount = extractRecordCount();
assertEquals(i, recordCount);
source.read();
- source.beforeSave();
+ source.update();
}
assertEquals(NUMBER_OF_RECORDS, extractRecordCount());
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 086dfcbbe..4368fc2df 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
@@ -94,7 +94,7 @@ public class StaxEventItemWriterTests extends TestCase {
// write record
writer.write(record);
// writer.mark();
- writer.beforeSave();
+ writer.update();
writer.close();
// create new writer from saved restart data and continue writing
@@ -124,7 +124,7 @@ public class StaxEventItemWriterTests extends TestCase {
final int NUMBER_OF_RECORDS = 10;
for (int i = 1; i <= NUMBER_OF_RECORDS; i++) {
writer.write(record);
- writer.beforeSave();
+ writer.update();
long writeStatistics = executionContext.getLong(StaxEventItemWriter.WRITE_STATISTICS_NAME);
assertEquals(i, writeStatistics);
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java
index df4da4e4d..6c2dbcaa7 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java
@@ -76,7 +76,7 @@ public class DelegatingItemReaderTests extends TestCase {
* Gets restart data from the input template
*/
public void testGetStreamContext() {
- itemProvider.beforeSave();
+ itemProvider.update();
assertEquals("foo", executionContext.getString("value"));
}
@@ -94,7 +94,7 @@ public class DelegatingItemReaderTests extends TestCase {
return PropertiesConverter.stringToProperties("a=b");
}
- public void beforeSave() {
+ public void update() {
executionContext.putString("value", "foo");
}
@@ -117,10 +117,6 @@ public class DelegatingItemReaderTests extends TestCase {
value = "after skip";
}
- public boolean isMarkSupported() {
- return false;
- }
-
public void mark() {
}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java
index edc3ee02f..7cbb46159 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java
@@ -34,10 +34,39 @@ public class SimpleStreamManagerTests extends TestCase {
private SimpleStreamManager manager = new SimpleStreamManager(new ResourcelessTransactionManager());
- private ItemStreamSupport stream = new StubItemStream();
-
private List list = new ArrayList();
+ /**
+ * Test method for
+ * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
+ */
+ public void testRegisterAndOpen() {
+ ItemStreamSupport stream = new ItemStreamSupport() {
+ public void open(ExecutionContext executionContext) throws StreamException {
+ list.add("bar");
+ }
+ };
+ manager.register(stream);
+ manager.open(null);
+ assertEquals(1, list.size());
+ }
+
+ /**
+ * Test method for
+ * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
+ */
+ public void testRegisterTwice() {
+ ItemStreamSupport stream = new ItemStreamSupport() {
+ public void open(ExecutionContext executionContext) throws StreamException {
+ list.add("bar");
+ }
+ };
+ manager.register(stream);
+ manager.register(stream);
+ manager.open(null);
+ assertEquals(1, list.size());
+ }
+
/**
* Test method for
* {@link org.springframework.batch.item.stream.SimpleStreamManager#SimpleStreamManager(org.springframework.transaction.PlatformTransactionManager)}.
@@ -68,13 +97,41 @@ public class SimpleStreamManagerTests extends TestCase {
assertEquals("bar", list.get(0));
}
+ /**
+ * Test method for
+ * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
+ */
+ public void testMark() {
+ manager.register(new ItemStreamSupport() {
+ public void update() {
+ list.add("bar");
+ }
+ });
+ manager.update();
+ assertEquals(1, list.size());
+ }
+
+ /**
+ * Test method for
+ * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
+ */
+ public void testClose() {
+ manager.register(new ItemStreamSupport() {
+ public void close() throws StreamException {
+ list.add("bar");
+ }
+ });
+ manager.close();
+ assertEquals(1, list.size());
+ }
+
/**
* Test method for
* {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
*/
public void testCommitWithoutMark() {
manager.register(new ItemStreamSupport() {
- public void mark() {
+ public void update() {
list.add("bar");
}
});
@@ -89,7 +146,7 @@ public class SimpleStreamManagerTests extends TestCase {
*/
public void testRollbackWithoutMark() {
manager.register( new ItemStreamSupport() {
- public void reset() {
+ public void update() {
list.add("bar");
}
});
@@ -98,20 +155,4 @@ public class SimpleStreamManagerTests extends TestCase {
assertEquals(0, list.size());
}
-
- private final class StubItemStream extends ItemStreamSupport {
-
- private ExecutionContext executionContext;
-
- public void open(ExecutionContext executionContext)
- throws StreamException {
- this.executionContext = executionContext;
- }
-
- public void beforeSave() {
- executionContext.putString("foo", "bar");
- }
-
- }
-
}
diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java
index 86df324a3..ad93341cf 100644
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java
@@ -66,13 +66,6 @@ public class GeneratingItemReader extends AbstractItemReaderRecoverer implements
}
- /* (non-Javadoc)
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark()
*/
@@ -96,7 +89,7 @@ public class GeneratingItemReader extends AbstractItemReaderRecoverer implements
/* (non-Javadoc)
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
*/
- public void beforeSave() {
+ public void update() {
}
public void open(ExecutionContext context) throws StreamException {
diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
index 0ad4242a6..fedff752a 100644
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
@@ -203,16 +203,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key
* {@link TransactionSynchronizationManager#bindResource(Object, Object)}),
* so they are thread bound.
*
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- return true;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
+ * @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
getBuffer().commit();
@@ -232,7 +223,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key
*
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
*/
- public void beforeSave() {
+ public void update() {
}
}
diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeWriter.java
index 3aa7b773b..251a03d1f 100644
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeWriter.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeWriter.java
@@ -76,7 +76,7 @@ public class SimpleTradeWriter extends AbstractItemWriter implements ItemStream
/* (non-Javadoc)
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
*/
- public void beforeSave() {
+ public void update() {
executionContext.putLong("trade.count", tradeCount);
}