IN PROGRESS - issue BATCH-382: FlatFileItemWriter and StaxEventItemWriter should buffer output

http://jira.springframework.org/browse/BATCH-382

cleaned up stax writer and tests

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

@@ -23,7 +23,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.exception.ClearFailedException;
 import org.springframework.batch.item.exception.FlushFailedException;
-import org.springframework.batch.item.exception.StreamException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.io.Resource;
 import org.springframework.dao.DataAccessResourceFailureException;
@@ -37,7 +36,7 @@ import org.springframework.util.CollectionUtils;
  * This item writer also provides restart, statistics and transaction features
  * by implementing corresponding interfaces.
  * 
- * Output is buffered until {@link #flush()} is called - only then the actual
+ * Output is buffered until {@link #flush()} is wasCalled - only then the actual
  * writing to file takes place.
  * 
  * @author Peter Zozom
@@ -80,9 +79,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	// root element attributes
 	private Map rootElementAttributes = null;
 
-	// signalizes that output source has been initialized
-	private boolean initialized = false;
-
 	// signalizes that marshalling was restarted
 	private boolean restarted = false;
 
@@ -220,10 +216,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 		this.overwriteOutput = overwriteOutput;
 	}
 
-	protected FileChannel getChannel() {
-		return channel;
-	}
-
 	/**
 	 * @throws Exception
 	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
@@ -251,7 +243,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 		open(startAtPosition);
 	}
 
-	/*
+	/**
 	 * Helper method for opening output source at given file position
 	 */
 	private void open(long position) {
@@ -283,8 +275,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 			throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", xse);
 		}
 
-		initialized = true;
-
 	}
 
 	/**
@@ -299,7 +289,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	 * @param writer XML event writer
 	 * @throws XMLStreamException
 	 */
