diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java
index 2d36c662b..12207646c 100644
--- a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java
+++ b/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java
@@ -30,12 +30,12 @@ import org.springframework.util.Assert;
/**
* Input source for reading XML input based on StAX.
- *
+ *
* It extracts fragments from the input XML document which correspond to records
* for processing. The fragments are wrapped with StartDocument and EndDocument
* events so that the fragments can be further processed like standalone XML
* documents.
- *
+ *
* @author Robert Kasanicky
*/
public class StaxEventReaderInputSource implements InputSource, ResourceLifecycle, Skippable, Restartable, StatisticsProvider, InitializingBean, DisposableBean {
@@ -45,7 +45,7 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
private static final String RESTART_DATA_NAME = "StaxEventReaderInputSource.recordcount";
private FragmentEventReader fragmentReader;
-
+
private TransactionalEventReader txReader;
private FragmentDeserializer fragmentDeserializer;
@@ -66,6 +66,12 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
private List skipRecords = new ArrayList();
+ /**
+ * Read in the next root element from the file, and return it.
+ *
+ * @return the next available record, if none exist, return null
+ * @see org.springframework.batch.io.InputSource#read()
+ */
public Object read() {
if (!initialized) {
open();
@@ -80,7 +86,7 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
fragmentReader.markFragmentProcessed();
}
} while (skipRecords.contains(new Long(currentRecordCount)));
-
+
if (item == null) {
currentRecordCount--;
}
@@ -135,13 +141,19 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
}
/**
- * @param fragmentRootElementName name of the root element of the fragment
+ * @param fragmentRootElementName name of the root element of the fragment
* TODO String can be ambiguous due to namespaces, use QName?
*/
public void setFragmentRootElementName(String fragmentRootElementName) {
this.fragmentRootElementName = fragmentRootElementName;
}
+ /**
+ * Mark the last read record as 'skipped', so that I will not be returned
+ * from read() in the case of a rollback.
+ *
+ * @see Skippable#skip()
+ */
public void skip() {
skipRecords.add(new Long(currentRecordCount));
}
@@ -155,14 +167,25 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
return statistics;
}
+ /**
+ * Ensure that all required dependencies for the InputSource to run are provided
+ * after all properties have been set.
+ *
+ * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+ * @throws IllegalArgumentException if the Resource, FragmentDeserializer or
+ * FragmentRootElementName is null, or if the root element is empty.
+ * @throws IllegalStateException if the Resource does not exist.
+ */
public void afterPropertiesSet() throws Exception {
- Assert.notNull(resource);
+ Assert.notNull(resource, "The Resource must not be null.");
Assert.state(resource.exists(), "Input resource does not exist: [" + resource + "]");
- Assert.notNull(fragmentDeserializer);
+ Assert.notNull(fragmentDeserializer, "The FragmentDeserializer must not be null.");
+ Assert.hasLength(fragmentRootElementName, "The FragmentRootElementName must not be null");
}
/**
* @return wrapped count of records read so far.
+ * @see Restartable#getRestartData()
*/
public RestartData getRestartData() {
Properties restartData = new Properties();
@@ -173,38 +196,43 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
}
/**
- * Rereads (skips) the number of records extracted from restart data.
+ * Restores the input source for the given restart data by rereading and skipping
+ * the number of records stored in the RestartData.
+ *
+ * @param RestartData that holds the line count from the last commit.
+ * @throws IllegalStateException if the InputSource has already been initialized
+ * or if the number of records to read and skip exceeds the available records.
*/
public void restoreFrom(RestartData data) {
- if (data == null || data.getProperties() == null ||
+ Assert.state(!initialized);
+ if (data == null || data.getProperties() == null ||
data.getProperties().getProperty(RESTART_DATA_NAME) == null) {
return;
}
- if (!initialized) {
- open();
- }
+ open();
long restoredRecordCount = Long.parseLong(data.getProperties().getProperty(RESTART_DATA_NAME));
- int REASONABLE_ADHOC_COMMIT_FREQUENCY = 10000;
+ int REASONABLE_ADHOC_COMMIT_FREQUENCY = 100;
while (currentRecordCount <= restoredRecordCount) {
currentRecordCount++;
if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) {
txReader.onCommit(); // reset the history buffer
}
+ Assert.state(fragmentReader.hasNext(), "restore point must be before end of input");
fragmentReader.next();
moveCursorToNextFragment(fragmentReader);
}
txReader.onCommit(); // reset the history buffer
}
-
+
/**
* Responsible for moving the cursor before the StartElement of the fragment root.
- *
+ *
* This implementation simply looks for the next corresponding element, it does not care
- * about element nesting. You will need to override this method to correctly handle
+ * about element nesting. You will need to override this method to correctly handle
* composite fragments.
- *
+ *
* @return true if next fragment was found, false otherwise.
*/
protected boolean moveCursorToNextFragment(XMLEventReader reader) {
@@ -228,12 +256,12 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
throw new DataAccessResourceFailureException("Error while reading from event reader", e);
}
}
-
- // package visibility method for simulating transaction events
+
+ // package visibility method for simulating transaction events in tests
TransactionSynchronization getSynchronization() {
return synchronization;
}
-
+
/**
* Encapsulates transaction events for the StaxEventReaderInputSource.
*/
@@ -255,13 +283,13 @@ public class StaxEventReaderInputSource implements InputSource, ResourceLifecycl
fragmentReader.reset();
}
}
-
+
}
public void destroy() throws Exception {
close();
}
-
+
private void registerSynchronization() {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
}
diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java
index 0f8fd1d9a..f0899fa5f 100644
--- a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java
+++ b/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java
@@ -15,6 +15,7 @@ import javax.xml.stream.events.XMLEvent;
import junit.framework.TestCase;
+import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.core.io.AbstractResource;
import org.springframework.core.io.ByteArrayResource;
@@ -49,12 +50,33 @@ public class StaxEventReaderInputSourceTests extends TestCase {
public void testAfterPropertesSetException() throws Exception{
source.setResource(null);
- try{
+ try {
source.afterPropertiesSet();
fail();
- }catch(IllegalArgumentException ex){
+ }
+ catch (IllegalArgumentException e){
//expected;
}
+
+ source = createNewInputSouce();
+ source.setFragmentRootElementName("");
+ try {
+ source.afterPropertiesSet();
+ fail();
+ }
+ catch (IllegalArgumentException e) {
+ // expected
+ }
+
+ source = createNewInputSouce();
+ source.setFragmentDeserializer(null);
+ try {
+ source.afterPropertiesSet();
+ fail();
+ }
+ catch (IllegalArgumentException e) {
+ // expected
+ }
}
/**
@@ -95,6 +117,8 @@ public class StaxEventReaderInputSourceTests extends TestCase {
public void testRestart() {
source.read();
RestartData restartData = source.getRestartData();
+ assertEquals("1", restartData.getProperties().
+ getProperty("StaxEventReaderInputSource.recordcount"));
List expectedAfterRestart = (List) source.read();
source = createNewInputSouce();
@@ -102,6 +126,34 @@ public class StaxEventReaderInputSourceTests extends TestCase {
List afterRestart = (List) source.read();
assertEquals(expectedAfterRestart.size(), afterRestart.size());
}
+
+ /**
+ * Restore point must not exceed end of file,
+ * input source must not be already initialized when restoring.
+ */
+ public void testInvalidRestore() {
+ Properties props = new Properties() {{
+ final String MORE_RECORDS_THAN_INPUT_CONTAINS = "100000";
+ setProperty("StaxEventReaderInputSource.recordcount", MORE_RECORDS_THAN_INPUT_CONTAINS);
+ }};
+ try {
+ source.restoreFrom(new GenericRestartData(props));
+ fail();
+ }
+ catch (IllegalStateException e) {
+ // expected
+ }
+
+ source = createNewInputSouce();
+ source.open();
+ try {
+ source.restoreFrom(new GenericRestartData(new Properties()));
+ fail();
+ }
+ catch (IllegalStateException e) {
+ // expected
+ }
+ }
/**
* Skipping marked records after rollback.