From 2c3d43d7bdff8225efe3b744d424f3193fe3e6cb Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 31 Jan 2008 22:36:26 +0000 Subject: [PATCH] BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering) http://jira.springframework.org/browse/BATCH-7 Use StreamManager to start/end transaction in step executor --- spring-batch-core/.classpath | 2 +- spring-batch-execution/.classpath | 2 +- .../.settings/org.eclipse.jdt.core.prefs | 8 +- .../step/simple/SimpleStepExecutor.java | 77 ++++---- .../scope/SimpleStepContextTests.java | 4 +- .../step/simple/SimpleStepExecutorTests.java | 2 +- spring-batch-infrastructure/.classpath | 2 +- .../io/cursor/HibernateCursorItemReader.java | 51 ++--- .../batch/io/cursor/JdbcCursorItemReader.java | 1 - .../io/driving/DrivingQueryItemReader.java | 1 - .../io/file/DefaultFlatFileItemReader.java | 63 +----- .../batch/io/file/FlatFileItemWriter.java | 1 - .../AbstractTransactionalIoSource.java | 51 ----- .../batch/io/xml/StaxEventItemReader.java | 39 +--- .../batch/io/xml/StaxEventItemWriter.java | 44 +---- .../item/stream/SimpleStreamManager.java | 100 +++++++++- .../batch/item/stream/StreamManager.java | 7 + ...atchTransactionSynchronizationManager.java | 187 ------------------ .../driving/DrivingQueryItemReaderTests.java | 10 +- .../file/DefaultFlatFileItemReaderTests.java | 36 +--- .../io/file/FlatFileItemWriterTests.java | 22 +-- ...bstractJdbcItemReaderIntegrationTests.java | 11 +- ...tDataSourceItemReaderIntegrationTests.java | 11 +- .../AbstractTransactionalIoSourceTests.java | 67 +------ .../io/xml/StaxEventItemReaderTests.java | 11 +- .../xml/StaxEventWriterItemWriterTests.java | 13 +- .../src/main/resources/jobs/restartSample.xml | 2 +- 27 files changed, 194 insertions(+), 631 deletions(-) delete mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java diff --git a/spring-batch-core/.classpath b/spring-batch-core/.classpath index 6ceeb8b67..6165ff76e 100644 --- a/spring-batch-core/.classpath +++ b/spring-batch-core/.classpath @@ -4,7 +4,7 @@ - + diff --git a/spring-batch-execution/.classpath b/spring-batch-execution/.classpath index 6ceeb8b67..6165ff76e 100644 --- a/spring-batch-execution/.classpath +++ b/spring-batch-execution/.classpath @@ -4,7 +4,7 @@ - + diff --git a/spring-batch-execution/.settings/org.eclipse.jdt.core.prefs b/spring-batch-execution/.settings/org.eclipse.jdt.core.prefs index b0e76afe9..cf0919644 100644 --- a/spring-batch-execution/.settings/org.eclipse.jdt.core.prefs +++ b/spring-batch-execution/.settings/org.eclipse.jdt.core.prefs @@ -1,5 +1,7 @@ -#Fri Aug 03 14:01:44 BST 2007 -org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.4 +#Thu Jan 31 17:05:10 GMT 2008 eclipse.preferences.version=1 -org.eclipse.jdt.core.compiler.source=1.4 +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.4 org.eclipse.jdt.core.compiler.compliance=1.4 +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enumIdentifier=warning +org.eclipse.jdt.core.compiler.source=1.4 diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 08232966f..576da8855 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -46,12 +46,9 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler; import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; /** @@ -94,7 +91,7 @@ public class SimpleStepExecutor { private AbstractStep step; - private StreamManager streamManager = new SimpleStreamManager(); + private StreamManager streamManager; /** * Package private constructor so the factory can create a the executor. @@ -106,8 +103,8 @@ public class SimpleStepExecutor { /** * Public setter for the {@link StreamManager}. This will be used to create * the {@link StepContext}, and hence any component that is a - * {@link ItemStream} and in step scope will be registered with the service. The - * {@link StepContext} is then a source of aggregate statistics for the + * {@link ItemStream} and in step scope will be registered with the service. + * The {@link StepContext} is then a source of aggregate statistics for the * step. * * @param streamManager the {@link StreamManager} to set. Default is a @@ -180,6 +177,10 @@ public class SimpleStepExecutor { ExitStatus status = ExitStatus.FAILED; + if (streamManager==null) { + streamManager = new SimpleStreamManager(transactionManager); + } + StepContext parentStepContext = StepSynchronizationManager.getContext(); final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager); StepSynchronizationManager.register(stepContext); @@ -192,7 +193,7 @@ public class SimpleStepExecutor { if (saveStreamContext && isRestart) { stepContext.setInitialStreamContext(stepInstance.getStreamContext()); } - + try { stepExecution.setStartTime(new Date(System.currentTimeMillis())); @@ -210,40 +211,31 @@ public class SimpleStepExecutor { ExitStatus result; + TransactionStatus transaction = streamManager.getTransaction(stepContext); + try { - result = (ExitStatus) new TransactionTemplate(transactionManager) - .execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - /* - * New transaction obtained, - * resynchronize - * TransactionSynchronization objects - */ - BatchTransactionSynchronizationManager.resynchronize(); - ExitStatus result; + /* + * New transaction obtained, resynchronize + * TransactionSynchronization objects + */ + result = processChunk(step, contribution); - result = processChunk(step, contribution); + // TODO: check that stepExecution can + // aggregate these contributions if they + // come in asynchronously. + StreamContext statistics = stepContext.getStreamContext(); + contribution.setStreamContext(new GenericStreamContext(statistics.getProperties())); + contribution.incrementCommitCount(); + // Apply the contribution to the step + // only if chunk was successful + stepExecution.apply(contribution); - // TODO: check that stepExecution can - // aggregate these contributions if they - // come in asynchronously. - StreamContext statistics = stepContext.getStreamContext(); - contribution.setStreamContext(new GenericStreamContext(statistics - .getProperties())); - contribution.incrementCommitCount(); - // Apply the contribution to the step - // only if chunk was successful - stepExecution.apply(contribution); - - if (saveStreamContext) { - stepInstance.setStreamContext(stepContext.getStreamContext()); - jobRepository.update(stepInstance); - } - jobRepository.saveOrUpdate(stepExecution); - return result; - } - }); + if (saveStreamContext) { + stepInstance.setStreamContext(stepContext.getStreamContext()); + jobRepository.update(stepInstance); + } + jobRepository.saveOrUpdate(stepExecution); } catch (Throwable t) { @@ -255,6 +247,7 @@ public class SimpleStepExecutor { * comes outside the transaction. */ stepExecution.rollback(); + streamManager.rollback(transaction); if (t instanceof RuntimeException) { throw (RuntimeException) t; } @@ -268,10 +261,11 @@ public class SimpleStepExecutor { // caller interruptionPolicy.checkInterrupted(context); + streamManager.commit(transaction); + return result; } - }); updateStatus(stepExecution, BatchStatus.COMPLETED); @@ -298,12 +292,7 @@ public class SimpleStepExecutor { } finally { // clear any registered synchronizations - try { - StepSynchronizationManager.close(); - } - finally { - BatchTransactionSynchronizationManager.clearSynchronizations(); - } + StepSynchronizationManager.close(); } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java index 9b74926b6..7c10d5884 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java @@ -27,7 +27,7 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.item.stream.ItemStreamAdapter; -import org.springframework.batch.item.stream.StreamManager; +import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.support.PropertiesConverter; /** @@ -155,7 +155,7 @@ public class SimpleStepContextTests extends TestCase { * @author Dave Syer * */ - private class StubStreamManager implements StreamManager { + private class StubStreamManager extends SimpleStreamManager { private final Map map; private StubStreamManager(Map map) { 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 38d21dfb2..9c78bef85 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 @@ -412,7 +412,7 @@ public class SimpleStepExecutorTests extends TestCase { assertEquals(false, stepExecution.getStreamContext().containsKey("foo")); final Map map = new HashMap(); - stepExecutor.setStreamManager(new SimpleStreamManager() { + stepExecutor.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { public StreamContext getStreamContext(Object key) { // TODO Auto-generated method stub return new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar")); diff --git a/spring-batch-infrastructure/.classpath b/spring-batch-infrastructure/.classpath index 623ecbc7a..cd4c91139 100644 --- a/spring-batch-infrastructure/.classpath +++ b/spring-batch-infrastructure/.classpath @@ -3,7 +3,7 @@ - + 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 c6776d62d..9ee02fa3f 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 @@ -25,15 +25,11 @@ import org.hibernate.SessionFactory; import org.hibernate.StatelessSession; import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; -import org.springframework.batch.item.reader.AbstractItemReader; +import org.springframework.batch.item.reader.AbstractItemStreamItemReader; import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; @@ -56,8 +52,8 @@ 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 AbstractItemStreamItemReader implements Skippable, InitializingBean, + DisposableBean { private static final String RESTART_DATA_ROW_NUMBER_KEY = ClassUtils.getShortName(HibernateCursorItemReader.class) + ".rowNumber"; @@ -88,8 +84,6 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite private boolean initialized = false; - private TransactionSynchronization synchronization = new HibernateItemReaderTransactionSynchronization(); - public Object read() { if (!initialized) { open(); @@ -140,8 +134,6 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite statefulSession = sessionFactory.openSession(); cursor = statefulSession.createQuery(queryString).scroll(); } - - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); initialized = true; } @@ -227,21 +219,6 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite skipCount++; } - /** - * Encapsulates transaction events handling. - */ - private class HibernateItemReaderTransactionSynchronization extends TransactionSynchronizationAdapter { - - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - 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. @@ -257,6 +234,17 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext) */ public void mark(StreamContext streamContext) { + lastCommitRowNumber = currentProcessedRow; + if (!useStatelessSession) { + statefulSession.clear(); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { currentProcessedRow = lastCommitRowNumber; if (lastCommitRowNumber == 0) { cursor.beforeFirst(); @@ -268,15 +256,4 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite } } - /* - * (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 db4ca72dd..ebd0d8a1a 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 @@ -321,7 +321,6 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen throw getExceptionTranslator().translate("Executing query", sql, se); } - super.registerSynchronization(); } /* 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 03cdcef92..161d4b969 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 @@ -143,7 +143,6 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource + ", call close() first."); keys = keyGenerator.retrieveKeys(); keysIterator = keys.listIterator(); - super.registerSynchronization(); initialized = true; } 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 cea8c92ad..6ee490e2b 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 @@ -26,9 +26,6 @@ import org.springframework.batch.io.file.separator.LineReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; /** *

