Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)

Browse Source 
http://jira.springframework.org/browse/BATCH-7

Use StreamManager to start/end transaction in step executor
This commit is contained in:
dsyer
2008-01-31 22:36:26 +00:00
parent f5d8a32cdb
commit 2c3d43d7bd
27 changed files with 194 additions and 631 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java
View File

@@ -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();
}
}
}

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java
View File

@@ -321,7 +321,6 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
throw getExceptionTranslator().translate("Executing query", sql, se);
}
super.registerSynchronization();
}
/*

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -143,7 +143,6 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
+ ", call close() first.");
keys = keyGenerator.retrieveKeys();
keysIterator = keys.listIterator();
super.registerSynchronization();
initialized = true;
}

63
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java
View File

@@ -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;
/**
* <p>
@@ -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();
}
}
}
}

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java
View File

@@ -242,7 +242,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
*/
public void open() {
getOutputState();
super.registerSynchronization();
}
/**

51
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
View File

@@ -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;
/**
* <p>
@@ -34,43 +29,11 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
* maintained regardless of rollbacks.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*
* @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()
*/

39
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java
View File

@@ -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()
*/

44
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java
View File

@@ -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()
*/

100
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java
View File

@@ -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);
}
}

7
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java
View File

@@ -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);
}

187
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java
View File

@@ -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;
/**
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*
* @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());
}
}