From 74d4ec612c271f07152b102b132e6ff8d84ff779 Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 31 Jan 2008 16:31:31 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering) http://jira.springframework.org/browse/BATCH-7 Added mark()/reset() to ItemStream, and implementations. --- .../step/simple/SimpleStepExecutorTests.java | 4 +- .../io/cursor/HibernateCursorItemReader.java | 105 +++++++++++------- .../batch/io/cursor/JdbcCursorItemReader.java | 4 +- .../io/driving/DrivingQueryItemReader.java | 27 ++++- .../io/file/DefaultFlatFileItemReader.java | 26 ++++- .../batch/io/file/FlatFileItemWriter.java | 28 ++++- .../AbstractTransactionalIoSource.java | 20 +++- .../batch/io/xml/StaxEventItemReader.java | 34 +++++- .../batch/io/xml/StaxEventItemWriter.java | 42 +++++-- .../batch/item/ItemStream.java | 32 +++++- .../reader/AbstractItemStreamItemReader.java | 29 +++++ .../item/reader/DelegatingItemReader.java | 28 +++++ .../batch/item/stream/ItemStreamAdapter.java | 21 ++++ .../writer/AbstractItemStreamItemWriter.java | 29 +++++ .../batch/io/driving/FooInputSource.java | 22 ++++ .../AbstractTransactionalIoSourceTests.java | 5 +- .../reader/DelegatingItemReaderTests.java | 10 ++ .../writer/ItemWriterItemProcessorTests.java | 3 +- 18 files changed, 388 insertions(+), 81 deletions(-) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemStreamItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemStreamItemWriter.java diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index a806ddd05..38d21dfb2 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -39,12 +39,12 @@ import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.GenericStreamContext; +import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.ExitStatus; @@ -428,7 +428,7 @@ public class SimpleStepExecutorTests extends TestCase { assertEquals(0, map.size()); } - private class MockRestartableTasklet implements Tasklet, ItemStream { + private class MockRestartableTasklet extends ItemStreamAdapter implements Tasklet { private boolean getStreamContextCalled = false; 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 2a3deefc8..c6776d62d 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 @@ -56,15 +56,13 @@ import org.springframework.util.StringUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class HibernateCursorItemReader extends AbstractItemReader implements ItemReader, ItemStream, - Skippable, InitializingBean, DisposableBean { +public class HibernateCursorItemReader extends AbstractItemReader implements ItemReader, ItemStream, Skippable, + InitializingBean, DisposableBean { - private static final String RESTART_DATA_ROW_NUMBER_KEY = ClassUtils - .getShortName(HibernateCursorItemReader.class) + private static final String RESTART_DATA_ROW_NUMBER_KEY = ClassUtils.getShortName(HibernateCursorItemReader.class) + ".rowNumber"; - private static final String SKIPPED_ROWS = ClassUtils - .getShortName(HibernateCursorItemReader.class) + private static final String SKIPPED_ROWS = ClassUtils.getShortName(HibernateCursorItemReader.class) + ".skippedRows";; private SessionFactory sessionFactory; @@ -124,7 +122,8 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite skipCount = 0; if (useStatelessSession) { statelessSession.close(); - } else { + } + else { statefulSession.close(); } } @@ -136,19 +135,18 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite if (useStatelessSession) { statelessSession = sessionFactory.openStatelessSession(); cursor = statelessSession.createQuery(queryString).scroll(); - } else { + } + else { statefulSession = sessionFactory.openSession(); cursor = statefulSession.createQuery(queryString).scroll(); } - BatchTransactionSynchronizationManager - .registerSynchronization(synchronization); + BatchTransactionSynchronizationManager.registerSynchronization(synchronization); initialized = true; } /** - * @param sessionFactory - * hibernate session factory + * @param sessionFactory hibernate session factory */ public void setSessionFactory(SessionFactory sessionFactory) { this.sessionFactory = sessionFactory; @@ -164,8 +162,7 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite } /** - * @param queryString - * HQL query string + * @param queryString HQL query string */ public void setQueryString(String queryString) { this.queryString = queryString; @@ -174,9 +171,9 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite /** * Can be set only in uninitialized state. * - * @param useStatelessSession - * true to use {@link StatelessSession} - * false to use standard hibernate {@link Session} + * @param useStatelessSession true to use + * {@link StatelessSession} false to use standard hibernate + * {@link Session} */ public void setUseStatelessSession(boolean useStatelessSession) { Assert.state(!initialized); @@ -188,10 +185,9 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite */ public StreamContext getStreamContext() { Properties props = new Properties(); - props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, ""+currentProcessedRow); + props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, "" + currentProcessedRow); String skipped = skippedRows.toString(); - props.setProperty(SKIPPED_ROWS, skipped.substring(1, - skipped.length() - 1)); + props.setProperty(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1)); return new GenericStreamContext(props); } @@ -200,25 +196,21 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite * Sets the cursor to the received row number. */ public void restoreFrom(StreamContext data) { - Assert - .state(!initialized, - "Cannot restore when already intialized. Call close() first before restore()"); + 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)); + currentProcessedRow = Integer.parseInt(props.getProperty(RESTART_DATA_ROW_NUMBER_KEY)); open(); - cursor.setRowNumber(currentProcessedRow-1); + cursor.setRowNumber(currentProcessedRow - 1); if (!props.containsKey(SKIPPED_ROWS)) { return; } - String[] skipped = StringUtils.commaDelimitedListToStringArray(props - .getProperty(SKIPPED_ROWS)); + String[] skipped = StringUtils.commaDelimitedListToStringArray(props.getProperty(SKIPPED_ROWS)); for (int i = 0; i < skipped.length; i++) { this.skippedRows.add(new Integer(skipped[i])); } @@ -238,26 +230,53 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite /** * Encapsulates transaction events handling. */ - private class HibernateItemReaderTransactionSynchronization extends - TransactionSynchronizationAdapter { + private class HibernateItemReaderTransactionSynchronization extends TransactionSynchronizationAdapter { public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - currentProcessedRow = lastCommitRowNumber; - if (lastCommitRowNumber == 0) { - cursor.beforeFirst(); - } else { - // Set the cursor so that next time it is advanced it will - // come back to the committed row. - cursor.setRowNumber(lastCommitRowNumber-1); - } - } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - lastCommitRowNumber = currentProcessedRow; - if (!useStatelessSession) { - statefulSession.clear(); - } + mark(null); + } + else if (status == TransactionSynchronization.STATUS_COMMITTED) { + reset(null); } } } + /** + * Always true, but only supported through a single processed row count, so + * do not use in an asynchronous setting. + * + * @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) + */ + public void mark(StreamContext streamContext) { + currentProcessedRow = lastCommitRowNumber; + if (lastCommitRowNumber == 0) { + cursor.beforeFirst(); + } + else { + // Set the cursor so that next time it is advanced it will + // come back to the committed row. + cursor.setRowNumber(lastCommitRowNumber - 1); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + lastCommitRowNumber = currentProcessedRow; + if (!useStatelessSession) { + statefulSession.clear(); + } + } + } 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 327d49535..db4ca72dd 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 @@ -233,7 +233,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen * Mark the current row. Calling reset will cause the result set to be set * to the current row when mark was called. */ - protected void transactionCommitted() { + public void mark(StreamContext streamContext) { lastCommittedRow = currentProcessedRow; skippedRows.clear(); } @@ -243,7 +243,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen * * @throws DataAccessException */ - protected void transactionRolledBack() { + public void reset(StreamContext streamContext) { try { currentProcessedRow = lastCommittedRow; if (currentProcessedRow > 0) { 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 93dad32ec..03cdcef92 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 @@ -209,11 +209,11 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource } protected void transactionCommitted() { - lastCommitIndex = currentIndex; + mark(null); } protected void transactionRolledBack() { - keysIterator = keys.listIterator(lastCommitIndex); + reset(null); } /** @@ -224,4 +224,27 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource return item; } + /** + * Returns 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) + */ + public void mark(StreamContext streamContext) { + lastCommitIndex = currentIndex; + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + keysIterator = keys.listIterator(lastCommitIndex); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index 31e439ccb..cea8c92ad 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -39,6 +39,7 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter * @author Waseem Malik * @author Tomas Slanina * @author Robert Kasanicky + * @author Dave Syer */ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implements Skippable, ItemStream { @@ -109,13 +110,34 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen return streamContext; } + /* (non-Javadoc) + * @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) + */ + public void mark(StreamContext streamContext) { + getReader().mark(); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + getReader().reset(); + } + /** * This method marks the start of a transaction. It marks the InputBuffer * Reader, so that in case of rollback it can position the file to start of * the transaction. */ private void transactionStarted() { - getReader().mark(); + mark(null); } /** @@ -130,7 +152,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen * Rollback the transaction. */ private void transactionRolledback() { - getReader().reset(); + reset(null); } /** 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 52d6ec0fe..e99ec7037 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 @@ -119,15 +119,14 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements * Commit the transaction. */ protected void transactionCommitted() { - getOutputState().mark(); + mark(null); } /** * Rollback the transaction. */ protected void transactionRolledBack() { - getOutputState().checkFileSize(); - resetPositionForRestart(); + reset(null); } // This method removes any information in the file before this reset point. @@ -539,4 +538,27 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements } } + + /** + * Returns 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) + */ + public void mark(StreamContext streamContext) { + getOutputState().mark(); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + getOutputState().checkFileSize(); + resetPositionForRestart(); + } } 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 0353c3710..64728edb9 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 @@ -17,6 +17,8 @@ package org.springframework.batch.io.support; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.support.TransactionSynchronization; @@ -54,7 +56,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager * @see TransactionSynchronization * @see TransactionSynchronizationManager */ -public abstract class AbstractTransactionalIoSource { +public abstract class AbstractTransactionalIoSource extends ItemStreamAdapter { private final TransactionSynchronization synchronization = new AbstractTransactionalIoSourceTransactionSynchronization(); @@ -74,14 +76,14 @@ public abstract class AbstractTransactionalIoSource { * * @see TransactionSynchronization#afterCompletion */ - protected abstract void transactionCommitted(); + public abstract void mark(StreamContext streamContext); /* * Called when a transaction has been rolled back. * * @see TransactionSynchronization#afterCompletion */ - protected abstract void transactionRolledBack(); + public abstract void reset(StreamContext streamContext); /** * Encapsulates transaction events handling. @@ -90,10 +92,18 @@ public abstract class AbstractTransactionalIoSource { TransactionSynchronizationAdapter { public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - transactionRolledBack(); + reset(null); } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - transactionCommitted(); + mark(null); } } } + + /* (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 7fba9d746..1aa3ecbf6 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 @@ -269,14 +269,10 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade */ public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_COMMITTED) { - lastCommitPointRecordCount = currentRecordCount; - txReader.onCommit(); - skipRecords = new ArrayList(); + mark(null); } else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - currentRecordCount = lastCommitPointRecordCount; - txReader.onRollback(); - fragmentReader.reset(); + reset(null); } } @@ -290,4 +286,30 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade BatchTransactionSynchronizationManager.registerSynchronization(synchronization); } + /* (non-Javadoc) + * @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) + */ + public void mark(StreamContext streamContext) { + lastCommitPointRecordCount = currentRecordCount; + txReader.onCommit(); + skipRecords = new ArrayList(); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + currentRecordCount = lastCommitPointRecordCount; + txReader.onRollback(); + fragmentReader.reset(); + } + + } 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 6c1156fad..0e42beb73 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 @@ -464,20 +464,11 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing } private void transactionComitted() { - lastCommitPointPosition = getPosition(); - lastCommitPointRecordCount = currentRecordCount; + mark(null); } private void transactionRolledback() { - currentRecordCount = lastCommitPointRecordCount; - - // close output - close(); - // and reopen it - we do this because we need to reopen stream - // reader at specified position - calling setPosition() is not - // enough! - restarted = true; - open(lastCommitPointPosition); + reset(null); } } @@ -486,4 +477,33 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing return synchronization; } + /* (non-Javadoc) + * @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) + */ + public void mark(StreamContext streamContext) { + lastCommitPointPosition = getPosition(); + lastCommitPointRecordCount = currentRecordCount; + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + currentRecordCount = lastCommitPointRecordCount; + // close output + close(); + // and reopen it - we do this because we need to reopen stream + // reader at specified position - calling setPosition() is not + // enough! + restarted = true; + open(lastCommitPointPosition); + } + } 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 a799c3076..f9d996079 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 @@ -31,7 +31,7 @@ package org.springframework.batch.item; * provided. *

