diff --git a/infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorInputSource.java index ef3d9a85b..6952ea3ff 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorInputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorInputSource.java @@ -31,8 +31,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.io.InputSource; import org.springframework.batch.io.Skippable; +import org.springframework.batch.io.support.AbstractTransactionalIoSource; import org.springframework.batch.item.ResourceLifecycle; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.restart.GenericRestartData; import org.springframework.batch.restart.RestartData; import org.springframework.batch.restart.Restartable; @@ -47,8 +47,6 @@ import org.springframework.jdbc.support.JdbcUtils; import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator; import org.springframework.jdbc.support.SQLExceptionTranslator; import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.util.Assert; /** @@ -117,7 +115,7 @@ import org.springframework.util.Assert; * @author Lucas Ward * @author Peter Zozom */ -public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, DisposableBean, +public class JdbcCursorInputSource extends AbstractTransactionalIoSource implements InputSource, ResourceLifecycle, DisposableBean, InitializingBean, Restartable, StatisticsProvider, Skippable { private static Log log = LogFactory.getLog(JdbcCursorInputSource.class); @@ -159,8 +157,6 @@ public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, Di private int lastCommittedRow = 0; - private final SqlInputTransactionSynchronization transactionSynchronization = new SqlInputTransactionSynchronization(); - private RowMapper mapper; private boolean initialized = false; @@ -242,7 +238,7 @@ public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, Di * Mark the current row. Calling reset will cause the result set to be set * to the current row when mark was called. */ - private void mark() { + protected void transactionCommitted() { lastCommittedRow = currentProcessedRow; skippedRows.clear(); } @@ -252,7 +248,7 @@ public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, Di * * @throws DataAccessException */ - private void reset() { + protected void transactionRolledBack() { try { currentProcessedRow = lastCommittedRow; if (currentProcessedRow > 0) { @@ -334,8 +330,7 @@ public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, Di sql, se); } - BatchTransactionSynchronizationManager - .registerSynchronization(transactionSynchronization); + super.registerSynchronization(); } /* @@ -554,22 +549,4 @@ public class JdbcCursorInputSource implements InputSource, ResourceLifecycle, Di initialized = true; } - - private class SqlInputTransactionSynchronization extends TransactionSynchronizationAdapter { - - /* - * @param status transaction status - * - * @see org.springframework.transaction.support.TransactionSynchronization#afterCompletion(int) - */ - public void afterCompletion(int status) { - - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - reset(); - } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - mark(); - } - } - } - } diff --git a/infrastructure/src/main/java/org/springframework/batch/io/cursor/package.html b/infrastructure/src/main/java/org/springframework/batch/io/cursor/package.html new file mode 100644 index 000000000..c85e3934e --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/cursor/package.html @@ -0,0 +1,8 @@ + + +

+Infrastructure implementations of cursor based input sources. All input sources within this package +open a cursor against the database, and return back a mapped object for each row. +

+ + diff --git a/infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryInputSource.java index 1ea6bce8f..9d93f5705 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryInputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryInputSource.java @@ -19,6 +19,7 @@ import java.util.Iterator; import java.util.List; import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.support.AbstractTransactionalIoSource; import org.springframework.batch.item.ResourceLifecycle; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.restart.RestartData; @@ -44,10 +45,10 @@ import org.springframework.util.Assert; * * * @author Lucas Ward - * + * @since 1.0 */ -public class DrivingQueryInputSource implements InputSource, ResourceLifecycle, InitializingBean, - DisposableBean, Restartable { +public class DrivingQueryInputSource extends AbstractTransactionalIoSource implements InputSource, + ResourceLifecycle, InitializingBean, DisposableBean, Restartable { private boolean initialized = false; @@ -61,9 +62,6 @@ public class DrivingQueryInputSource implements InputSource, ResourceLifecycle, private KeyGenerator keyGenerator; - private TransactionSynchronization synchronization = - new DrivingQueryInputSourceTransactionSynchronization(); - public DrivingQueryInputSource() { } @@ -140,7 +138,7 @@ public class DrivingQueryInputSource implements InputSource, ResourceLifecycle, ", call close() first."); keys = keyGenerator.retrieveKeys(); keysIterator = keys.listIterator(); - BatchTransactionSynchronizationManager.registerSynchronization(synchronization); + super.registerSynchronization(); initialized = true; } @@ -186,19 +184,6 @@ public class DrivingQueryInputSource implements InputSource, ResourceLifecycle, return keyGenerator.getKeyAsRestartData(getCurrentKey()); } - /** - * Encapsulates transaction events handling. - */ - private class DrivingQueryInputSourceTransactionSynchronization extends TransactionSynchronizationAdapter { - public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - keysIterator = keys.listIterator(lastCommitIndex); - } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - lastCommitIndex = currentIndex; - } - } - } - public void afterPropertiesSet() throws Exception { Assert.notNull(keyGenerator, "The KeyGenerator must not be null."); } @@ -213,4 +198,12 @@ public class DrivingQueryInputSource implements InputSource, ResourceLifecycle, this.keyGenerator = keyGenerator; } + protected void transactionCommitted() { + lastCommitIndex = currentIndex; + } + + protected void transactionRolledBack() { + keysIterator = keys.listIterator(lastCommitIndex); + } + } diff --git a/infrastructure/src/main/java/org/springframework/batch/io/driving/package.html b/infrastructure/src/main/java/org/springframework/batch/io/driving/package.html new file mode 100644 index 000000000..f5be70a91 --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/driving/package.html @@ -0,0 +1,8 @@ + + +

