- * 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. - 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/sql/AbstractJdbcItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java index 2c35d9e1d..8089e48db 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 @@ -7,7 +7,6 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; 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.test.AbstractTransactionalDataSourceSpringContextTests; @@ -35,14 +34,12 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra protected void onSetUp()throws Exception{ super.onSetUp(); - BatchTransactionSynchronizationManager.clearSynchronizations(); source = createItemReader(); getAsInitializingBean(source).afterPropertiesSet(); } protected void onTearDown()throws Exception { getAsDisposableBean(source).destroy(); - BatchTransactionSynchronizationManager.clearSynchronizations(); super.onTearDown(); } 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 638e971ba..59f08ece9 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 @@ -8,7 +8,6 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; 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.test.AbstractTransactionalDataSourceSpringContextTests; @@ -39,7 +38,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr */ protected void onSetUpInTransaction() throws Exception { super.onSetUpInTransaction(); - BatchTransactionSynchronizationManager.clearSynchronizations(); source = createItemReader(); } @@ -48,7 +46,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr */ protected void onTearDownAfterTransaction() throws Exception { getAsDisposableBean(source).destroy(); - BatchTransactionSynchronizationManager.clearSynchronizations(); super.onTearDownAfterTransaction(); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManagerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManagerTests.java deleted file mode 100644 index 671f5694c..000000000 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManagerTests.java +++ /dev/null @@ -1,143 +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.List; - -import junit.framework.TestCase; - -import org.springframework.batch.repeat.context.RepeatContextSupport; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; -import org.springframework.transaction.support.TransactionSynchronizationManager; - -public class BatchTransactionSynchronizationManagerTests extends TestCase { - - private TransactionSynchronization synchronization = new TransactionSynchronizationAdapter() { - }; - - protected void setUp() throws Exception { - super.setUp(); - BatchTransactionSynchronizationManager.clearSynchronizations(); - if (TransactionSynchronizationManager.isSynchronizationActive()) { - TransactionSynchronizationManager.clearSynchronization(); - } - TransactionSynchronizationManager.initSynchronization(); - RepeatSynchronizationManager.register(new RepeatContextSupport(null)); - } - - /* - * (non-Javadoc) - * @see junit.framework.TestCase#tearDown() - */ - protected void tearDown() throws Exception { - super.tearDown(); - RepeatSynchronizationManager.clear(); - } - - public void testRegisterWhenContextMissing() throws Exception { - RepeatSynchronizationManager.clear(); - try { - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - } - catch (IllegalStateException e) { - fail("Unexpected IllegalStateException"); - // Unexpected - - } - } - - public void testRegisterSynchronization() { - - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - - // There should be only one transaction synchronization object in the - // list. - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - assertSame(synchronizations.get(0), synchronization); - - if (RepeatSynchronizationManager.getContext() != null) { - assertEquals(1, RepeatSynchronizationManager.getContext().attributeNames().length); - } - } - - public void testRegisterSynchronizationWithParentContext() { - - RepeatSynchronizationManager.register(new RepeatContextSupport(RepeatSynchronizationManager.getContext())); - - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - - // There should be only one transaction synchronization object in the - // list. - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - assertSame(synchronizations.get(0), synchronization); - - assertEquals(0, RepeatSynchronizationManager.getContext().attributeNames().length); - assertEquals(1, RepeatSynchronizationManager.getContext().getParent().attributeNames().length); - } - - public void testSynchronizeTwiceWithSameObject() throws Exception { - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - testRegisterSynchronization(); - } - - public void testSynchronizeTwiceWithSameObjectAndNoContext() throws Exception { - RepeatSynchronizationManager.clear(); - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - testRegisterSynchronization(); - } - - public void testReregisterSynchronization() { - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - TransactionSynchronizationManager.clearSynchronization(); - - TransactionSynchronizationManager.initSynchronization(); - BatchTransactionSynchronizationManager.resynchronize(); - - // There should be only one transaction synchronization object in the - // list. - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - assertSame(synchronizations.get(0), synchronization); - - } - - public void testResynchronizeWithNoSynchronizations() throws Exception { - BatchTransactionSynchronizationManager.resynchronize(); - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(0, synchronizations.size()); - } - - public void testSynchronizeWhenNotInTransaction() throws Exception { - - TransactionSynchronizationManager.clearSynchronization(); - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); - TransactionSynchronizationManager.initSynchronization(); - - BatchTransactionSynchronizationManager.resynchronize(); - - // There should be only one transaction synchronization object in the - // list. - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - assertEquals(1, synchronizations.size()); - assertSame(synchronizations.get(0), synchronization); - } - -} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java index c944ca3ac..9ff2943c3 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java @@ -11,8 +11,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.execution.scope.StepContext; import org.springframework.batch.execution.scope.StepContextAware; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.KeyedItemReader; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.item.StreamContext; import org.springframework.batch.sample.item.writer.StagingItemWriter; import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.OptimisticLockingFailureException; @@ -20,12 +21,10 @@ import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.support.JdbcDaoSupport; import org.springframework.jdbc.support.lob.DefaultLobHandler; import org.springframework.jdbc.support.lob.LobHandler; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; -public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader, DisposableBean, +public class StagingItemReader extends JdbcDaoSupport implements ItemStream, KeyedItemReader, DisposableBean, StepContextAware { // Key for buffer in transaction synchronization manager @@ -43,8 +42,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader private volatile Iterator keys; - private final TransactionSynchronization synchronization = new StagingInputTransactionSynchronization(); - /** * * @see org.springframework.batch.io.driving.DrivingQueryItemReader#close() @@ -151,8 +148,8 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader getBuffer().add(next); key = next; logger.debug("Retrieved key from list: " + key); - Assert.state(TransactionSynchronizationManager.getSynchronizations().contains(synchronization), - "Appropriate transaction synchronization not registered for this thread."); + Assert.state(TransactionSynchronizationManager.isActualTransactionActive(), + "Transaction not active for this thread."); } } } @@ -166,7 +163,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader private StagingBuffer getBuffer() { if (!TransactionSynchronizationManager.hasResource(BUFFER_KEY)) { TransactionSynchronizationManager.bindResource(BUFFER_KEY, new StagingBuffer()); - registerSynchronization(); } return (StagingBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY); } @@ -175,48 +171,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader return false; } - /** - * 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. - * - * @see TransactionSynchronization#afterCompletion - */ - protected void transactionCommitted() { - getBuffer().commit(); - } - - /* - * Called when a transaction has been rolled back. - * - * @see TransactionSynchronization#afterCompletion - */ - protected void transactionRolledBack() { - getBuffer().rollback(); - } - - /** - * Encapsulates transaction events handling. - */ - private class StagingInputTransactionSynchronization extends TransactionSynchronizationAdapter { - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - transactionRolledBack(); - } - else if (status == TransactionSynchronization.STATUS_COMMITTED) { - transactionCommitted(); - } - } - } - private class StagingBuffer { private List list = new ArrayList(); @@ -250,4 +204,44 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader } } + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#isMarkSupported() + */ + public boolean isMarkSupported() { + return true; + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext) + */ + public void mark(StreamContext streamContext) { + getBuffer().commit(); + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) + */ + public void reset(StreamContext streamContext) { + getBuffer().rollback(); + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext) + */ + public void restoreFrom(StreamContext context) { + // no-op + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.StreamContextProvider#getStreamContext() + */ + public StreamContext getStreamContext() { + return new StreamContext(); + } + } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java index 1e88f9cb8..927eba403 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java @@ -9,7 +9,6 @@ import org.springframework.batch.execution.scope.SimpleStepContext; import org.springframework.batch.execution.scope.StepContext; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.repeat.context.RepeatContextSupport; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; import org.springframework.batch.sample.item.writer.StagingItemWriter; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -87,12 +86,10 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin public void testProviderRollsBackMultipleTimes() throws Exception { + provider.mark(null); setComplete(); endTransaction(); startNewTransaction(); - // After a rollback we have to resynchronize the TX to simulate a real - // batch - BatchTransactionSynchronizationManager.resynchronize(); int count = getJdbcTemplate().queryForInt("SELECT COUNT(*) from BATCH_STAGING where JOB_ID=? AND PROCESSED=?", new Object[] { jobId, StagingItemWriter.NEW }); @@ -103,9 +100,9 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin item = provider.read(); assertEquals("BAR", item); + provider.reset(null); endTransaction(); startNewTransaction(); - BatchTransactionSynchronizationManager.resynchronize(); item = provider.read(); assertEquals("FOO", item); @@ -114,9 +111,9 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin item = provider.read(); assertEquals("SPAM", item); + provider.reset(null); endTransaction(); startNewTransaction(); - BatchTransactionSynchronizationManager.resynchronize(); item = provider.read(); assertEquals("FOO", item); @@ -125,12 +122,12 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin public void testProviderRollsBackProcessIndicator() throws Exception { + provider.mark(null); setComplete(); endTransaction(); startNewTransaction(); // After a rollback we have to resynchronize the TX to simulate a real // batch - BatchTransactionSynchronizationManager.resynchronize(); long id = getJdbcTemplate().queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", new Object[] { jobId }); @@ -141,11 +138,11 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin Object item = provider.read(); assertEquals("FOO", item); + provider.reset(null); endTransaction(); startNewTransaction(); // After a rollback we have to resynchronize the TX to simulate a real // batch - BatchTransactionSynchronizationManager.resynchronize(); String after = (String) getJdbcTemplate().queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", new Object[] { new Long(id) }, String.class);