-	protected void startDocument(XMLEventWriter writer) throws XMLStreamException {
+	private void startDocument(XMLEventWriter writer) throws XMLStreamException {
 
 		XMLEventFactory factory = XMLEventFactory.newInstance();
 
@@ -328,14 +318,14 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	 * @param writer XML event writer
 	 * @throws XMLStreamException
 	 */
-	protected void endDocument(XMLEventWriter writer) throws XMLStreamException {
+	private void endDocument(XMLEventWriter writer) throws XMLStreamException {
 
 		// writer.writeEndDocument(); <- this doesn't work after restart
 		// we need to write end tag of the root element manually
 		writer.flush();
 		ByteBuffer bbuf = ByteBuffer.wrap(("</" + getRootTagName() + ">").getBytes());
 		try {
-			getChannel().write(bbuf);
+			channel.write(bbuf);
 		}
 		catch (IOException ioe) {
 			throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", ioe);
@@ -348,7 +338,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	 * @see org.springframework.batch.item.ResourceLifecycle#close(ExecutionContext)
 	 */
 	public void close(ExecutionContext executionContext) {
-		initialized = false;
 		flush();
 		try {
 			endDocument(delegateEventWriter);
@@ -364,15 +353,13 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	}
 
 	/**
-	 * Write the value object to XML stream.
+	 * Write the value object to internal buffer.
 	 * 
 	 * @param item the value object
-	 * @see org.springframework.batch.item.ItemWriter#write(java.lang.Object)
+	 * @see #flush()
 	 */
 	public void write(Object item) {
-
-		Assert.state(initialized, "ItemStream must be open before it can be used.");
-
+		
 		currentRecordCount++;
 		buffer.add(item);
 	}
@@ -382,9 +369,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	 * @see org.springframework.batch.item.ItemStream#update(ExecutionContext)
 	 */
 	public void update(ExecutionContext executionContext) {
-		if (!initialized) {
-			throw new StreamException("ItemStream is not open, or may have been closed.  Cannot access context.");
-		}
 
 		if (saveState) {
 			Assert.notNull(executionContext, "ExecutionContext must not be null");
@@ -393,15 +377,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 		}
 	}
 
-	/**
-	 * Restore processing from provided restart data.
-	 * @param data the restart data
-	 * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionContext)
-	 */
-	public void restoreFrom(ExecutionContext data) {
-
-	}
-
 	/*
 	 * Get the actual position in file channel. This method flushes any buffered
 	 * data before position is read.
@@ -443,7 +418,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements
 	}
 
 	/**
-	 * Writes buffered items to file and marks restore point.
+	 * Writes buffered items to XML stream and marks restore point.
 	 */
 	public void flush() throws FlushFailedException {
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java

@@ -33,8 +33,8 @@ public class StaxEventItemWriterTests extends TestCase {
 	
 	private ExecutionContext executionContext;
 
-	// test record for writing to output
-	private Object record = new Object() {
+	// test item for writing to output
+	private Object item = new Object() {
 		public String toString() {
 			return TEST_STRING;
 		}
@@ -51,37 +51,51 @@ public class StaxEventItemWriterTests extends TestCase {
 	}
 
 	/**
-	 * Write should pass its argument and StaxResult object to Serializer
+	 * Flush should pass buffered items to Serializer.
 	 */
-	public void testWrite() throws Exception {
+	public void testFlush() throws Exception {
 		writer.open(executionContext);
-		Marshaller marshaller = new InputCheckMarshaller();
+		InputCheckMarshaller marshaller = new InputCheckMarshaller();
 		MarshallingEventWriterSerializer serializer = new MarshallingEventWriterSerializer(marshaller);
 		writer.setSerializer(serializer);
 
 		// see asserts in the marshaller
-		writer.write(record);
+		writer.write(item);
+		assertFalse(marshaller.wasCalled);
+		
+		writer.flush();
+		assertTrue(marshaller.wasCalled);
 
 	}
+	
+	public void testClear() throws Exception {
+		writer.open(executionContext);
+		writer.write(item);
+		writer.write(item);
+		writer.clear();
+		//writer.write(item);
+		writer.flush();
+		assertFalse(outputFileContent().contains(TEST_STRING));
+	}
 
 	/**
 	 * Rolled back records should not be written to output file.
 	 */
 	public void testRollback() throws Exception {
 		writer.open(executionContext);
-		writer.write(record);
+		writer.write(item);
 		// rollback
 		writer.clear();
 		assertEquals("", outputFileContent());
 	}
 
 	/**
-	 * Commited output is written to the output file.
+	 * Item is written to the output file only after flush.
 	 */
-	public void testCommit() throws Exception {
+	public void testWriteAndFlush() throws Exception {
 		writer.open(executionContext);
-		writer.write(record);
-		// commit
+		writer.write(item);
+		assertEquals("", outputFileContent());
 		writer.flush();
 		assertTrue(outputFileContent().contains(TEST_STRING));
 	}
@@ -91,8 +105,8 @@ public class StaxEventItemWriterTests extends TestCase {
 	 */
 	public void testRestart() throws Exception {
 		writer.open(executionContext);
-		// write record
-		writer.write(record);
+		// write item
+		writer.write(item);
 		writer.flush();
 		writer.update(executionContext);
 		writer.close(executionContext);
@@ -100,7 +114,7 @@ public class StaxEventItemWriterTests extends TestCase {
 		// create new writer from saved restart data and continue writing
 		writer = createItemWriter();
 		writer.open(executionContext);
-		writer.write(record);
+		writer.write(item);
 		writer.close(executionContext);
 		
 		// check the output is concatenation of 'before restart' and 'after
@@ -123,7 +137,7 @@ public class StaxEventItemWriterTests extends TestCase {
 		writer.open(executionContext);
 		final int NUMBER_OF_RECORDS = 10;
 		for (int i = 1; i <= NUMBER_OF_RECORDS; i++) {
-			writer.write(record);
+			writer.write(item);
 			writer.update(executionContext);
 			long writeStatistics = executionContext.getLong(StaxEventItemWriter.class.getSimpleName() + ".record.count");
 
@@ -154,9 +168,13 @@ public class StaxEventItemWriterTests extends TestCase {
 	 * Checks the received parameters.
 	 */
 	private class InputCheckMarshaller implements Marshaller {
+		
+		boolean wasCalled = false;
+		
 		public void marshal(Object graph, Result result) {
+			wasCalled = true;
 			assertTrue(result instanceof StaxResult);
-			assertSame(record, graph);
+			assertSame(item, graph);
 		}
 
 		public boolean supports(Class clazz) {