BATCH-166 - additional asserts in StaxEventReaderInputSource.

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 <code>true</code> if next fragment was found, <code>false</code> 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);
 	}

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.