@@ -51,19 +48,12 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen private Set skippedLines = new HashSet(); - private TransactionSynchronization transactionSynchronization = new ResourceLineReaderTransactionSynchronization(); - /** * Initialize the input source. */ public void open() { - registerSynchronization(); super.open(); - } - - // Registers transaction synchronization. - private void registerSynchronization() { - BatchTransactionSynchronizationManager.registerSynchronization(transactionSynchronization); + mark(null); } /** @@ -92,6 +82,8 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen while (reader.getCurrentLineCount() < lineCount && record != null) { record = readLine(); } + + mark(data); } @@ -131,30 +123,6 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen 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() { - mark(null); - } - - /** - * Commit the transaction. At each commit point we clear the lines which we - * had skipped during the commit interval. - */ - private void transactionComitted() { - transactionStarted(); - } - - /** - * Rollback the transaction. - */ - private void transactionRolledback() { - reset(null); - } - /** * Skip the current line which is being processed. */ @@ -173,29 +141,4 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen return line; } - // added package visibility method so that tests can invoke transaction - // events - TransactionSynchronization getTransactionSynchronization() { - return this.transactionSynchronization; - } - - /** - * Encapsulates transaction events. - */ - private class ResourceLineReaderTransactionSynchronization extends TransactionSynchronizationAdapter { - /** - * TransactionSynchronization method indicating that a transaction has - * completed. - * - * @param status indicates whether it was a rollback or commit - */ - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_COMMITTED) { - transactionComitted(); - } - else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - transactionRolledback(); - } - } - } } 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 e99ec7037..9782621ed 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 @@ -242,7 +242,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements */ public void open() { getOutputState(); - super.registerSynchronization(); } /** 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 64728edb9..4741b7902 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 @@ -19,11 +19,6 @@ 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; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; -import org.springframework.transaction.support.TransactionSynchronizationManager; /** *

