@@ -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("