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 ddc7da6dc..39aa80af1 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
@@ -16,6 +16,8 @@
package org.springframework.batch.item;
+import org.springframework.batch.item.exception.MarkFailedException;
+import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.reader.AbstractItemReader;
/**
@@ -50,5 +52,36 @@ 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#getExecutionContext()}.
+ *
+ * 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
+ * will not be visible after a call to close.
+ *
+ * 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
+ * exception, it may be important for a client to catch it explicitly.
+ */
+ void reset() throws ResetFailedException;
}
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 eb256954c..82e362453 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
@@ -16,8 +16,6 @@
package org.springframework.batch.item;
-import org.springframework.batch.item.exception.MarkFailedException;
-import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
/**
@@ -61,49 +59,4 @@ public interface ItemStream extends ExecutionContextProvider {
* (except open) may throw an exception.
*/
void close() throws StreamException;
-
- /**
- * Clients are expected to check this flag before calling mark or reset.
- *
- * Implementations should also document explicitly, if mark is supported,
- * how it will behave in a multi-threaded environment. Generally, if the
- * stream is being accessed from multiple threads concurrently, it will have
- * to manage that internally, and also reflect only the completed marks
- * (independent of the order they happen) when
- * {@link ExecutionContextProvider#getExecutionContext()} is called.
- *
- * @return true if mark and reset are supported by the {@link ItemStream}
- */
- boolean isMarkSupported();
-
- /**
- * 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()}.
- *
- * 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
- * will not be visible after a call to close.
- *
- * 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
- * exception, it may be important for a client to catch it explicitly.
- */
- void reset() throws ResetFailedException;
}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java
index 5b060d614..c0a8e3c89 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java
@@ -17,6 +17,8 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.exception.MarkFailedException;
+import org.springframework.batch.item.exception.ResetFailedException;
/**
* Base class for {@link ItemReader} implementations.
@@ -25,4 +27,9 @@ import org.springframework.batch.item.ItemReader;
*/
public abstract class AbstractItemReader implements ItemReader {
+ public void mark() throws MarkFailedException {
+ }
+
+ public void reset() throws ResetFailedException {
+ }
}
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 3d59471ad..5898d5422 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
@@ -33,10 +33,10 @@ import org.springframework.util.Assert;
*/
public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, ItemStream {
- private ItemReader inputSource;
+ private ItemReader itemReader;
public void afterPropertiesSet() throws Exception {
- Assert.notNull(inputSource, "ItemReader must not be null.");
+ Assert.notNull(itemReader, "ItemReader must not be null.");
}
/**
@@ -45,7 +45,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemReader#read()
*/
public Object read() throws Exception {
- return inputSource.read();
+ return itemReader.read();
}
/**
@@ -54,8 +54,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* {@link ItemStream}.
*/
public ExecutionContext getExecutionContext() {
- if (inputSource instanceof ItemStream) {
- return ((ItemStream) inputSource).getExecutionContext();
+ if (itemReader instanceof ItemStream) {
+ return ((ItemStream) itemReader).getExecutionContext();
}
return new ExecutionContext();
}
@@ -66,8 +66,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* {@link ItemStream}.
*/
public void restoreFrom(ExecutionContext data) {
- if (inputSource instanceof ItemStream) {
- ((ItemStream) inputSource).restoreFrom(data);
+ if (itemReader instanceof ItemStream) {
+ ((ItemStream) itemReader).restoreFrom(data);
}
}
@@ -76,16 +76,16 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @param source
*/
public void setItemReader(ItemReader source) {
- this.inputSource = source;
+ this.itemReader = source;
}
public ItemReader getItemReader() {
- return inputSource;
+ return itemReader;
}
public void skip() {
- if (inputSource instanceof Skippable) {
- ((Skippable) inputSource).skip();
+ if (itemReader instanceof Skippable) {
+ ((Skippable) itemReader).skip();
}
}
@@ -94,8 +94,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#open()
*/
public void open() throws StreamException {
- if (inputSource instanceof ItemStream) {
- ((ItemStream) inputSource).open();
+ if (itemReader instanceof ItemStream) {
+ ((ItemStream) itemReader).open();
}
}
@@ -104,31 +104,17 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#open()
*/
public void close() throws StreamException {
- if (inputSource instanceof ItemStream) {
- ((ItemStream) inputSource).close();
+ if (itemReader instanceof ItemStream) {
+ ((ItemStream) itemReader).close();
}
}
- /**
- * Delegates the call if the delegate is an {@link ItemStream}.
- *
- * @see org.springframework.batch.item.ItemStream#isMarkSupported()
- */
- public boolean isMarkSupported() {
- if (inputSource instanceof ItemStream) {
- return ((ItemStream) inputSource).isMarkSupported();
- }
- return false;
- }
-
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
- if (inputSource instanceof ItemStream) {
- ((ItemStream) inputSource).mark();
- }
+ itemReader.mark();
}
/*
@@ -136,8 +122,6 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
- if (inputSource instanceof ItemStream) {
- ((ItemStream) inputSource).reset();
- }
+ itemReader.reset();
}
}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java
index f8102c571..c96891814 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java
@@ -17,6 +17,8 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.exception.MarkFailedException;
+import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.support.AbstractMethodInvokingDelegator;
@@ -43,5 +45,10 @@ public class ItemReaderAdapter extends AbstractMethodInvokingDelegator implement
}
+ public void mark() throws MarkFailedException {
+ }
+
+ public void reset() throws ResetFailedException {
+ }
}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java
index 2a398eb04..d907f02ed 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java
@@ -65,14 +65,12 @@ public class ItemStreamAdapter implements ItemStream {
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
- throw new UnsupportedOperationException("Mark operation not supported.");
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
- throw new UnsupportedOperationException("Reset operation not supported.");
}
}
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 34e85c776..d1266871a 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
@@ -29,9 +29,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.transaction.support.TransactionSynchronization;
-import org.springframework.transaction.support.TransactionSynchronizationAdapter;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.ClassUtils;
/**
@@ -193,28 +190,6 @@ public class SimpleStreamManager implements StreamManager {
*/
public TransactionStatus getTransaction(final Object key) {
TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition());
- TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
- public void afterCompletion(int status) {
- if (status == TransactionSynchronization.STATUS_COMMITTED) {
- iterate(key, new Callback() {
- public void execute(ItemStream stream) {
- if (stream.isMarkSupported()) {
- stream.mark();
- }
- }
- });
- }
- else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
- iterate(key, new Callback() {
- public void execute(ItemStream stream) {
- if (stream.isMarkSupported()) {
- stream.reset();
- }
- }
- });
- }
- }
- });
return transaction;
}
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 d41a2b760..c56f34e36 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
@@ -16,7 +16,7 @@ import org.springframework.util.Assert;
public class DrivingQueryItemReaderTests extends TestCase {
- ItemReader source;
+ ItemReader itemReader;
static {
TransactionSynchronizationManager.initSynchronization();
@@ -25,7 +25,7 @@ public class DrivingQueryItemReaderTests extends TestCase {
protected void setUp() throws Exception {
super.setUp();
- source = createItemReader();
+ itemReader = createItemReader();
}
private ItemReader createItemReader() throws Exception{
@@ -41,24 +41,24 @@ public class DrivingQueryItemReaderTests extends TestCase {
* Regular scenario - read all rows and eventually return null.
*/
public void testNormalProcessing() throws Exception {
- getAsInitializingBean(source).afterPropertiesSet();
+ getAsInitializingBean(itemReader).afterPropertiesSet();
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- Foo foo3 = (Foo) source.read();
+ Foo foo3 = (Foo) itemReader.read();
assertEquals(3, foo3.getValue());
- Foo foo4 = (Foo) source.read();
+ Foo foo4 = (Foo) itemReader.read();
assertEquals(4, foo4.getValue());
- Foo foo5 = (Foo) source.read();
+ Foo foo5 = (Foo) itemReader.read();
assertEquals(5, foo5.getValue());
- assertNull(source.read());
+ assertNull(itemReader.read());
}
/**
@@ -67,20 +67,20 @@ public class DrivingQueryItemReaderTests extends TestCase {
*/
public void testRestart() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- ExecutionContext streamContext = getAsRestartable(source).getExecutionContext();
+ ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext();
// create new input source
- source = createItemReader();
+ itemReader = createItemReader();
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
- Foo fooAfterRestart = (Foo) source.read();
+ Foo fooAfterRestart = (Foo) itemReader.read();
assertEquals(3, fooAfterRestart.getValue());
}
@@ -89,22 +89,22 @@ public class DrivingQueryItemReaderTests extends TestCase {
*/
public void testInvalidRestore() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- ExecutionContext streamContext = getAsRestartable(source).getExecutionContext();
+ ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext();
// create new input source
- source = createItemReader();
+ itemReader = createItemReader();
- Foo foo = (Foo) source.read();
+ Foo foo = (Foo) itemReader.read();
assertEquals(1, foo.getValue());
try {
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
fail();
}
catch (IllegalStateException ex) {
@@ -119,9 +119,9 @@ public class DrivingQueryItemReaderTests extends TestCase {
public void testRestoreFromEmptyData() throws Exception {
ExecutionContext streamContext = new ExecutionContext(new Properties());
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
- Foo foo = (Foo) source.read();
+ Foo foo = (Foo) itemReader.read();
assertEquals(1, foo.getValue());
}
@@ -130,28 +130,28 @@ public class DrivingQueryItemReaderTests extends TestCase {
* @throws Exception
*/
public void testRollback() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
commit();
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
Assert.state(!foo2.equals(foo1));
- Foo foo3 = (Foo) source.read();
+ Foo foo3 = (Foo) itemReader.read();
Assert.state(!foo2.equals(foo3));
rollback();
- assertEquals(foo2, source.read());
+ assertEquals(foo2, itemReader.read());
}
private void commit() {
- ((ItemStream) source).mark();
+ itemReader.mark();
}
private void rollback() {
- ((ItemStream) source).reset();
+ itemReader.reset();
}
private InitializingBean getAsInitializingBean(ItemReader source) {
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 8d41b5862..e4d2f8697 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
@@ -17,7 +17,7 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests {
- protected ItemReader source;
+ protected ItemReader itemReader;
/**
@@ -31,12 +31,12 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
protected void onSetUp()throws Exception{
super.onSetUp();
- source = createItemReader();
- getAsInitializingBean(source).afterPropertiesSet();
+ itemReader = createItemReader();
+ getAsInitializingBean(itemReader).afterPropertiesSet();
}
protected void onTearDown()throws Exception {
- getAsDisposableBean(source).destroy();
+ getAsDisposableBean(itemReader).destroy();
super.onTearDown();
}
@@ -44,24 +44,24 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
* Regular scenario - read all rows and eventually return null.
*/
public void testNormalProcessing() throws Exception {
- getAsInitializingBean(source).afterPropertiesSet();
+ getAsInitializingBean(itemReader).afterPropertiesSet();
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- Foo foo3 = (Foo) source.read();
+ Foo foo3 = (Foo) itemReader.read();
assertEquals(3, foo3.getValue());
- Foo foo4 = (Foo) source.read();
+ Foo foo4 = (Foo) itemReader.read();
assertEquals(4, foo4.getValue());
- Foo foo5 = (Foo) source.read();
+ Foo foo5 = (Foo) itemReader.read();
assertEquals(5, foo5.getValue());
- assertNull(source.read());
+ assertNull(itemReader.read());
}
/**
@@ -70,20 +70,20 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
*/
public void testRestart() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- ExecutionContext streamContext = getAsRestartable(source).getExecutionContext();
+ ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext();
// create new input source
- source = createItemReader();
+ itemReader = createItemReader();
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
- Foo fooAfterRestart = (Foo) source.read();
+ Foo fooAfterRestart = (Foo) itemReader.read();
assertEquals(3, fooAfterRestart.getValue());
}
@@ -92,22 +92,22 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
*/
public void testInvalidRestore() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
assertEquals(1, foo1.getValue());
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
assertEquals(2, foo2.getValue());
- ExecutionContext streamContext = getAsRestartable(source).getExecutionContext();
+ ExecutionContext streamContext = getAsRestartable(itemReader).getExecutionContext();
// create new input source
- source = createItemReader();
+ itemReader = createItemReader();
- Foo foo = (Foo) source.read();
+ Foo foo = (Foo) itemReader.read();
assertEquals(1, foo.getValue());
try {
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
fail();
}
catch (IllegalStateException ex) {
@@ -122,9 +122,9 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
public void testRestoreFromEmptyData() throws Exception {
ExecutionContext streamContext = new ExecutionContext();
- getAsRestartable(source).restoreFrom(streamContext);
+ getAsRestartable(itemReader).restoreFrom(streamContext);
- Foo foo = (Foo) source.read();
+ Foo foo = (Foo) itemReader.read();
assertEquals(1, foo.getValue());
}
@@ -133,28 +133,28 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
* @throws Exception
*/
public void testRollback() throws Exception {
- Foo foo1 = (Foo) source.read();
+ Foo foo1 = (Foo) itemReader.read();
commit();
- Foo foo2 = (Foo) source.read();
+ Foo foo2 = (Foo) itemReader.read();
Assert.state(!foo2.equals(foo1));
- Foo foo3 = (Foo) source.read();
+ Foo foo3 = (Foo) itemReader.read();
Assert.state(!foo2.equals(foo3));
rollback();
- assertEquals(foo2, source.read());
+ assertEquals(foo2, itemReader.read());
}
private void commit() {
- ((ItemStream) source).mark();
+ itemReader.mark();
}
private void rollback() {
- ((ItemStream) source).reset();
+ itemReader.reset();
}
private ItemStream getAsRestartable(ItemReader 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 e1c511519..f687781f6 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
@@ -225,11 +225,11 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
}
private void commit() {
- ((ItemStream) reader).mark();
+ reader.mark();
}
private void rollback() {
- ((ItemStream) reader).reset();
+ reader.reset();
}
private Skippable getAsSkippable(ItemReader source) {
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 5dcfffb04..d929a5fb0 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
@@ -168,25 +168,6 @@ public class SimpleStreamManagerTests extends TestCase {
assertEquals(1, list.size());
}
- /**
- * Test method for
- * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
- */
- public void testCommit() {
- manager.register("foo", new ItemStreamAdapter() {
- public boolean isMarkSupported() {
- return true;
- }
-
- public void mark() {
- list.add("bar");
- }
- });
- TransactionStatus status = manager.getTransaction("foo");
- manager.commit(status);
- assertEquals(1, list.size());
- }
-
/**
* Test method for
* {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}.
@@ -202,25 +183,6 @@ public class SimpleStreamManagerTests extends TestCase {
assertEquals(0, list.size());
}
- /**
- * Test method for
- * {@link org.springframework.batch.item.stream.SimpleStreamManager#rollback(org.springframework.transaction.TransactionStatus)}.
- */
- public void testRollback() {
- manager.register("foo", new ItemStreamAdapter() {
- public boolean isMarkSupported() {
- return true;
- }
-
- public void reset() {
- list.add("bar");
- }
- });
- TransactionStatus status = manager.getTransaction("foo");
- manager.rollback(status);
- assertEquals(1, list.size());
- }
-
/**
* Test method for
* {@link org.springframework.batch.item.stream.SimpleStreamManager#rollback(org.springframework.transaction.TransactionStatus)}.