@@ -34,43 +29,11 @@ import org.springframework.transaction.support.TransactionSynchronizationManager * maintained regardless of rollbacks. *

* - *

- * This class is primarily useful because it allows its subclasses to implement - * a single method to be notified of a commit or rollback, rather than having an - * inner class that implements {@link TransactionSyncrhonization} and likely - * calls another method with similar semantics as commit and rollback. - *

- * - *

- * It should be noted that this implementation will only register for - * synchronization if a call to registerSynchronization() has been made. This is - * less than ideal, however, it is the best solution until {@link StepScope} is - * modified to handle registering synchronizations in a scoped manner. - * Otherwise, registering at instantiation or initialization (such as via the - * Spring {@link InitializingBean} interface) would cause commits to be called - * on input sources for all steps, rather than the currently running step. - *

- * * @author Lucas Ward * @since 1.0 - * @see TransactionSynchronization - * @see TransactionSynchronizationManager */ public abstract class AbstractTransactionalIoSource extends ItemStreamAdapter { - private final TransactionSynchronization synchronization = new AbstractTransactionalIoSourceTransactionSynchronization(); - - /** - * Register for Synchronization. This method is left protected because - * clients of this class should not be registering for synchronization, but - * rather only subclasses, at the appropriate time, i.e. when they are not - * initialized. - */ - protected void registerSynchronization() { - BatchTransactionSynchronizationManager - .registerSynchronization(synchronization); - } - /* * Called when a transaction has been committed. * @@ -85,20 +48,6 @@ public abstract class AbstractTransactionalIoSource extends ItemStreamAdapter { */ public abstract void reset(StreamContext streamContext); - /** - * Encapsulates transaction events handling. - */ - private class AbstractTransactionalIoSourceTransactionSynchronization extends - TransactionSynchronizationAdapter { - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - reset(null); - } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - mark(null); - } - } - } - /* (non-Javadoc) * @see org.springframework.batch.item.stream.ItemStreamAdapter#isMarkSupported() */ 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 1aa3ecbf6..f830eea84 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 @@ -20,13 +20,10 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.reader.AbstractItemReader; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessResourceFailureException; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.util.Assert; /** @@ -58,8 +55,6 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade private boolean initialized = false; - private TransactionSynchronization synchronization = new StaxEventReaderItemReaderTransactionSychronization(); - private long lastCommitPointRecordCount = 0; private long currentRecordCount = 0; @@ -116,8 +111,6 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade public void open() { Assert.state(resource.exists(), "Input resource does not exist: [" + resource + "]"); - - registerSynchronization(); try { inputStream = resource.getInputStream(); txReader = new DefaultTransactionalEventReader(XMLInputFactory.newInstance().createXMLEventReader( @@ -131,6 +124,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade throw new DataAccessResourceFailureException("Unable to get input stream", ioe); } initialized = true; + mark(null); } public void setResource(Resource resource) { @@ -216,7 +210,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade fragmentReader.next(); moveCursorToNextFragment(fragmentReader); } - txReader.onCommit(); // reset the history buffer + mark(null); // reset the history buffer } /** @@ -253,39 +247,10 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade } } - // package visibility method for simulating transaction events in tests - TransactionSynchronization getSynchronization() { - return synchronization; - } - - /** - * Encapsulates transaction events for the StaxEventReaderItemReader. - */ - private class StaxEventReaderItemReaderTransactionSychronization extends TransactionSynchronizationAdapter { - - /** - * @param status - * @see org.springframework.transaction.support.TransactionSynchronizationAdapter#afterCompletion(int) - */ - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_COMMITTED) { - mark(null); - } - else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - reset(null); - } - } - - } - public void destroy() throws Exception { close(); } - private void registerSynchronization() { - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - } - /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#isMarkSupported() */ 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 0e42beb73..44853109e 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 @@ -19,13 +19,10 @@ 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.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessResourceFailureException; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -94,9 +91,6 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing // XML event writer private XMLEventWriter delegateEventWriter; - // transaction synchronization object - private TransactionSynchronization synchronization = new StaxEventWriterItemWriterTransactionSychronization(); - // byte offset in file channel at last commit point private long lastCommitPointPosition = 0; @@ -228,13 +222,6 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing close(); } - /** - * Register the input source for transaction synchronization. - */ - private void registerSynchronization() { - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - } - /** * Open the output source * @@ -249,8 +236,6 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing */ private void open(long position) { - registerSynchronization(); - File file; FileOutputStream os = null; @@ -279,6 +264,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing } initialized = true; + } /** @@ -449,34 +435,6 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing } - /** - * Encapsulates transaction events for the StaxEventWriterOutputSource. - */ - private class StaxEventWriterItemWriterTransactionSychronization extends TransactionSynchronizationAdapter { - - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_COMMITTED) { - transactionComitted(); - } - else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - transactionRolledback(); - } - } - - private void transactionComitted() { - mark(null); - } - - private void transactionRolledback() { - reset(null); - } - - } - - TransactionSynchronization getSynchronization() { - return synchronization; - } - /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#isMarkSupported() */ 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 54aa9150d..12ebadd82 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 @@ -26,6 +26,12 @@ import java.util.Set; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.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; /** * Simple {@link StreamManager} that tries to resolve conflicts between key @@ -40,6 +46,31 @@ public class SimpleStreamManager implements StreamManager { private Map registry = new HashMap(); + private PlatformTransactionManager transactionManager; + + /** + * @param transactionManager a {@link PlatformTransactionManager} + */ + public SimpleStreamManager(PlatformTransactionManager transactionManager) { + this(); + this.transactionManager = transactionManager; + } + + /** + * + */ + public SimpleStreamManager() { + super(); + } + + /** + * Public setter for the {@link PlatformTransactionManager}. + * @param transactionManager the {@link PlatformTransactionManager} to set + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + /** * Simple aggregate statistics provider for the contributions registered * under the given key. @@ -67,7 +98,8 @@ public class SimpleStreamManager implements StreamManager { ItemStream provider = (ItemStream) iterator.next(); Properties properties = provider.getStreamContext().getProperties(); if (properties != null) { - String prefix = ""; // ClassUtils.getShortClassName(provider.getClass()) + "."; + String prefix = ""; // ClassUtils.getShortClassName(provider.getClass()) + // + "."; for (Iterator propiter = properties.keySet().iterator(); propiter.hasNext();) { String key = (String) propiter.next(); String value = properties.getProperty(key); @@ -82,7 +114,8 @@ public class SimpleStreamManager implements StreamManager { * Register a {@link ItemStream} as one of the interesting providers under * the provided key. * - * @see org.springframework.batch.item.stream.StreamManager#register(java.lang.Object, org.springframework.batch.item.ItemStream) + * @see org.springframework.batch.item.stream.StreamManager#register(java.lang.Object, + * org.springframework.batch.item.ItemStream) */ public void register(Object key, ItemStream provider) { synchronized (registry) { @@ -102,6 +135,46 @@ public class SimpleStreamManager implements StreamManager { * @see StreamManager#restoreFrom(Object, StreamContext) */ public void close(Object key) throws StreamException { + iterate(key, new Callback() { + public void execute(ItemStream stream) { + stream.close(); + } + }); + } + + /** + * Delegate to the {@link PlatformTransactionManager} to create a new + * transaction. + * + * @see org.springframework.batch.item.stream.StreamManager#getTransaction() + */ + 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) { + stream.mark(stream.getStreamContext()); + } + }); + } + else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + iterate(key, new Callback() { + public void execute(ItemStream stream) { + stream.reset(stream.getStreamContext()); + } + }); + } + } + }); + return transaction; + } + + /** + * @param key + */ + private void iterate(Object key, Callback callback) { Set set = new LinkedHashSet(); synchronized (registry) { Collection collection = (Collection) registry.get(key); @@ -111,8 +184,29 @@ public class SimpleStreamManager implements StreamManager { } for (Iterator iterator = set.iterator(); iterator.hasNext();) { ItemStream stream = (ItemStream) iterator.next(); - stream.close(); + callback.execute(stream); } } + /* + * (non-Javadoc) + * @see org.springframework.batch.item.stream.StreamManager#commit(java.lang.Object) + */ + public void commit(TransactionStatus status) { + transactionManager.commit(status); + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.stream.StreamManager#rollback(java.lang.Object) + */ + public void rollback(TransactionStatus status) { + transactionManager.rollback(status); + } + + private interface Callback { + void execute(ItemStream stream); + } + + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java index eea390488..fa1061e18 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 @@ -18,6 +18,7 @@ package org.springframework.batch.item.stream; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; +import org.springframework.transaction.TransactionStatus; /** * Generalised stream management broadcast strategy. Clients register @@ -58,5 +59,11 @@ public interface StreamManager { * been registered. */ void close(Object key) throws StreamException; + + TransactionStatus getTransaction(Object key); + + void commit(TransactionStatus transaction); + + void rollback(TransactionStatus transaction); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java deleted file mode 100644 index 81f15150f..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.repeat.synch; - -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; - -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.core.AttributeAccessor; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationManager; - -/** - *