+Infrastructure implementations of driving query based input sources. All input sources within this package +query a database to return a list of keys, and return back one key per call to read(). +

+ + diff --git a/infrastructure/src/main/java/org/springframework/batch/io/file/support/FlatFileOutputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/FlatFileOutputSource.java index 291501e1b..c41d675dd 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/file/support/FlatFileOutputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/FlatFileOutputSource.java @@ -32,6 +32,7 @@ import org.springframework.batch.io.OutputSource; import org.springframework.batch.io.exception.BatchCriticalException; import org.springframework.batch.io.exception.BatchEnvironmentException; import org.springframework.batch.io.file.support.transform.Converter; +import org.springframework.batch.io.support.AbstractTransactionalIoSource; import org.springframework.batch.item.ResourceLifecycle; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.restart.GenericRestartData; @@ -62,8 +63,9 @@ import org.springframework.util.Assert; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Restartable, StatisticsProvider, InitializingBean, - DisposableBean { +public class FlatFileOutputSource extends AbstractTransactionalIoSource implements + OutputSource, ResourceLifecycle, Restartable, StatisticsProvider, InitializingBean, + DisposableBean { /** * @author dsyer @@ -89,8 +91,6 @@ public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Re private RestartData restartData = new GenericRestartData(new Properties()); - private TransactionSynchronization transactionSynchronization = new FlatFileOutputTemplateTransactionSynchronization(); - private OutputState state = new OutputState(); private Converter converter = new Converter() { @@ -125,16 +125,16 @@ public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Re } /** - * Commit the transaction. + * Commit the transaction. */ - private void transactionComitted() { + protected void transactionCommitted() { getOutputState().mark(); } /** * Rollback the transaction. */ - private void transactionRolledback() { + protected void transactionRolledBack() { getOutputState().checkFileSize(); resetPositionForRestart(); } @@ -246,7 +246,7 @@ public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Re * @see ResourceLifecycle#open() */ public void open() { - registerSynchronization(); + super.registerSynchronization(); } /** @@ -281,22 +281,11 @@ public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Re } - // Registers a new transaction synchronization for the current thread. - private void registerSynchronization() { - BatchTransactionSynchronizationManager.registerSynchronization(this.transactionSynchronization); - } - // Returns object representing state. private OutputState getOutputState() { return (OutputState) state; } - // added package visibility method so that tests can invoke transaction - // events - TransactionSynchronization getTransactionSynchronization() { - return this.transactionSynchronization; - } - /** * Encapsulates the runtime state of the output source. All state changing * operations on the output source go through this class. @@ -554,25 +543,4 @@ public class FlatFileOutputSource implements OutputSource, ResourceLifecycle, Re } } - - /** - * Encapsulates transaction events. - */ - private class FlatFileOutputTemplateTransactionSynchronization 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/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventReaderInputSource.java similarity index 92% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventReaderInputSource.java index 12207646c..d90f9933f 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventReaderInputSource.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support; import java.io.IOException; import java.io.InputStream; @@ -14,6 +14,11 @@ import javax.xml.stream.events.StartElement; import org.springframework.batch.io.InputSource; import org.springframework.batch.io.Skippable; +import org.springframework.batch.io.file.support.stax.DefaultFragmentEventReader; +import org.springframework.batch.io.file.support.stax.DefaultTransactionalEventReader; +import org.springframework.batch.io.file.support.stax.FragmentDeserializer; +import org.springframework.batch.io.file.support.stax.FragmentEventReader; +import org.springframework.batch.io.file.support.stax.TransactionalEventReader; import org.springframework.batch.item.ResourceLifecycle; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.batch.restart.GenericRestartData; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSource.java similarity index 94% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSource.java index e1b639987..45534bb53 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSource.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support; import java.io.File; import java.io.FileOutputStream; @@ -15,6 +15,8 @@ import javax.xml.stream.XMLOutputFactory; import javax.xml.stream.XMLStreamException; import org.springframework.batch.io.OutputSource; +import org.springframework.batch.io.file.support.stax.NoStartEndDocumentStreamWriter; +import org.springframework.batch.io.file.support.stax.ObjectToXmlSerializer; import org.springframework.batch.io.support.FileUtils; import org.springframework.batch.item.ResourceLifecycle; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializer.java similarity index 84% rename from infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializer.java index 8ee8b5444..a60d00eb4 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializer.java @@ -1,11 +1,11 @@ -package org.springframework.batch.io.oxm; +package org.springframework.batch.io.file.support.oxm; import java.io.IOException; import javax.xml.stream.XMLEventWriter; import javax.xml.transform.Result; -import org.springframework.batch.io.stax.ObjectToXmlSerializer; +import org.springframework.batch.io.file.support.stax.ObjectToXmlSerializer; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.oxm.Marshaller; import org.springframework.xml.transform.StaxResult; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializer.java similarity index 85% rename from infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializer.java index 6f025419f..6543027a6 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializer.java @@ -1,10 +1,10 @@ -package org.springframework.batch.io.oxm; +package org.springframework.batch.io.file.support.oxm; import java.io.IOException; import javax.xml.stream.XMLEventReader; -import org.springframework.batch.io.stax.FragmentDeserializer; +import org.springframework.batch.io.file.support.stax.FragmentDeserializer; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.oxm.Unmarshaller; import org.springframework.util.Assert; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapper.java similarity index 92% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapper.java index 782790c21..4d8bcfa4d 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapper.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; import javax.xml.stream.XMLStreamException; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventWriterWrapper.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapper.java similarity index 93% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventWriterWrapper.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapper.java index ddad90987..506853242 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventWriterWrapper.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapper.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.namespace.NamespaceContext; import javax.xml.stream.XMLEventReader; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReader.java similarity index 93% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReader.java index 8ba235473..5a87b5c5b 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReader.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import java.util.NoSuchElementException; @@ -19,7 +19,7 @@ import org.springframework.dao.DataAccessResourceFailureException; * * @author Robert Kasanicky */ -class DefaultFragmentEventReader extends AbstractEventReaderWrapper implements FragmentEventReader { +public class DefaultFragmentEventReader extends AbstractEventReaderWrapper implements FragmentEventReader { // true when the next event is the StartElement of next fragment private boolean startFragmentFollows = false; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReader.java similarity index 88% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReader.java index c4bbf9531..81c990672 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReader.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import java.util.NoSuchElementException; @@ -16,7 +16,7 @@ import org.springframework.util.Assert; * @author Tomas Slanina * @author Robert Kasanicky */ -class DefaultTransactionalEventReader extends AbstractEventReaderWrapper implements TransactionalEventReader, InitializingBean { +public class DefaultTransactionalEventReader extends AbstractEventReaderWrapper implements TransactionalEventReader, InitializingBean { private EventSequence recorder = new EventSequence(); diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/EventSequence.java similarity index 91% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/EventSequence.java index 05a84b0ec..f6e6ad57c 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/EventSequence.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import java.util.ArrayList; import java.util.List; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentDeserializer.java similarity index 81% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentDeserializer.java index 7db2506c9..72d34a261 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentDeserializer.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentEventReader.java similarity index 83% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentEventReader.java index d9e4c09ee..a3db4058b 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/FragmentEventReader.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; @@ -9,7 +9,7 @@ import javax.xml.stream.XMLEventReader; * * @author Robert Kasanicky */ -interface FragmentEventReader extends XMLEventReader { +public interface FragmentEventReader extends XMLEventReader { /** * Tells the event reader its cursor position is exactly before the fragment. diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentStreamWriter.java similarity index 78% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentStreamWriter.java index 7b5bb6dae..75c6fdb99 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentStreamWriter.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventWriter; import javax.xml.stream.XMLStreamException; @@ -11,7 +11,7 @@ import javax.xml.stream.events.XMLEvent; * @author peter.zozom * @author Robert Kasanicky */ -class NoStartEndDocumentStreamWriter extends AbstractEventWriterWrapper { +public class NoStartEndDocumentStreamWriter extends AbstractEventWriterWrapper { public NoStartEndDocumentStreamWriter(XMLEventWriter wrappedEventWriter) { super(wrappedEventWriter); diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/ObjectToXmlSerializer.java similarity index 86% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/ObjectToXmlSerializer.java index 5ac416078..7b951d19a 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/ObjectToXmlSerializer.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventWriter; diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/TransactionalEventReader.java similarity index 69% rename from infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java rename to infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/TransactionalEventReader.java index 2f75b67fd..001634e75 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/file/support/stax/TransactionalEventReader.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; @@ -7,7 +7,7 @@ import javax.xml.stream.XMLEventReader; * * @author Robert Kasanicky */ -interface TransactionalEventReader extends XMLEventReader{ +public interface TransactionalEventReader extends XMLEventReader{ /** * Callback on transaction rollback. diff --git a/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java b/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java new file mode 100644 index 000000000..96ae95eff --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java @@ -0,0 +1,89 @@ +/* + * 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.io.support; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.OutputSource; +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; + +/** + *

Abstract class that abstracts away transaction handling from input and output sources. + * Since every {@link InputSource} or {@link OutputSource by nature wants to be notified of + * transaction events to maintain the contract that all calls to read or write will ensure + * that correct ordering is 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 + * symantics 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 { + + private final TransactionSynchronization synchronization = + new InputSourceTransactionSynchronization(); + + /** + * 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 abstract void transactionCommitted(); + + /* + * Called when a transaction has been rolled back. + * + * @see TransactionSynchronization#afterCompletion + */ + protected abstract void transactionRolledBack(); + + /** + * Encapsulates transaction events handling. + */ + private class InputSourceTransactionSynchronization extends TransactionSynchronizationAdapter { + public void afterCompletion(int status) { + if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + transactionRolledBack(); + } else if (status == TransactionSynchronization.STATUS_COMMITTED) { + transactionCommitted(); + } + } + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/cursor/SqlCursorInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorInputSourceIntegrationTests.java similarity index 78% rename from infrastructure/src/test/java/org/springframework/batch/io/cursor/SqlCursorInputSourceIntegrationTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorInputSourceIntegrationTests.java index 570860b8d..ebe98bb26 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/cursor/SqlCursorInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/cursor/JdbcCursorInputSourceIntegrationTests.java @@ -3,14 +3,14 @@ package org.springframework.batch.io.cursor; import org.springframework.batch.io.InputSource; import org.springframework.batch.io.cursor.JdbcCursorInputSource; import org.springframework.batch.io.driving.FooRowMapper; -import org.springframework.batch.io.sql.AbstractSqlInputSourceIntegrationTests; +import org.springframework.batch.io.sql.AbstractJdbcInputSourceIntegrationTests; /** * Tests for {@link JdbcCursorInputSource} * * @author Robert Kasanicky */ -public class SqlCursorInputSourceIntegrationTests extends AbstractSqlInputSourceIntegrationTests{ +public class JdbcCursorInputSourceIntegrationTests extends AbstractJdbcInputSourceIntegrationTests{ protected InputSource createInputSource() throws Exception { JdbcCursorInputSource result = new JdbcCursorInputSource(); diff --git a/infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryInputSourceIntegrationTests.java index e359b02bc..1f16b3fb4 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/driving/MultipleColumnJdbcDrivingQueryInputSourceIntegrationTests.java @@ -17,14 +17,14 @@ package org.springframework.batch.io.driving; import org.springframework.batch.io.InputSource; import org.springframework.batch.io.driving.support.MultipleColumnJdbcKeyGenerator; -import org.springframework.batch.io.sql.AbstractSqlInputSourceIntegrationTests; +import org.springframework.batch.io.sql.AbstractJdbcInputSourceIntegrationTests; /** * @author Lucas Ward * */ public class MultipleColumnJdbcDrivingQueryInputSourceIntegrationTests extends - AbstractSqlInputSourceIntegrationTests { + AbstractJdbcInputSourceIntegrationTests { protected InputSource createInputSource() throws Exception { diff --git a/infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryInputSourceIntegrationTests.java index 6b39faaa5..65044c36f 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/driving/SingleColumnJdbcDrivingQueryInputSourceIntegrationTests.java @@ -2,9 +2,9 @@ package org.springframework.batch.io.driving; import org.springframework.batch.io.InputSource; import org.springframework.batch.io.driving.support.SingleColumnJdbcKeyGenerator; -import org.springframework.batch.io.sql.AbstractSqlInputSourceIntegrationTests; +import org.springframework.batch.io.sql.AbstractJdbcInputSourceIntegrationTests; -public class SingleColumnJdbcDrivingQueryInputSourceIntegrationTests extends AbstractSqlInputSourceIntegrationTests { +public class SingleColumnJdbcDrivingQueryInputSourceIntegrationTests extends AbstractJdbcInputSourceIntegrationTests { protected InputSource source; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/file/support/FieldSetTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/FieldSetTests.java similarity index 99% rename from infrastructure/src/test/java/org/springframework/batch/io/file/support/FieldSetTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/FieldSetTests.java index a91a1e2ec..9cd7662db 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/file/support/FieldSetTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/FieldSetTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.batch.io.file.support; +package org.springframework.batch.io.file; import java.math.BigDecimal; import java.text.ParseException; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/EventHelper.java similarity index 85% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/EventHelper.java index 14223c326..e35e440ec 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/EventHelper.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support; import javax.xml.stream.events.EndElement; import javax.xml.stream.events.StartElement; @@ -9,7 +9,7 @@ import javax.xml.stream.events.XMLEvent; * * @author Robert Kasanicky */ -class EventHelper { +public class EventHelper { //utility class private EventHelper() {} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/file/support/FlatFileOutputSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/FlatFileOutputSourceTests.java index 392981292..41f85d0f1 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/file/support/FlatFileOutputSourceTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/FlatFileOutputSourceTests.java @@ -28,6 +28,8 @@ import org.springframework.batch.io.file.support.transform.Converter; import org.springframework.batch.restart.RestartData; 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 FlatFileOutputSource} Exception cases will @@ -41,7 +43,7 @@ import org.springframework.transaction.support.TransactionSynchronization; public class FlatFileOutputSourceTests extends TestCase { // object under test - private FlatFileOutputSource template = new FlatFileOutputSource(); + private FlatFileOutputSource inputSource = new FlatFileOutputSource(); // String to be written into file by the FlatFileInputTemplate private static final String TEST_STRING = "FlatFileOutputTemplateTest-OutputData"; @@ -58,12 +60,17 @@ public class FlatFileOutputSourceTests extends TestCase { */ protected void setUp() throws Exception { + if(TransactionSynchronizationManager.isSynchronizationActive()){ + TransactionSynchronizationManager.clearSynchronization(); + } + TransactionSynchronizationManager.initSynchronization(); + outputFile = File.createTempFile("flatfile-output-", ".tmp"); - template.setResource(new FileSystemResource(outputFile)); - template.afterPropertiesSet(); + inputSource.setResource(new FileSystemResource(outputFile)); + inputSource.afterPropertiesSet(); - template.open(); + inputSource.open(); } @@ -74,7 +81,7 @@ public class FlatFileOutputSourceTests extends TestCase { if( reader != null){ reader.close(); } - template.close(); + inputSource.close(); outputFile.delete(); } @@ -97,8 +104,8 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteString() throws IOException { - template.write(TEST_STRING); - template.close(); + inputSource.write(TEST_STRING); + inputSource.close(); String lineFromFile = readLine(); assertEquals(TEST_STRING, lineFromFile); @@ -108,8 +115,8 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteCollection() throws IOException { - template.write(Collections.singleton(TEST_STRING)); - template.close(); + inputSource.write(Collections.singleton(TEST_STRING)); + inputSource.close(); String lineFromFile = readLine(); assertEquals(TEST_STRING, lineFromFile); } @@ -118,14 +125,14 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverter() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { public Object convert(Object input) { return "FOO:" + input; } }); Object data = new Object(); - template.write(data); - template.close(); + inputSource.write(data); + inputSource.close(); String lineFromFile = readLine(); // converter not used if input is String assertEquals("FOO:" + data.toString(), lineFromFile); @@ -135,14 +142,14 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverterAndInfiniteLoop() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { public Object convert(Object input) { return "FOO:" + input; } }); Object data = new Object(); - template.write(data); - template.close(); + inputSource.write(data); + inputSource.close(); String lineFromFile = readLine(); // converter not used if input is String assertEquals("FOO:" + data.toString(), lineFromFile); @@ -152,14 +159,14 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverterAndInfiniteLoopInCollection() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { public Object convert(Object input) { return "FOO:" + input; } }); Object data = new Object(); - template.write(new Object[] {data, data}); - template.close(); + inputSource.write(new Object[] {data, data}); + inputSource.close(); String lineFromFile = readLine(); assertEquals("FOO:" + data.toString(), lineFromFile); lineFromFile = readLine(); @@ -170,7 +177,7 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverterAndInfiniteLoopInConvertedCollection() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { boolean converted = false; public Object convert(Object input) { if (converted) { @@ -182,13 +189,13 @@ public class FlatFileOutputSourceTests extends TestCase { }); Object data = new Object(); try { - template.write(data); + inputSource.write(data); fail("Expected IllegalStateException"); } catch (IllegalStateException e) { // expected assertTrue("Wrong message: "+e, e.getMessage().toLowerCase().indexOf("infinite")>=0); } - template.close(); + inputSource.close(); String lineFromFile = readLine(); assertNull(lineFromFile); } @@ -197,13 +204,13 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverterAndString() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { public Object convert(Object input) { return "FOO:" + input; } }); - template.write(Collections.singleton(TEST_STRING)); - template.close(); + inputSource.write(Collections.singleton(TEST_STRING)); + inputSource.close(); String lineFromFile = readLine(); // converter not used if input is String assertEquals(TEST_STRING, lineFromFile); @@ -213,13 +220,13 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteWithConverterAndCollectionOfString() throws IOException { - template.setConverter(new Converter() { + inputSource.setConverter(new Converter() { public Object convert(Object input) { return "FOO:" + input; } }); - template.write(TEST_STRING); - template.close(); + inputSource.write(TEST_STRING); + inputSource.close(); String lineFromFile = readLine(); // converter not used if input is String assertEquals(TEST_STRING, lineFromFile); @@ -229,8 +236,8 @@ public class FlatFileOutputSourceTests extends TestCase { * Regular usage of write(String) method */ public void testWriteArray() throws IOException { - template.write(new String[] { TEST_STRING, TEST_STRING }); - template.close(); + inputSource.write(new String[] { TEST_STRING, TEST_STRING }); + inputSource.close(); String lineFromFile = readLine(); assertEquals(TEST_STRING, lineFromFile); lineFromFile = readLine(); @@ -244,35 +251,35 @@ public class FlatFileOutputSourceTests extends TestCase { String args = "1"; // AggregatorStub ignores the LineDescriptor, so we pass null - template.write(args); - template.close(); + inputSource.write(args); + inputSource.close(); String lineFromFile = readLine(); assertEquals(args, lineFromFile); } public void testRollback() throws Exception { - template.write("testLine1"); + inputSource.write("testLine1"); // rollback - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); - template.close(); + rollback(); + inputSource.close(); String lineFromFile = readLine(); assertEquals(null, lineFromFile); } public void testCommit() throws Exception { - template.write("testLine1"); + inputSource.write("testLine1"); // rollback - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); - template.close(); + commit(); + inputSource.close(); String lineFromFile = readLine(); assertEquals("testLine1", lineFromFile); } public void testUnknown() throws Exception { - template.write("testLine1"); + inputSource.write("testLine1"); // rollback - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_UNKNOWN); - template.close(); + unknown(); + inputSource.close(); String lineFromFile = readLine(); assertEquals("testLine1", lineFromFile); } @@ -280,38 +287,38 @@ public class FlatFileOutputSourceTests extends TestCase { public void testRestart() throws IOException { // write some lines - template.write("testLine1"); - template.write("testLine2"); - template.write("testLine3"); + inputSource.write("testLine1"); + inputSource.write("testLine2"); + inputSource.write("testLine3"); // commit - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); - + commit(); + // this will be rolled back... - template.write("this will be rolled back"); + inputSource.write("this will be rolled back"); // rollback - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); - + rollback(); + // write more lines - template.write("testLine4"); - template.write("testLine5"); + inputSource.write("testLine4"); + inputSource.write("testLine5"); // commit - template.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); - + commit(); + // get restart data - RestartData restartData = template.getRestartData(); + RestartData restartData = inputSource.getRestartData(); // close template - template.close(); + inputSource.close(); // init for restart - template.setBufferSize(0); - template.open(); + inputSource.setBufferSize(0); + inputSource.open(); // try empty restart data... try { - template.restoreFrom(null); + inputSource.restoreFrom(null); assertTrue(true); } catch (IllegalArgumentException iae) { @@ -319,15 +326,15 @@ public class FlatFileOutputSourceTests extends TestCase { } // init with correct data - template.restoreFrom(restartData); + inputSource.restoreFrom(restartData); // write more lines - template.write("testLine6"); - template.write("testLine7"); - template.write("testLine8"); + inputSource.write("testLine6"); + inputSource.write("testLine7"); + inputSource.write("testLine8"); // close template - template.close(); + inputSource.close(); // verify what was written to the file for (int i = 1; i < 9; i++) { @@ -344,9 +351,9 @@ public class FlatFileOutputSourceTests extends TestCase { } public void testAfterPropertiesSetChecksMandatory() throws Exception { - template = new FlatFileOutputSource(); + inputSource = new FlatFileOutputSource(); try { - template.afterPropertiesSet(); + inputSource.afterPropertiesSet(); fail("Expected IllegalArgumentException"); } catch (IllegalArgumentException e) { @@ -355,10 +362,28 @@ public class FlatFileOutputSourceTests extends TestCase { } public void testDefaultRestartData() throws Exception { - template = new FlatFileOutputSource(); - RestartData restartData = template.getRestartData(); + inputSource = new FlatFileOutputSource(); + RestartData restartData = inputSource.getRestartData(); assertNotNull(restartData); // TODO: assert the properties of the default restart data assertEquals(1, restartData.getProperties().size()); } + + private void commit() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_COMMITTED); + } + + private void rollback() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_ROLLED_BACK); + } + + private void unknown() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_UNKNOWN); + } } diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventReaderInputSourceTests.java similarity index 94% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventReaderInputSourceTests.java index 840b764dd..002aaf715 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventReaderInputSourceTests.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support; import java.io.IOException; import java.io.InputStream; @@ -15,6 +15,8 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; +import org.springframework.batch.io.file.support.StaxEventReaderInputSource; +import org.springframework.batch.io.file.support.stax.FragmentDeserializer; import org.springframework.batch.restart.GenericRestartData; import org.springframework.batch.restart.RestartData; import org.springframework.core.io.AbstractResource; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSourceTests.java similarity index 93% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSourceTests.java index e973d5405..cbb6fd0ab 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/StaxEventWriterOutputSourceTests.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support; import java.io.File; import java.io.IOException; @@ -11,7 +11,8 @@ import javax.xml.transform.Result; import junit.framework.TestCase; import org.apache.commons.io.FileUtils; -import org.springframework.batch.io.oxm.MarshallingObjectToXmlSerializer; +import org.springframework.batch.io.file.support.StaxEventWriterOutputSource; +import org.springframework.batch.io.file.support.oxm.MarshallingObjectToXmlSerializer; import org.springframework.batch.restart.RestartData; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializerTests.java similarity index 92% rename from infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializerTests.java index 06e5cc5fc..b9ae18f6d 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/MarshallingObjectToXmlSerializerTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.batch.io.oxm; +package org.springframework.batch.io.file.support.oxm; import java.io.IOException; @@ -26,6 +26,7 @@ import javax.xml.transform.Result; import junit.framework.TestCase; +import org.springframework.batch.io.file.support.oxm.MarshallingObjectToXmlSerializer; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.oxm.Marshaller; import org.springframework.oxm.XmlMappingException; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializerTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializerTests.java similarity index 91% rename from infrastructure/src/test/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializerTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializerTests.java index f1a92ba36..2e73dd3cb 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializerTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/oxm/UnmarshallingFragmentDeserializerTests.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.oxm; +package org.springframework.batch.io.file.support.oxm; import java.io.IOException; @@ -8,6 +8,7 @@ import javax.xml.stream.XMLInputFactory; import junit.framework.TestCase; import org.easymock.MockControl; +import org.springframework.batch.io.file.support.oxm.UnmarshallingFragmentDeserializer; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessException; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventReaderWrapperTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapperTests.java similarity index 93% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventReaderWrapperTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapperTests.java index 812642e90..e0358f6c7 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventReaderWrapperTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventReaderWrapperTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; import javax.xml.stream.XMLStreamException; @@ -22,6 +22,7 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; import org.easymock.MockControl; +import org.springframework.batch.io.file.support.stax.AbstractEventReaderWrapper; import com.bea.xml.stream.events.StartDocumentEvent; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventWriterWrapperTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapperTests.java similarity index 93% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventWriterWrapperTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapperTests.java index d7d7165e5..5140e1194 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/AbstractEventWriterWrapperTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/AbstractEventWriterWrapperTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.namespace.NamespaceContext; import javax.xml.stream.XMLEventReader; @@ -24,6 +24,7 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; import org.easymock.MockControl; +import org.springframework.batch.io.file.support.stax.AbstractEventWriterWrapper; import com.bea.xml.stream.events.StartDocumentEvent; import com.bea.xml.stream.util.NamespaceContextImpl; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReaderTests.java similarity index 90% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReaderTests.java index b5bbcd27e..f6fbfc1ea 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultFragmentEventReaderTests.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import java.util.NoSuchElementException; @@ -9,6 +9,10 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; +import org.springframework.batch.io.file.support.EventHelper; +import org.springframework.batch.io.file.support.stax.DefaultFragmentEventReader; +import org.springframework.batch.io.file.support.stax.DefaultTransactionalEventReader; +import org.springframework.batch.io.file.support.stax.FragmentEventReader; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReaderTests.java similarity index 84% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReaderTests.java index 5e2ad459d..00fd301d7 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/DefaultTransactionalEventReaderTests.java @@ -1,10 +1,13 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventReader; import javax.xml.stream.XMLInputFactory; import junit.framework.TestCase; +import org.springframework.batch.io.file.support.EventHelper; +import org.springframework.batch.io.file.support.stax.DefaultTransactionalEventReader; +import org.springframework.batch.io.file.support.stax.TransactionalEventReader; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/EventSequenceTests.java similarity index 83% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/EventSequenceTests.java index 7a96d6552..c86ae5d59 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/EventSequenceTests.java @@ -1,8 +1,10 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventFactory; import javax.xml.stream.events.XMLEvent; +import org.springframework.batch.io.file.support.stax.EventSequence; + import junit.framework.TestCase; /** diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/NoStartEndDocumentWriterTests.java b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentWriterTests.java similarity index 87% rename from infrastructure/src/test/java/org/springframework/batch/io/stax/NoStartEndDocumentWriterTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentWriterTests.java index 5607863ad..3c787f643 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/stax/NoStartEndDocumentWriterTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/file/support/stax/NoStartEndDocumentWriterTests.java @@ -1,4 +1,4 @@ -package org.springframework.batch.io.stax; +package org.springframework.batch.io.file.support.stax; import javax.xml.stream.XMLEventFactory; import javax.xml.stream.XMLEventWriter; @@ -7,6 +7,7 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; import org.easymock.MockControl; +import org.springframework.batch.io.file.support.stax.NoStartEndDocumentStreamWriter; /** * Tests for {@link NoStartEndDocumentStreamWriter} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcInputSourceIntegrationTests.java similarity index 94% rename from infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcInputSourceIntegrationTests.java index 26b1872bd..f0c2d4fe9 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcInputSourceIntegrationTests.java @@ -22,7 +22,7 @@ import org.springframework.util.Assert; * @author Lucas Ward * @author Robert Kasanicky */ -public abstract class AbstractSqlInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { +public abstract class AbstractJdbcInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { protected InputSource source; diff --git a/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java new file mode 100644 index 000000000..cf8c62767 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractTransactionalIoSourceTests.java @@ -0,0 +1,133 @@ +/* + * 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.io.support; + +import java.util.List; + +import junit.framework.TestCase; + +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionSynchronizationUtils; +import org.springframework.util.Assert; + +/** + * @author Lucas Ward + * + */ +public class AbstractTransactionalIoSourceTests extends TestCase { + + private MockIoSource source; + + protected void setUp() throws Exception { + super.setUp(); + + source = new MockIoSource(); + if(TransactionSynchronizationManager.isSynchronizationActive()){ + TransactionSynchronizationManager.clearSynchronization(); + } + TransactionSynchronizationManager.initSynchronization(); + } + + //AbstractInputSource 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(); + + assertTrue(source.commitCalled); + assertFalse(source.rollbackCalled); + } + + public void testRollback(){ + + source.registerSynchronization(); + + rollback(); + + 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; + private boolean rollbackCalled = false; + + protected void transactionCommitted() { + Assert.isTrue(!commitCalled, "Commit aleady called"); + commitCalled = true; + } + + protected void transactionRolledBack() { + Assert.isTrue(!rollbackCalled, "Rollback aleady called"); + rollbackCalled = true; + } + } + + 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/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventReaderInputSourceTests.java b/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventReaderInputSourceTests.java index 9b3879227..5db2e9d4f 100644 --- a/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventReaderInputSourceTests.java +++ b/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventReaderInputSourceTests.java @@ -6,8 +6,9 @@ import java.util.List; import junit.framework.TestCase; +import org.springframework.batch.io.file.support.StaxEventReaderInputSource; +import org.springframework.batch.io.file.support.oxm.UnmarshallingFragmentDeserializer; import org.springframework.batch.io.oxm.domain.Trade; -import org.springframework.batch.io.stax.StaxEventReaderInputSource; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; import org.springframework.oxm.Unmarshaller; diff --git a/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventWriterOutputSourceTests.java b/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventWriterOutputSourceTests.java index fbd282ba2..f88945318 100644 --- a/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventWriterOutputSourceTests.java +++ b/integration/src/test/java/org/springframework/batch/io/oxm/AbstractStaxEventWriterOutputSourceTests.java @@ -11,8 +11,9 @@ import junit.framework.TestCase; import org.custommonkey.xmlunit.XMLAssert; import org.custommonkey.xmlunit.XMLUnit; +import org.springframework.batch.io.file.support.StaxEventWriterOutputSource; +import org.springframework.batch.io.file.support.oxm.MarshallingObjectToXmlSerializer; import org.springframework.batch.io.oxm.domain.Trade; -import org.springframework.batch.io.stax.StaxEventWriterOutputSource; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; diff --git a/samples/20070122.testStream.ParallelCustomerReportStep.TEMP.txt b/samples/20070122.testStream.ParallelCustomerReportStep.TEMP.txt index 5e1f0b317..e1c0cdf5a 100644 --- a/samples/20070122.testStream.ParallelCustomerReportStep.TEMP.txt +++ b/samples/20070122.testStream.ParallelCustomerReportStep.TEMP.txt @@ -1,5 +1,5 @@ -Trade: [isin=UK21341EAH41,quantity=211,price=31.11,customer=customer1] -Trade: [isin=UK21341EAH42,quantity=212,price=32.11,customer=customer2] -Trade: [isin=UK21341EAH43,quantity=213,price=33.11,customer=customer3] -Trade: [isin=UK21341EAH44,quantity=214,price=34.11,customer=customer4] -Trade: [isin=UK21341EAH45,quantity=215,price=35.11,customer=customer5] +Trade: [isin=UK21341EAH41,quantity=211,price=31.11,customer=customer1] +Trade: [isin=UK21341EAH42,quantity=212,price=32.11,customer=customer2] +Trade: [isin=UK21341EAH43,quantity=213,price=33.11,customer=customer3] +Trade: [isin=UK21341EAH44,quantity=214,price=34.11,customer=customer4] +Trade: [isin=UK21341EAH45,quantity=215,price=35.11,customer=customer5] diff --git a/samples/20070122.testStream.multilineStep.txt b/samples/20070122.testStream.multilineStep.txt index e26cadbc9..72a5c17e7 100644 --- a/samples/20070122.testStream.multilineStep.txt +++ b/samples/20070122.testStream.multilineStep.txt @@ -1,2 +1,2 @@ -[Trade: [isin=UK21341EAH45,quantity=978,price=98.34,customer=customer1], Trade: [isin=UK21341EAH46,quantity=112,price=18.12,customer=customer2]] -[Trade: [isin=UK21341EAH47,quantity=245,price=12.78,customer=customer2], Trade: [isin=UK21341EAH48,quantity=108,price=9.25,customer=customer3], Trade: [isin=UK21341EAH49,quantity=854,price=23.39,customer=customer4]] +[Trade: [isin=UK21341EAH45,quantity=978,price=98.34,customer=customer1], Trade: [isin=UK21341EAH46,quantity=112,price=18.12,customer=customer2]] +[Trade: [isin=UK21341EAH47,quantity=245,price=12.78,customer=customer2], Trade: [isin=UK21341EAH48,quantity=108,price=9.25,customer=customer3], Trade: [isin=UK21341EAH49,quantity=854,price=23.39,customer=customer4]] diff --git a/samples/20070122.teststream.multilineOrderStep.TEMP.txt b/samples/20070122.teststream.multilineOrderStep.TEMP.txt index 40214ffda..b70ccda2a 100644 --- a/samples/20070122.teststream.multilineOrderStep.TEMP.txt +++ b/samples/20070122.teststream.multilineOrderStep.TEMP.txt @@ -1,17 +1,17 @@ -BEGIN_ORDER:13100345 2007/02/15 -CUSTOMER:20014539 Peter Smith -ADDRESS:Oak Street 31/A Small Town00235 -BILLING:VISA VISA-12345678903 -ITEM:104439104137.49 -ITEM:2134776319221.99 -END_ORDER:267.34 -BEGIN_ORDER:13100346 2007/02/15 -CUSTOMER:72155919 -ADDRESS:St. Andrews Road 31 London 55342 -BILLING:AMEX AMEX-72345678903 -ITEM:10443191011070.50 -ITEM:213472721921.79 -ITEM:104433930179.95 -ITEM:213474731955.29 -ITEM:1044359501339.99 -END_ORDER:14043.74 +BEGIN_ORDER:13100345 2007/02/15 +CUSTOMER:20014539 Peter Smith +ADDRESS:Oak Street 31/A Small Town00235 +BILLING:VISA VISA-12345678903 +ITEM:104439104137.49 +ITEM:2134776319221.99 +END_ORDER:267.34 +BEGIN_ORDER:13100346 2007/02/15 +CUSTOMER:72155919 +ADDRESS:St. Andrews Road 31 London 55342 +BILLING:AMEX AMEX-72345678903 +ITEM:10443191011070.50 +ITEM:213472721921.79 +ITEM:104433930179.95 +ITEM:213474731955.29 +ITEM:1044359501339.99 +END_ORDER:14043.74 diff --git a/samples/src/main/resources/jobs/xmlStaxJob.xml b/samples/src/main/resources/jobs/xmlStaxJob.xml index 8d6c267ec..eae25c183 100644 --- a/samples/src/main/resources/jobs/xmlStaxJob.xml +++ b/samples/src/main/resources/jobs/xmlStaxJob.xml @@ -22,12 +22,12 @@ - + - + @@ -50,14 +50,14 @@ - + - +