* - * @author Lucas Ward + * @author Dave Syer * */ public interface ItemStream extends StreamContextProvider { @@ -55,4 +55,34 @@ public interface ItemStream extends StreamContextProvider { * (except open) may throw an exception. */ void close() throws StreamException; + + /** + * Clients are expected to check this flag before calling mark or reset. + * + * @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. + * Implementations may use the information in the provided context to make + * calculations that account for things like multiple open cursors. The + * context should also be updated with any information of this nature that + * might be needed by a reset or by future calls to mark. + * + * @param the context which might contain information needed to determine + * what action to take, and into which the current mark information can go. + * + * @throws UnsupportedOperationException if the operation is not supported + */ + void mark(StreamContext streamContext); + + /** + * 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 + * with the same context will not be visible after a call to close. + * + * @throws UnsupportedOperationException if the operation is not supported + */ + void reset(StreamContext streamContext); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemStreamItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemStreamItemReader.java new file mode 100644 index 000000000..9961c248e --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemStreamItemReader.java @@ -0,0 +1,29 @@ +/* + * 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.reader; + +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.stream.ItemStreamAdapter; + +/** + * Base class for {@link ItemReader} implementations. + * @author Dave Syer + * + */ +public abstract class AbstractItemStreamItemReader extends ItemStreamAdapter implements ItemReader { + +} 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 ca7f5b3da..0b27ea197 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 @@ -103,4 +103,32 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl ((ItemStream) inputSource).close(); } } + + /* (non-Javadoc) + * @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.StreamContext) + */ + public void mark(StreamContext streamContext) { + if (inputSource instanceof ItemStream) { + ((ItemStream) inputSource).mark(streamContext); + } + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + if (inputSource instanceof ItemStream) { + ((ItemStream) inputSource).reset(streamContext); + } + } } 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 81b462c33..c57867479 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 @@ -56,4 +56,25 @@ public class ItemStreamAdapter implements ItemStream { return new GenericStreamContext(new Properties()); } + /* (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.StreamContext) + */ + public void mark(StreamContext streamContext) { + throw new UnsupportedOperationException("Mark operation not supported."); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + throw new UnsupportedOperationException("Reset operation not supported."); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemStreamItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemStreamItemWriter.java new file mode 100644 index 000000000..9ac1273fb --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemStreamItemWriter.java @@ -0,0 +1,29 @@ +/* + * 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.writer; + +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.stream.ItemStreamAdapter; + +/** + * Base class for {@link ItemWriter} implementations. + * @author Dave Syer + * + */ +public abstract class AbstractItemStreamItemWriter extends ItemStreamAdapter implements ItemWriter { + +} 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 7f7a5da21..ecfe9cecc 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 @@ -50,5 +50,27 @@ 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) + */ + public void mark(StreamContext streamContext) { + inputSource.mark(streamContext); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + inputSource.reset(streamContext); }; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java index 134558582..ba5e65fcc 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java @@ -20,6 +20,7 @@ import java.util.List; import junit.framework.TestCase; +import org.springframework.batch.item.StreamContext; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionSynchronizationUtils; @@ -102,12 +103,12 @@ public class AbstractTransactionalIoSourceTests extends TestCase { private boolean commitCalled = false; private boolean rollbackCalled = false; - protected void transactionCommitted() { + public void mark(StreamContext streamContext) { Assert.isTrue(!commitCalled, "Commit aleady called"); commitCalled = true; } - protected void transactionRolledBack() { + public void reset(StreamContext streamContext) { Assert.isTrue(!rollbackCalled, "Rollback aleady called"); rollbackCalled = true; } 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 c6cf2f9a8..452c0a6c1 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 @@ -126,6 +126,16 @@ public class DelegatingItemReaderTests extends TestCase { value = "after skip"; } + public boolean isMarkSupported() { + return false; + } + + public void mark(StreamContext streamContext) { + } + + public void reset(StreamContext streamContext) { + } + } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java index 0589fb385..1dfa6b7ca 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java @@ -22,7 +22,6 @@ import java.util.Properties; import junit.framework.TestCase; import org.springframework.batch.io.Skippable; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; @@ -128,7 +127,7 @@ public class ItemWriterItemProcessorTests extends TestCase { * @author Dave Syer * */ - public class MockOutputSource implements ItemWriter, ItemStream, Skippable { + public class MockOutputSource extends AbstractItemStreamItemWriter implements Skippable { private String value;