- * Contains static methods for registering objects for transaction - * synchronization. Because there are many non-standard inputs that need to be - * aware of transactions, such as file input, this facade provides a hook into - * Spring TransactionSyncrhonization. The spring class - * TransactionSynchronizationManager has public static methods that are used by - * the AbstractPlatformTransactionManager to ensure other resources are - * synchronizaed with it's transaction. This means that any spring transaction - * manager which extends the afore mentioned abstract class will be notified of - * changes in a transaction. For more information on the type of transaction - * events that can be handled, please see the TransactionSynchronization - * interface. - *

- * - *

- * Spring's intended use for the TransactionSyncrhonizationManager is that any - * class that wishes can register for the current transaction only. When commit - * or rollback is called, this list will be used to notify interested classes. - * However, once a new transaction is obtained, this list will be cleared. This - * is problematic for batch processing, since input templates need to always be - * made aware of transaction events, without being forced to register every - * time. To solve this issue, classes should register with the - * BatchTransactionFacade, which will ensure that any time a new transaction is - * obtained, they are re-registered with spring's transaction synchronization. - *

- * - * @author Lucas Ward - * @author Dave Syer - * - * @see TransactionSynchronizationManager - * @see RepeatSynchronizationManager - * @see TransactionSynchronization - */ -public class BatchTransactionSynchronizationManager { - - /** - * The key in the context attributes for the list of synchronizations. - */ - private static final String SYNCHS_ATTR_KEY = BatchTransactionSynchronizationManager.class.getName() - + ".SYNCHRONIZATIONS"; - - /** - * Static method to register synchronizations. A TransactionSyncrhonization - * object will be added to the internal list within a threadLocal. After - * ensuring that there is a reference for later re-synchronization, the - * object is added to spring's TransactionSynchronizationManager. - */ - public static void registerSynchronization(TransactionSynchronization synchronization) { - List synchs = (List) getSynchronizations(); - - if (!synchs.contains(synchronization)) { - synchs.add(synchronization); - if (TransactionSynchronizationManager.isSynchronizationActive() - && !TransactionSynchronizationManager.getSynchronizations().contains(synchronization)) { - TransactionSynchronizationManager.registerSynchronization(synchronization); - } - } - - } - - /** - * The internal list of synchronizations is iterated, and each - * synchronization object is registered with the - * TransactionSynchronizationManager again. This is necessary because any - * call to PlatformTransactionManager.getTransaction() will result in a - * clearing of the synchronizationManager's list. - */ - public static void resynchronize() { - List batchSynchs = (List) getSynchronizations(); - - if (batchSynchs != null) { - for (Iterator it = batchSynchs.iterator(); it.hasNext();) { - TransactionSynchronization synchronization = (TransactionSynchronization) it.next(); - if (TransactionSynchronizationManager.isSynchronizationActive() - && !TransactionSynchronizationManager.getSynchronizations().contains(synchronization)) { - TransactionSynchronizationManager.registerSynchronization(synchronization); - } - } - } - } - - /** - * Set the synchronizations list to null. Usually called when the step is - * complete, to ensure no issues when the next step is called within the - * same thread, which should only happen when running out of container. Does - * not throw an exception if there is no batch context. - */ - public static void clearSynchronizations() { - AttributeAccessor context = getContext(); - if (context == null) { - return; // Nothing to do - } - setSynchronizations(null); - } - - /** - * Set the current synchronizations to the given list. - * @param synchs a list of {@link TransactionSynchronization} instances. - * @throws IllegalStateException if there is no batch context available. - */ - private static void setSynchronizations(List synchs) { - AttributeAccessor context = getContext(); - if (context == null) { - return; - } - context.setAttribute(SYNCHS_ATTR_KEY, synchs); - } - - /** - * Get the current list of synchronizations if there is one. - * - * @return a list of {@link TransactionSynchronization} instances or null. - * - * @throws IllegalStateException if there is no batch context available. - */ - private static List getSynchronizations() { - - AttributeAccessor context = getContext(); - if (context == null) { - // N.B. this returns a modifiable list on purpose - it is used - // internally to set up the list if there is no context available - // (useful in testing). - return new ArrayList(); - } - List synchs = (List) context.getAttribute(SYNCHS_ATTR_KEY); - - if (synchs == null) { - synchs = new ArrayList(); - } - - setSynchronizations(synchs); - - return synchs; - } - - /** - * @return the current context as an {@link AttributeAccessor}. - */ - private static AttributeAccessor getContext() { - RepeatContext context = RepeatSynchronizationManager.getContext(); - return getSynchContext(context); - } - - /** - * Locate the context that will contain the synchronisations by walking up - * the context hierarchy until the synchronisations are found or the top is - * reached. - * - * @param context - * @return - */ - private static AttributeAccessor getSynchContext(RepeatContext context) { - if (context == null || context.hasAttribute(SYNCHS_ATTR_KEY) || context.getParent() == null) { - return context; - } - return getSynchContext(context.getParent()); - } - -} 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 f72f97668..4cd7e78cf 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 @@ -12,9 +12,7 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.beans.factory.InitializingBean; -import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionSynchronizationUtils; import org.springframework.util.Assert; public class DrivingQueryItemReaderTests extends TestCase { @@ -150,15 +148,11 @@ public class DrivingQueryItemReaderTests extends TestCase { private void commit() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_COMMITTED); + ((ItemStream) source).mark(null); } private void rollback() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_ROLLED_BACK); + ((ItemStream) source).reset(null); } private InitializingBean getAsInitializingBean(ItemReader source) { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java index aea1eadae..3d08f986b 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java @@ -28,7 +28,6 @@ import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; -import org.springframework.transaction.support.TransactionSynchronization; /** * Tests for {@link DefaultFlatFileItemReader} @@ -98,13 +97,13 @@ public class DefaultFlatFileItemReaderTests extends TestCase { inputSource.read(); // #1 inputSource.read(); // #2 // commit them - inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); + inputSource.mark(null); // read next record inputSource.read(); // # 3 // mark record as skipped inputSource.skip(); // read next records - inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + inputSource.reset(null); // we should now process all records after first commit point, that are // not marked as skipped @@ -136,7 +135,7 @@ public class DefaultFlatFileItemReaderTests extends TestCase { // mark record as skipped inputSource.skip(); // rollback - inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + inputSource.reset(null); // read next record inputSource.read(); // should be #1 @@ -146,33 +145,6 @@ public class DefaultFlatFileItemReaderTests extends TestCase { } - /** - * Test skip and skipRollback functionality - * @throws IOException - */ - public void testTransactionSynchronizationUnknown() throws Exception { - - inputSource.close(); - inputSource.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - inputSource.open(); - - // read some records - inputSource.read(); - inputSource.skip(); - inputSource.read(); - - StreamContext statistics = inputSource.getStreamContext(); - long skipped = statistics.getLong(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME); - long read = statistics.getLong(DefaultFlatFileItemReader.READ_STATISTICS_NAME); - - // call unknown, which has no influence and therefore statistics should - // be the same - inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_UNKNOWN); - statistics = inputSource.getStreamContext();; - assertEquals(skipped, statistics.getLong(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME)); - assertEquals(read, statistics.getLong(DefaultFlatFileItemReader.READ_STATISTICS_NAME)); - } - public void testRestartFromNullData() throws Exception { inputSource.restoreFrom(null); assertEquals("[FlatFileInputTemplate-TestData]", inputSource.read().toString()); @@ -201,7 +173,7 @@ public class DefaultFlatFileItemReaderTests extends TestCase { inputSource.read(); inputSource.read(); // commit them - inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); + inputSource.mark(null); // read next two records inputSource.read(); inputSource.read(); 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 8c35a2730..8230cdb24 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 @@ -24,12 +24,11 @@ import java.util.Collections; import junit.framework.TestCase; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.writer.ItemTransformer; import org.springframework.core.io.FileSystemResource; -import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionSynchronizationUtils; /** * Tests of regular usage for {@link FlatFileItemWriter} Exception cases will be @@ -286,15 +285,6 @@ public class FlatFileItemWriterTests extends TestCase { assertEquals("testLine1", lineFromFile); } - public void testUnknown() throws Exception { - inputSource.write("testLine1"); - // rollback - unknown(); - inputSource.close(); - String lineFromFile = readLine(); - assertEquals("testLine1", lineFromFile); - } - public void testRestart() throws Exception { // write some lines @@ -380,17 +370,11 @@ public class FlatFileItemWriterTests extends TestCase { } private void commit() { - TransactionSynchronizationUtils.invokeAfterCompletion(TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_COMMITTED); + ((ItemStream) inputSource).mark(null); } private void rollback() { - TransactionSynchronizationUtils.invokeAfterCompletion(TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_ROLLED_BACK); + ((ItemStream) inputSource).reset(null); } - private void unknown() { - TransactionSynchronizationUtils.invokeAfterCompletion(TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_UNKNOWN); - } } 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 dd56f2bc7..2c35d9e1d 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 @@ -11,9 +11,6 @@ import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationMan import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionSynchronizationUtils; import org.springframework.util.Assert; /** @@ -159,15 +156,11 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra private void commit() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_COMMITTED); + ((ItemStream) source).mark(null); } private void rollback() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_ROLLED_BACK); + ((ItemStream) source).reset(null); } 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 e99ef2c55..638e971ba 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 @@ -12,9 +12,6 @@ import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationMan import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionSynchronizationUtils; import org.springframework.util.Assert; /** @@ -230,15 +227,11 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr } private void commit() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_COMMITTED); + ((ItemStream) source).mark(((ItemStream) source).getStreamContext()); } private void rollback() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_ROLLED_BACK); + ((ItemStream) source).reset(((ItemStream) source).getStreamContext()); } private Skippable getAsSkippable(ItemReader source) { 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 ba5e65fcc..25cf00f63 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 @@ -16,14 +16,10 @@ package org.springframework.batch.io.support; -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; import org.springframework.util.Assert; /** @@ -44,60 +40,18 @@ public class AbstractTransactionalIoSourceTests extends TestCase { TransactionSynchronizationManager.initSynchronization(); } - //AbstractItemReader should synchronize on first call to read. - public void testSynchronizationRegistration(){ - - source.registerSynchronization(); - - List synchronizations = (List)TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - } - public void testCommit(){ - - source.registerSynchronization(); - commit(); - + source.mark(null); assertTrue(source.commitCalled); assertFalse(source.rollbackCalled); } public void testRollback(){ - - source.registerSynchronization(); - - rollback(); - + source.reset(null); assertFalse(source.commitCalled); assertTrue(source.rollbackCalled); } - public void testCommitUnsynchronizedSource(){ - - commit(); - - assertFalse(source.commitCalled); - assertFalse(source.rollbackCalled); - } - - public void testMultipleSynchronizations(){ - - source.registerSynchronization(); - source.registerSynchronization(); - - //multiple calls to read should result in only one synchronization - List synchronizations = (List)TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - } - - public void testUnknownStatus(){ - - invokeUnknown(); - - assertFalse(source.commitCalled); - assertFalse(source.rollbackCalled); - } - private static class MockIoSource extends AbstractTransactionalIoSource { private boolean commitCalled = false; @@ -118,21 +72,4 @@ public class AbstractTransactionalIoSourceTests extends TestCase { } } - private void commit() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_COMMITTED); - } - - private void rollback() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_ROLLED_BACK); - } - - private void invokeUnknown() { - TransactionSynchronizationUtils.invokeAfterCompletion( - TransactionSynchronizationManager.getSynchronizations(), - TransactionSynchronization.STATUS_UNKNOWN); - } } 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 7e507fb3a..2d2a380ba 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 @@ -19,7 +19,6 @@ import org.springframework.core.io.AbstractResource; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessResourceFailureException; -import org.springframework.transaction.support.TransactionSynchronization; /** * Tests for {@link StaxEventItemReader}. @@ -157,7 +156,7 @@ public class StaxEventItemReaderTests extends TestCase { source.skip(); List second = (List) source.read(); assertFalse(first.equals(second)); - source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + source.reset(null); assertEquals(second, source.read()); } @@ -169,21 +168,21 @@ public class StaxEventItemReaderTests extends TestCase { // rollback between deserializing records List first = (List) source.read(); - source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); + source.mark(null); List second = (List) source.read(); assertFalse(first.equals(second)); - source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + source.reset(null); assertEquals(second, source.read()); // rollback while deserializing record - source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + source.reset(null); source.setFragmentDeserializer(new ExceptionFragmentDeserializer()); try { source.read(); } catch (Exception expected) { - source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + source.reset(null); } source.setFragmentDeserializer(deserializer); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java index 88a81dc89..feff19ab6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java @@ -17,7 +17,6 @@ import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.oxm.Marshaller; import org.springframework.oxm.XmlMappingException; -import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.util.Assert; import org.springframework.xml.transform.StaxResult; @@ -67,9 +66,8 @@ public class StaxEventWriterItemWriterTests extends TestCase { */ public void testRollback() throws Exception { writer.write(record); - // rollback - writer.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); + writer.reset(null); assertEquals("", outputFileContent()); } @@ -78,10 +76,9 @@ public class StaxEventWriterItemWriterTests extends TestCase { */ public void testCommit() throws Exception { writer.write(record); - // commit - writer.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); - assertTrue(outputFileContent().indexOf(TEST_STRING) != NOT_FOUND); + writer.mark(null); + assertTrue(outputFileContent().contains(TEST_STRING)); } /** @@ -90,7 +87,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { public void testRestart() throws Exception { // write records writer.write(record); - writer.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); + writer.mark(null); StreamContext streamContext = writer.getStreamContext(); // create new writer from saved restart data and continue writing @@ -134,7 +131,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { put("attribute", "value"); }}); writer.open(); - writer.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); + writer.mark(null); assertTrue(outputFileContent().indexOf(" + class="org.springframework.batch.item.reader.ValidatingItemReader">