OPEN - issue BATCH-116: Create XML input/output source which will work directly with StAX parser

http://opensource.atlassian.com/projects/spring/browse/BATCH-116

infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java

@@ -0,0 +1,44 @@
+package org.springframework.batch.io.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.dao.DataAccessResourceFailureException;
+import org.springframework.oxm.Marshaller;
+import org.springframework.util.Assert;
+import org.springframework.xml.transform.StaxResult;
+
+/**
+ * Object to xml serializer that wraps a Spring-OXM
+ * Marshaller object.
+ *
+ * @author Lucas Ward
+ *
+ */
+public class MarshallingObjectToXmlSerializer implements ObjectToXmlSerializer{
+
+	private Marshaller marshaller;
+
+	private Result result;
+
+	public MarshallingObjectToXmlSerializer(Marshaller marshaller){
+		this.marshaller = marshaller;
+	}
+
+	public void setEventWriter(XMLEventWriter writer) {
+		result = new StaxResult(writer);
+	}
+
+	public void serializeObject(Object output) {
+
+		try {
+			marshaller.marshal(output, result);
+		} catch (IOException xse) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + result.getSystemId() + "]", xse);
+		}
+	}
+}

infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java

@@ -0,0 +1,41 @@
+package org.springframework.batch.io.oxm;
+
+import java.io.IOException;
+
+import javax.xml.stream.XMLEventReader;
+
+import org.springframework.batch.io.stax.FragmentDeserializer;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.oxm.Unmarshaller;
+import org.springframework.oxm.UnmarshallingFailureException;
+import org.springframework.oxm.XmlMappingException;
+import org.springframework.xml.transform.StaxSource;
+
+/**
+ * Delegates deserializing to Spring-WS {@link Unmarshaller}.
+ *
+ * @author Robert Kasanicky
+ * @authoer Lucas Ward
+ */
+public class UnmarshallingFragmentDeserializer implements FragmentDeserializer {
+
+	private Unmarshaller unmarshaller;
+
+	public UnmarshallingFragmentDeserializer(Unmarshaller unmarshaller){
+		this.unmarshaller = unmarshaller;
+	}
+
+	public Object deserializeFragment(XMLEventReader eventReader) {
+		Object item = null;
+		try {
+			item = unmarshaller.unmarshal(new StaxSource(eventReader));
+		}
+		catch (XmlMappingException e) {
+			throw new UnmarshallingFailureException("Mapping failure during unmarshalling", e);
+		}
+		catch (IOException e) {
+			throw new DataAccessResourceFailureException("IO error during unmarshalling", e);
+		}
+		return item;
+	}
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java

@@ -0,0 +1,58 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.XMLEvent;
+
+/**
+ * Delegates all functionality to the wrapped reader allowing
+ * subclasses to override only the methods they want to change.
+ * 
+ * @author Robert Kasanicky
+ */
+abstract class AbstractEventReaderWrapper implements XMLEventReader {
+
+	protected XMLEventReader wrappedEventReader;
+	
+	public AbstractEventReaderWrapper(XMLEventReader wrappedEventReader) {
+		this.wrappedEventReader = wrappedEventReader;
+	}
+	
+	public void close() throws XMLStreamException {
+		wrappedEventReader.close();
+		
+	}
+
+	public String getElementText() throws XMLStreamException {
+		return wrappedEventReader.getElementText();
+	}
+
+	public Object getProperty(String name) throws IllegalArgumentException {
+		return wrappedEventReader.getProperty(name);
+	}
+
+	public boolean hasNext() {
+		return wrappedEventReader.hasNext();
+	}
+
+	public XMLEvent nextEvent() throws XMLStreamException {
+		return wrappedEventReader.nextEvent();
+	}
+
+	public XMLEvent nextTag() throws XMLStreamException {
+		return wrappedEventReader.nextTag();
+	}
+
+	public XMLEvent peek() throws XMLStreamException {
+		return wrappedEventReader.peek();
+	}
+
+	public Object next() {
+		return wrappedEventReader.next();
+	}
+
+	public void remove() {
+		wrappedEventReader.remove();
+	}
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java

@@ -0,0 +1,180 @@
+package org.springframework.batch.io.stax;
+
+import java.util.NoSuchElementException;
+
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLEventFactory;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.EndDocument;
+import javax.xml.stream.events.EndElement;
+import javax.xml.stream.events.StartDocument;
+import javax.xml.stream.events.StartElement;
+import javax.xml.stream.events.XMLEvent;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+
+/**
+ * Default implementation of {@link FragmentEventReader}
+ * 
+ * @author Robert Kasanicky
+ */
+class DefaultFragmentEventReader extends AbstractEventReaderWrapper implements FragmentEventReader {
+
+	// true when the next event is the StartElement of next fragment
+	private boolean startFragmentFollows = false;
+
+	// true when the next event is the EndElement of current fragment
+	private boolean endFragmentFollows = false;
+
+	// true while cursor is inside fragment
+	private boolean insideFragment = false;
+
+	// true when reader should behave like the cursor was at the end of document
+	private boolean fakeDocumentEnd = false;
+
+	private StartDocument startDocumentEvent = null;
+
+	private EndDocument endDocumentEvent = null;
+
+	// fragment root name is remembered so that the matching closing element can
+	// be identified
+	private QName fragmentRootName = null;
+
+	// counts the occurrences of current fragmentRootName (increased for
+	// StartElement, decreased for EndElement)
+	private int matchCounter = 0;
+
+	/**
+	 * Caches the StartDocument event for later use.
+	 * @param wrappedEventReader the original wrapped event reader
+	 */
+	public DefaultFragmentEventReader(XMLEventReader wrappedEventReader) {
+		super(wrappedEventReader);
+		try {
+			startDocumentEvent = (StartDocument) wrappedEventReader.peek();
+		}
+		catch (XMLStreamException e) {
+			throw new DataAccessResourceFailureException("Error reading start document from event reader", e);
+		}
+
+		endDocumentEvent = XMLEventFactory.newInstance().createEndDocument();
+	}
+
+	public void markStartFragment() {
+		startFragmentFollows = true;
+		fragmentRootName = null;
+	}
+
+	public boolean hasNext() {
+		try {
+			if (peek() != null) {
+				return true;
+			}
+		}
+		catch (XMLStreamException e) {
+			throw new DataAccessResourceFailureException("Error reading XML stream", e);
+		}
+		return false;
+	}
+
+	public Object next() {
+		try {
+			return nextEvent();
+		}
+		catch (XMLStreamException e) {
+			throw new DataAccessResourceFailureException("Error reading XML stream", e);
+		}
+	}
+
+	public XMLEvent nextEvent() throws XMLStreamException {
+		if (fakeDocumentEnd) {
+			throw new NoSuchElementException();
+		}
+		XMLEvent event = wrappedEventReader.peek();
+		XMLEvent proxyEvent = alterEvent(event, false);
+		checkFragmentEnd(proxyEvent);
+		if (event == proxyEvent) {
+			wrappedEventReader.nextEvent();
+		}
+
+		return proxyEvent;
+	}
+
+	/**
+	 * Sets the endFragmentFollows flag to true if next event is the last event
+	 * of the fragment.
+	 * @param event peek() from wrapped event reader
+	 */
+	private void checkFragmentEnd(XMLEvent event) {
+		if (event.isStartElement() && ((StartElement) event).getName().equals(fragmentRootName)) {
+			matchCounter++;
+		}
+		else if (event.isEndElement() && ((EndElement) event).getName().equals(fragmentRootName)) {
+			matchCounter--;
+			if (matchCounter == 0) {
+				endFragmentFollows = true;
+			}
+		}
+	}
+
+	/**
+	 * @param event peek() from wrapped event reader
+	 * @param peek if true do not change the internal state
+	 * @return StartDocument event if peek() points to beginning of fragment
+	 * EndDocument event if cursor is right behind the end of fragment original
+	 * event otherwise
+	 */
+	private XMLEvent alterEvent(XMLEvent event, boolean peek) {
+		if (startFragmentFollows) {
+			fragmentRootName = ((StartElement) event).getName();
+			if (!peek) {
+				startFragmentFollows = false;
+				insideFragment = true;
+			}
+			return startDocumentEvent;
+		}
+		else if (endFragmentFollows) {
+			if (!peek) {
+				endFragmentFollows = false;
+				insideFragment = false;
+				fakeDocumentEnd = true;
+			}
+			return endDocumentEvent;
+		}
+		return event;
+	}
+
+	public XMLEvent peek() throws XMLStreamException {
+		if (fakeDocumentEnd) {
+			return null;
+		}
+		return alterEvent(wrappedEventReader.peek(), true);
+	}
+
+	/**
+	 * Finishes reading the fragment in case the fragment was processed without
+	 * being read until the end.
+	 */
+	public void markFragmentProcessed() {
+		if (insideFragment) {
+			try {
+				while (!(nextEvent() instanceof EndDocument)) {
+					// just read all events until EndDocument
+				}
+			}
+			catch (XMLStreamException e) {
+				throw new DataAccessResourceFailureException("Error reading XML stream", e);
+			}
+		}
+		fakeDocumentEnd = false;
+	}
+
+	public void reset() {
+		insideFragment = false;
+		startFragmentFollows = false;
+		endFragmentFollows = false;
+		fakeDocumentEnd = false;
+	}
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java

@@ -0,0 +1,220 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.Characters;
+import javax.xml.stream.events.XMLEvent;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+/**
+ * Class used to wrap XMLEventReader. Events from wrapped reader are stored in
+ * {@link EventSequence} to support transactions.
+ * 
+ * @author tomas.slanina
+ */
+class DefaultTransactionalEventReader implements TransactionalEventReader, InitializingBean {
+
+	private EventSequence recorder = new EventSequence();
+
+	private XMLEventReader parent;
+
+
+	/**
+	 * Creates instance of this class and wraps XMLEventReader.
+	 * 
+	 * @param parent event reader to be wrapped.
+	 */
+	public DefaultTransactionalEventReader(XMLEventReader parent) {
+		setParent(parent);
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(parent);
+	}
+
+	/**
+	 * Callback on transaction rollback.
+	 */
+	public void onRollback() {
+		recorder.reset();
+	}
+
+	/**
+	 * Callback on transacion commit.
+	 * 
+	 */
+	public void onCommit() {
+		recorder.clear();
+	}
+
+	/**
+	 * @return the parent
+	 */
+	public XMLEventReader getParent() {
+		return parent;
+	}
+
+	/**
+	 * @param parent the parent to set
+	 */
+	public void setParent(XMLEventReader parent) {
+		this.parent = parent;
+	}
+
+	/**
+	 * @param recorder the recorder to set
+	 */
+	public void setRecorder(EventSequence recorder) {
+		this.recorder = recorder;
+	}
+
+	/**
+	 * Returns the xml event recorder
+	 * @return the xml event recorder
+	 */
+	public EventSequence getRecorder() {
+		return recorder;
+	}
+
+	/**
+	 * Frees any resources associated with this Reader. This method does not
+	 * close the underlying input source.
+	 * 
+	 * @throws XMLStreamException if there are errors freeing associated
+	 * resources
+	 */
+	public void close() throws XMLStreamException {
+		parent.close();
+
+	}
+
+	/**
+	 * Reads the content of a text-only element. Precondition: the current event
+	 * is START_ELEMENT. Postcondition: The current event is the corresponding
+	 * END_ELEMENT.
+	 * 
+	 * @throws XMLStreamException if the current event is not a START_ELEMENT or
+	 * if a non text element is encountered
+	 */
+	public String getElementText() throws XMLStreamException {
+		StringBuffer buf = new StringBuffer();
+		XMLEvent e = nextEvent();
+		if (!e.isStartElement()) {
+			throw new XMLStreamException(
+					"Precondition for readText is nextEvent().getEventType() == START_ELEMENT (got " + e.getEventType()
+							+ ")");
+		}
+
+		while (hasNext()) {
+			e = peek();
+			if (e.isStartElement()) {
+				throw new XMLStreamException("Unexpected Element start");
+			}
+			if (e.isCharacters()) {
+				buf.append(((Characters) e).getData());
+			}
+			if (e.isEndElement()) {
+				return buf.toString();
+			}
+			nextEvent();
+		}
+
+		throw new XMLStreamException("Unexpected end of Document");
+	}
+
+	/**
+	 * Get the value of a feature/property from the underlying implementation
+	 * 
+	 * @param name The name of the property
+	 * @return The value of the property
+	 * @throws IllegalArgumentException if the property is not supported
+	 */
+	public Object getProperty(String name) throws IllegalArgumentException {
+		return parent.getProperty(name);
+	}
+
+	/**
+	 * Check if there are more events. Returns true if there are more events and
+	 * false otherwise.
+	 * 
+	 * @return true if the event reader has more events, false otherwise
+	 */
+	public boolean hasNext() {
+		return recorder.hasNext() || parent.hasNext();
+	}
+
+	/**
+	 * Get the next XMLEvent
+	 * 
+	 * @see XMLEvent
+	 * @throws XMLStreamException if there is an error with the underlying XML.
+	 * @throws NoSuchElementException iteration has no more elements.
+	 */
+	public XMLEvent nextEvent() throws XMLStreamException {
+		if (!recorder.hasNext()) {
+			recorder.addEvent(parent.nextEvent());
+		}
+		return recorder.nextEvent();
+	}
+
+	/**
+	 * Skips any insignificant space events until a START_ELEMENT or END_ELEMENT
+	 * is reached. If anything other than space characters are encountered, an
+	 * exception is thrown. This method should be used when processing
+	 * element-only content because the parser is not able to recognize
+	 * ignorable whitespace if the DTD is missing or not interpreted.
+	 * 
+	 * @throws XMLStreamException if anything other than space characters are
+	 * encountered
+	 */
+	public XMLEvent nextTag() throws XMLStreamException {
+		while (hasNext()) {
+			XMLEvent e = nextEvent();
+			if (e.isCharacters() && !((Characters) e).isWhiteSpace()) {
+				throw new XMLStreamException("Unexpected text");
+			}
+			if (e.isStartElement() || e.isEndElement()) {
+				return e;
+			}
+		}
+		throw new XMLStreamException("Unexpected end of Document");
+	}
+
+	/**
+	 * Check the next XMLEvent without reading it from the stream. Returns null
+	 * if the stream is at EOF or has no more XMLEvents. A call to peek() will
+	 * be equal to the next return of next().
+	 * 
+	 * @see XMLEvent
+	 * @throws XMLStreamException
+	 */
+	public XMLEvent peek() throws XMLStreamException {
+		return (recorder.hasNext()) ? recorder.peek() : parent.peek();
+	}
+
+	/**
+	 * Returns the next element in the iteration. Calling this method repeatedly
+	 * until the {@link #hasNext()} method returns false will return each
+	 * element in the underlying collection exactly once.
+	 * 
+	 * @return the next element in the iteration.
+	 * @exception NoSuchElementException iteration has no more elements.
+	 */
+	public Object next() {
+		try {
+			return nextEvent();
+		}
+		catch (XMLStreamException e) {
+			return null;
+		}
+	}
+
+	/**
+	 * In this implementation throws UnsupportedOperationException.
+	 */
+	public void remove() {
+		throw new java.lang.UnsupportedOperationException();
+	}
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java

@@ -0,0 +1,92 @@
+package org.springframework.batch.io.stax;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.xml.stream.events.XMLEvent;
+
+/**
+ * Holds a list of XML events, typically corresponding to a single record.
+ *  
+ * @author tomas.slanina
+ */
+class EventSequence {
+	
+	private static final int BEFORE_BEGINNING = -1;
+	
+	private List events;
+	
+	private int currentIndex;
+	
+	/**
+	 * Creates instance of this class.
+	 *
+	 */
+	public EventSequence() {
+		init();
+	}
+	
+	/**
+	 * Adds event to the list of stored events.
+	 * 
+	 * @param event
+	 */
+	public void addEvent(XMLEvent event) {
+		events.add(event);
+	}
+	
+	/**
+	 * Gets next XMLEvent from cache and moves cursor to next event.
+	 * If cache contains no more events, null is returned.
+	 * 
+	 * @return
+	 */
+	public XMLEvent nextEvent() {
+		return (hasNext()) ? (XMLEvent)events.get(++currentIndex) :null;
+ 	}
+	
+	/**
+	 * Gets next XMLEvent from cache but cursor remains on the same position.
+	 * If cache contains no more events, null is returned.
+	 * 
+	 * @return
+	 */
+	public XMLEvent peek() {
+		return (hasNext()) ? (XMLEvent)events.get(currentIndex+1) :null;
+	}
+	
+	/**
+	 * Removes events from the internal cache.
+	 *
+	 */
+	public void clear() {
+		init();
+	}
+	
+	/**
+	 * Resets cursor to the cache start.
+	 *
+	 */
+	public void reset() {
+		currentIndex = BEFORE_BEGINNING;
+	}
+	
+	/**
+	 * Check if there are more events. Returns true if there are more events and
+	 * false otherwise.
+	 * 
+	 * @return true if the event reader has more events, false otherwise
+	 */
+	public boolean hasNext() {
+		return currentIndex + 1 < events.size();
+	}
+	
+	private void init() {
+		events = (events != null) ? new ArrayList(events.size())
+			: new ArrayList(1000);
+		
+		reset();
+	}
+	
+	
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java

@@ -0,0 +1,14 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+
+/**
+ * Deserializes XML fragment to domain object. 
+ * XML fragment is a standalone XML document corresponding to a single record.
+ * 
+ * @author Robert Kasanicky
+ */
+public interface FragmentDeserializer {
+
+	Object deserializeFragment(XMLEventReader eventReader);
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java

@@ -0,0 +1,33 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+
+
+/**
+ * Interface for event readers which support treating XML fragments as standalone XML documents
+ * by wrapping the fragments with StartDocument and EndDocument events.
+ * 
+ * @author Robert Kasanicky
+ */
+interface FragmentEventReader extends XMLEventReader {
+
+	/**
+	 * Tells the event reader its cursor position is exactly before the fragment.
+	 */
+	void markStartFragment();
+	
+	/**
+	 * Tells the event reader the current fragment has been processed.
+	 * If the cursor is still inside the fragment it should be moved
+	 * after the end of the fragment.
+	 */
+	void markFragmentProcessed();
+	
+	/**
+	 * Reset the state of the fragment reader - make it forget
+	 * it assumptions about current position of cursor
+	 * (e.g. in case of rollback of the wrapped reader).
+	 */
+	void reset();
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java

@@ -0,0 +1,62 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.namespace.NamespaceContext;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLEventWriter;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.XMLEvent;
+
+/**
+ * Delegating XMLEventWriter, which ignores start and end document events,
+ * but passes through everything else.
+ * 
+ * @author peter.zozom
+ */
+class NoStartEndDocumentStreamWriter implements XMLEventWriter {
+
+	private XMLEventWriter delegate;
+	
+	public NoStartEndDocumentStreamWriter(XMLEventWriter delegate) {
+		this.delegate = delegate;
+	}
+		
+	public void add(XMLEvent event) throws XMLStreamException {
+		if ((!event.isStartDocument()) && (!event.isEndDocument())) {
+			delegate.add(event);
+		}
+	}
+
+	public void add(XMLEventReader reader) throws XMLStreamException {
+		delegate.add(reader);
+	}
+
+	public void close() throws XMLStreamException {
+		delegate.close();
+	}
+
+	public void flush() throws XMLStreamException {
+		delegate.flush();
+	}
+
+	public NamespaceContext getNamespaceContext() {
+		return delegate.getNamespaceContext();
+	}
+
+	public String getPrefix(String uri) throws XMLStreamException {
+		return delegate.getPrefix(uri);
+	}
+
+	public void setDefaultNamespace(String uri) throws XMLStreamException {
+		delegate.setDefaultNamespace(uri);
+	}
+
+	public void setNamespaceContext(NamespaceContext context)
+			throws XMLStreamException {
+		delegate.setNamespaceContext(context);
+	}
+
+	public void setPrefix(String prefix, String uri) throws XMLStreamException {
+		delegate.setPrefix(prefix, uri);
+	}
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java

@@ -0,0 +1,28 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventWriter;
+
+/**
+ * Interface wrapping the serialization of an object
+ * to xml.  Primarily useful for abstracting how an object
+ * is serialized to an XMLEvent from a specific marshaller.
+ *
+ * @author Lucas Ward
+ *
+ */
+public interface ObjectToXmlSerializer {
+
+	/**
+	 * Set event writer objects should be serialized to.
+	 *
+	 * @param writer
+	 */
+	void setEventWriter(XMLEventWriter writer);
+
+	/**
+	 * Serialize an Object.
+	 *
+	 * @param output
+	 */
+	void serializeObject(Object output);
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventReaderInputSource.java

@@ -0,0 +1,269 @@
+package org.springframework.batch.io.stax;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.StartElement;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.Skippable;
+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;
+import org.springframework.batch.statistics.StatisticsProvider;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.io.Resource;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.util.Assert;
+
+/**
+ * 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, Skippable, Restartable, StatisticsProvider, InitializingBean, DisposableBean {
+
+	public static final String READ_COUNT_STATISTICS_NAME = "StaxEventReaderInputSource.readCount";
+
+	private static final String RESTART_DATA_NAME = "StaxEventReaderInputSource.recordcount";
+
+	private FragmentEventReader fragmentReader;
+	
+	private TransactionalEventReader txReader;
+
+	private FragmentDeserializer fragmentDeserializer;
+
+	private Resource resource;
+
+	private InputStream inputStream;
+
+	private String fragmentRootElementName;
+
+	private boolean initialized = false;
+
+	private TransactionSynchronization synchronization = new StaxEventReaderInputSourceTransactionSychronization();
+
+	private long lastCommitPointRecordCount = 0;
+
+	private long currentRecordCount = 0;
+
+	private List skipRecords = new ArrayList();
+
+	public Object read() {
+		if (!initialized) {
+			open();
+		}
+		Object item = null;
+
+		do {
+			currentRecordCount++;
+			if (moveCursorToNextFragment(fragmentReader)) {
+				fragmentReader.markStartFragment();
+				item = fragmentDeserializer.deserializeFragment(fragmentReader);
+				fragmentReader.markFragmentProcessed();
+			}
+		} while (skipRecords.contains(new Long(currentRecordCount)));
+		
+		if (item == null) {
+			currentRecordCount--;
+		}
+		return item;
+	}
+
+	// TODO make sure exception stack is not lost in any case.
+	public void close() {
+		try {
+			initialized = false;
+			fragmentReader.close();
+		}
+		catch (XMLStreamException e) {
+			throw new DataAccessResourceFailureException("Error while closing event reader", e);
+		}
+		finally {
+			try {
+				inputStream.close();
+			}
+			catch (IOException e) {
+				throw new DataAccessResourceFailureException("Error while closing input stream", e);
+			}
+		}
+	}
+
+	public void open() {
+		registerSynchronization();
+		try {
+			inputStream = resource.getInputStream();
+			txReader = new DefaultTransactionalEventReader(XMLInputFactory
+					.newInstance().createXMLEventReader(inputStream));
+			fragmentReader = new DefaultFragmentEventReader(txReader);
+		}
+		catch (XMLStreamException xse) {
+			throw new DataAccessResourceFailureException("Unable to create XML reader", xse);
+		}
+		catch (IOException ioe) {
+			throw new DataAccessResourceFailureException("Unable to get input stream", ioe);
+		}
+		initialized = true;
+	}
+
+	public void setResource(Resource resource) {
+		this.resource = resource;
+	}
+
+	/**
+	 * @param fragmentDeserializer maps xml fragments corresponding to records to
+	 * objects
+	 */
+	public void setFragmentDeserializer(FragmentDeserializer fragmentDeserializer) {
+		this.fragmentDeserializer = fragmentDeserializer;
+	}
+
+	/**
+	 * @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;
+	}
+
+	public void skip() {
+		skipRecords.add(new Long(currentRecordCount));
+	}
+
+	/**
+	 * @return Properties wrapper for the count of records read so far.
+	 */
+	public Properties getStatistics() {
+		Properties statistics = new Properties();
+		statistics.setProperty(READ_COUNT_STATISTICS_NAME, String.valueOf(currentRecordCount));
+		return statistics;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(resource);
+		Assert.state(resource.exists(), "Input resource does not exist: [" + resource + "]");
+		Assert.notNull(fragmentDeserializer);
+	}
+
+	/**
+	 * @return wrapped count of records read so far.
+	 */
+	public RestartData getRestartData() {
+		Properties restartData = new Properties();
+
+		restartData.setProperty(RESTART_DATA_NAME, String.valueOf(currentRecordCount));
+
+		return new GenericRestartData(restartData);
+	}
+
+	/**
+	 * Rereads (skips) the number of records extracted from restart data.
+	 */
+	public void restoreFrom(RestartData data) {
+		if (data == null || data.getProperties() == null || 
+				data.getProperties().getProperty(RESTART_DATA_NAME) == null) {
+			return;
+		}
+
+		if (!initialized) {
+			open();
+		}
+
+		long restoredRecordCount = Long.parseLong(data.getProperties().getProperty(RESTART_DATA_NAME));
+		int REASONABLE_ADHOC_COMMIT_FREQUENCY = 10000;
+		while (currentRecordCount <= restoredRecordCount) {
+			currentRecordCount++;
+			if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) {
+				txReader.onCommit(); // reset the history buffer
+			}
+			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 
+	 * composite fragments.
+	 * 
+	 * @return <code>true</code> if next fragment was found, <code>false</code> otherwise.
+	 */
+	protected boolean moveCursorToNextFragment(XMLEventReader reader) {
+		try {
+			while (true) {
+				while (reader.peek() != null && !reader.peek().isStartElement()) {
+					reader.nextEvent();
+				}
+				if (reader.peek() == null) {
+					return false;
+				}
+				QName startElementName = ((StartElement) reader.peek()).getName();
+				if (startElementName.getLocalPart().equals(fragmentRootElementName)) {
+					return true;
+				} else {
+					reader.nextEvent();
+				}
+			}
+		}
+		catch (XMLStreamException e) {
+			throw new DataAccessResourceFailureException("Error while reading from event reader", e);
+		}
+	}
+	
+	// package visibility method for simulating transaction events
+	TransactionSynchronization getSynchronization() {
+		return synchronization;
+	}
+	
+	/**
+	 * Encapsulates transaction events for the StaxEventReaderInputSource.
+	 */
+	private class StaxEventReaderInputSourceTransactionSychronization extends TransactionSynchronizationAdapter {
+
+		/**
+		 * @param status
+		 * @see org.springframework.transaction.support.TransactionSynchronizationAdapter#afterCompletion(int)
+		 */
+		public void afterCompletion(int status) {
+			if (status == TransactionSynchronization.STATUS_COMMITTED) {
+				lastCommitPointRecordCount = currentRecordCount;
+				txReader.onCommit();
+				skipRecords = new ArrayList();
+			}
+			else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
+				currentRecordCount = lastCommitPointRecordCount;
+				txReader.onRollback();
+				fragmentReader.reset();
+			}
+		}
+		
+	}
+
+	public void destroy() throws Exception {
+		close();
+	}
+	
+	private void registerSynchronization() {
+		BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
+	}
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java

@@ -0,0 +1,512 @@
+package org.springframework.batch.io.stax;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.xml.stream.XMLEventFactory;
+import javax.xml.stream.XMLEventWriter;
+import javax.xml.stream.XMLOutputFactory;
+import javax.xml.stream.XMLStreamException;
+
+import org.springframework.batch.io.OutputSource;
+import org.springframework.batch.io.support.FileUtils;
+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;
+import org.springframework.batch.statistics.StatisticsProvider;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.io.Resource;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * StaxEventWriterOutputSource is implementation of {@link OutputSource} which uses
+ * StAX and {@link ObjectToXmlSerializer} for serializing object to XML.
+ *
+ * This output source also provides restart, statistics and transaction
+ * features by implementing corresponding interfaces.
+ *
+ * @author Peter Zozom
+ *
+ */
+public class StaxEventWriterOutputSource implements OutputSource, Restartable,
+		StatisticsProvider, InitializingBean, DisposableBean {
+
+	// default encoding
+	private static final String DEFAULT_ENCODING = "UTF-8";
+
+	// default encoding
+	private static final String DEFAULT_XML_VERSION = "1.0";
+
+	// default root tag name
+	private static final String DEFAULT_ROOT_TAG_NAME = "root";
+
+	// restart data property name
+	private static final String RESTART_DATA_NAME = "staxstreamoutputsource.position";
+
+	// read statistics property name
+	public static final String WRITE_STATISTICS_NAME = "staxstreamoutputsource.processedrecordcount";
+
+	// file system resource
+	private Resource resource;
+
+	// xml serializer
+	private ObjectToXmlSerializer serializer;
+
+	// encoding to be used while reading from the resource
+	private String encoding = DEFAULT_ENCODING;
+
+	// XML version
+	private String version = DEFAULT_XML_VERSION;
+
+	// name of the root tag
+	private String rootTagName = DEFAULT_ROOT_TAG_NAME;
+
+	// 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;
+
+	// TRUE means, that output file will be overwritten if exists - default is TRUE
+	private boolean overwriteOutput = true;
+
+	// file channel
+	private FileChannel channel;
+
+	// wrapper for XML event writer that swallows StartDocument and EndDocument events
+	private XMLEventWriter eventWriter;
+
+	// XML event writer
+	private XMLEventWriter delegateEventWriter;
+
+	// transaction synchronization object
+	private TransactionSynchronization synchronization = new StaxEventWriterOutputSourceTransactionSychronization();
+
+	// byte offset in file channel at last commit point
+	private long lastCommitPointPosition = 0;
+
+	// processed record count at last commit point
+	private long lastCommitPointRecordCount = 0;
+
+	// current count of processed records
+	private long currentRecordCount = 0;
+
+
+	/**
+	 * Set output file.
+	 *
+	 * @param resource the output file
+	 */
+	public void setResource(Resource resource) {
+		this.resource = resource;
+	}
+
+	/**
+	 * Set Object to XML serializer.
+	 *
+	 * @param serializer the Object to XML serializer
+	 */
+	public void setSerializer(ObjectToXmlSerializer serializer) {
+		this.serializer = serializer;
+	}
+
+	/**
+	 * Get used encoding.
+	 *
+	 * @return the encoding used
+	 */
+	public String getEncoding() {
+		return encoding;
+	}
+
+	/**
+	 * Set encoding to be used for output file.
+	 *
+	 * @param encoding the encoding to be used
+	 */
+	public void setEncoding(String encoding) {
+		this.encoding = encoding;
+	}
+
+	/**
+	 * Get XML version.
+	 *
+	 * @return the XML version used
+	 */
+	public String getVersion() {
+		return version;
+	}
+
+	/**
+	 * Set XML version to be used for output XML.
+	 *
+	 * @param version the XML version to be used
+	 */
+	public void setVersion(String version) {
+		this.version = version;
+	}
+
+	/**
+	 * Get the tag name of the root element.
+	 *
+	 * @return the root element tag name
+	 */
+	public String getRootTagName() {
+		return rootTagName;
+	}
+
+	/**
+	 * Set the tag name of the root element. If not set, default name is used ("root").
+	 *
+	 * @param rootTagName the tag name to be used for the root element
+	 */
+	public void setRootTagName(String rootTagName) {
+		this.rootTagName = rootTagName;
+	}
+
+	/**
+	 * Get attributes of the root element.
+	 *
+	 * @return attributes of the root element
+	 */
+	public Map getRootElementAttributes() {
+		return rootElementAttributes;
+	}
+
+	/**
+	 * Set the root element attributes to be written.
+	 *
+	 * @param rootElementAttributes attributes of the root element
+	 */
+	public void setRootElementAttributes(Map rootElementAttributes) {
+		this.rootElementAttributes = rootElementAttributes;
+	}
+
+	/**
+	 * Set "overwrite" flag for the output file. Flag is ignored when output file processing is restarted.
+	 *
+	 * @param shouldDeleteIfExists
+	 */
+	public void setOverwriteOutput(boolean overwriteOutput) {
+		this.overwriteOutput = overwriteOutput;
+	}
+
+	protected FileChannel getChannel() {
+		return channel;
+	}
+
+	/**
+	 * @throws Exception
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(resource);
+		Assert.notNull(serializer);
+	}
+
+	/**
+	 * @throws Exception
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
+	public void destroy() throws Exception {
+		close();
+	}
+
+	/**
+	 * Register the input source for transaction synchronization.
+	 */
+	private void registerSynchronization() {
+		BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
+	}
+
+	/**
+	 * Open the output source
+	 *
+	 * @see org.springframework.batch.item.ResourceLifecycle#open()
+	 */
+	public void open() {
+		open(0);
+	}
+
+	/*
+	 * Helper method for opening output source at given file position
+	 */
+	private void open(long position) {
+
+		registerSynchronization();
+
+		File file;
+		try {
+			file = resource.getFile();
+			FileUtils.setUpOutputFile(file, restarted, overwriteOutput);
+		} catch (IOException ioe) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + resource + "]", ioe);
+		}
+
+		FileOutputStream os = null;
+
+		try {
+			os = new FileOutputStream(file, true);
+			channel = os.getChannel();
+			setPosition(position);
+		} catch (IOException ioe) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + resource + "]", ioe);
+		}
+
+		XMLOutputFactory outputFactory = XMLOutputFactory.newInstance();
+
+		try {
+			delegateEventWriter = outputFactory.createXMLEventWriter(os, encoding);
+			eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter);
+			serializer.setEventWriter(eventWriter);
+			if (!restarted) {
+				startDocument(delegateEventWriter);
+			}
+		} catch (XMLStreamException xse) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + resource + "]", xse);
+		}
+
+		initialized = true;
+	}
+
+	/**
+	 * Writes simple XML header containing:
+	 * <ul>
+	 * <li>xml declaration - defines encoding and XML version</li>
+	 * <li>opening tag of the root element and its attributes</li>
+	 * </ul>
+	 * If this is not sufficient for you, simply override this method. Encoding,
+	 * version and root tag name can be retrieved with corresponding getters.
+	 *
+	 * @param writer
+	 *            XML event writer
+	 * @throws XMLStreamException
+	 */
+	protected void startDocument(XMLEventWriter writer) throws XMLStreamException {
+
+		XMLEventFactory factory = XMLEventFactory.newInstance();
+
+		//write start document
+		writer.add(factory.createStartDocument(getEncoding(), getVersion()));
+
+		//write root tag
+		writer.add(factory.createStartElement("", "", getRootTagName()));
+
+		//write root tag attributes
+		if (!CollectionUtils.isEmpty(getRootElementAttributes())) {
+
+			for (Iterator i = getRootElementAttributes().entrySet().iterator(); i.hasNext();) {
+				Map.Entry entry = (Map.Entry)i.next();
+				writer.add(factory.createAttribute((String)entry.getKey(), (String)entry.getValue()));
+			}
+
+		}
+
+	}
+
+	/**
+	 * Finishes the XML document. It closes any start tag and writes
+	 * corresponding end tags.
+	 *
+	 * @param writer
+	 *            XML event writer
+	 * @throws XMLStreamException
+	 */
+	protected 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);
+		} catch (IOException ioe) {
+			throw new DataAccessResourceFailureException(
+					"Unable to close file resource: [" + resource + "]", ioe);
+		}
+	}
+
+	/**
+	 * Close the output source.
+	 *
+	 * @see org.springframework.batch.item.ResourceLifecycle#close()
+	 */
+	public void close() {
+		initialized = false;
+		try {
+			endDocument(delegateEventWriter);
+			eventWriter.close();
+			channel.close();
+		} catch (XMLStreamException xse) {
+			throw new DataAccessResourceFailureException(
+					"Unable to close file resource: [" + resource + "]", xse);
+		}
+		catch (IOException ioe) {
+			throw new DataAccessResourceFailureException(
+					"Unable to close file resource: [" + resource + "]", ioe);
+		}
+	}
+
+	/**
+	 * Write the value object to XML stream.
+	 *
+	 * @param output the value object
+	 * @see org.springframework.batch.io.OutputSource#write(java.lang.Object)
+	 */
+	public void write(Object output) {
+
+		if (!initialized) {
+			open();
+		}
+
+		currentRecordCount++;
+		serializer.serializeObject(output);
+	}
+
+	/**
+	 * Get the restart data.
+	 * @return the restart data
+	 * @see org.springframework.batch.restart.Restartable#getRestartData()
+	 */
+	public RestartData getRestartData() {
+
+		Properties properties = new Properties();
+
+		properties.setProperty(RESTART_DATA_NAME, String.valueOf(getPosition()));
+
+		return new GenericRestartData(properties);
+	}
+
+	/**
+	 * Restore processing from provided restart data.
+	 * @param data the restart data
+	 * @see org.springframework.batch.restart.Restartable#restoreFrom(org.springframework.batch.restart.RestartData)
+	 */
+	public void restoreFrom(RestartData data) {
+
+		long startAtPosition = 0;
+
+		//if restart data is provided, restart from provided offset
+		//otherwise start from beginning
+		if (data != null && data.getProperties() != null
+				&& data.getProperties().getProperty(RESTART_DATA_NAME) != null) {
+			startAtPosition = Long.parseLong(data.getProperties().getProperty(
+					RESTART_DATA_NAME));
+			restarted = true;
+		}
+
+		if (!initialized) {
+			open(startAtPosition);
+		}
+
+	}
+
+	/**
+	 * Get actual statistics for output source.
+	 * @return
+	 * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics()
+	 */
+	public Properties getStatistics() {
+		Properties p = new Properties();
+		p.setProperty(WRITE_STATISTICS_NAME, String.valueOf(currentRecordCount));
+		return p;
+	}
+
+	/*
+	 * Get the actual position in file channel.
+	 * This method flushes any buffered data before position is read.
+	 *
+	 * @return byte offset in file channel
+	 */
+	private long getPosition() {
+
+		long position;
+
+		try {
+			eventWriter.flush();
+			position = channel.position();
+		} catch (Exception e) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + resource + "]", e);
+		}
+
+		return position;
+	}
+
+	/*
+	 * Set the file channel position.
+	 *
+	 * @param newPosition new file channel position
+	 */
+	private void setPosition(long newPosition) {
+
+		try {
+			Assert.state(channel.size() >= lastCommitPointPosition,
+					"Current file size is smaller than size at last commit");
+			channel.truncate(newPosition);
+			channel.position(newPosition);
+		} catch (IOException e) {
+			throw new DataAccessResourceFailureException(
+					"Unable to write to file resource: [" + resource + "]", e);
+		}
+
+
+	}
+
+	/**
+	 * Encapsulates transaction events for the StaxEventWriterOutputSource.
+	 */
+	private class StaxEventWriterOutputSourceTransactionSychronization extends
+			TransactionSynchronizationAdapter {
+
+		public void afterCompletion(int status) {
+			if (status == TransactionSynchronization.STATUS_COMMITTED) {
+				transactionComitted();
+			} else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
+				transactionRolledback();
+			}
+		}
+
+		private void transactionComitted() {
+			lastCommitPointPosition = getPosition();
+			lastCommitPointRecordCount = currentRecordCount;
+		}
+
+		private void transactionRolledback() {
+			currentRecordCount = lastCommitPointRecordCount;
+
+			//close output
+			close();
+			//and reopen it - we do this because we need to reopen stream
+			//reader at specified position - calling setPosition() is not enough!
+			restarted = true;
+			open(lastCommitPointPosition);
+		}
+
+	}
+
+	TransactionSynchronization getSynchronization() {
+		return synchronization;
+	}
+
+
+}

infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java

@@ -0,0 +1,22 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+
+/**
+ * XMLEventReader with transactional capabilities (ability to rollback to last commit point).
+ * 
+ * @author Robert Kasanicky
+ */
+interface TransactionalEventReader extends XMLEventReader{
+
+	/**
+	 * Callback on transaction rollback.
+	 */
+	public void onRollback();
+
+	/**
+	 * Callback on transaction commit.
+	 */
+	public void onCommit();
+
+}

infrastructure/src/main/java/org/springframework/batch/io/support/FileUtils.java

@@ -0,0 +1,61 @@
+package org.springframework.batch.io.support;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.util.Assert;
+
+/**
+ * Utility methods for files used in batch processing.
+ * 
+ * @author peter.zozom
+ */
+public class FileUtils {
+
+	/**
+	 * Set up output file for batch processing. This method implements common logic for 
+	 * handling output files when starting or restarting job/step.
+	 * <p> When starting output file processing, method creates/overwrites new file. 
+	 * When restaring output file processing, method checks whether file is writable. 
+	 * 
+	 * @param file file to be set up
+	 * @param restarted TRUE signalizes that we are restarting output file processing
+	 * @param overwriteOutputFile If set to TRUE, output file will be overwritten 
+	 * (this flag is ignored when processing is restart)
+	 * 
+	 * @throws IllegalArgumentException when file is NULL
+	 * @throws IllegalStateException when staring output file processing, file exists and 
+	 * flag "shouldDeleteExisting" is set to FALSE 
+	 * @throws DataAccessResourceFailureException when unable to create file or file is not writable
+	 */
+	public static void setUpOutputFile(File file, boolean restarted,
+			boolean overwriteOutputFile) {
+
+		Assert.notNull(file);
+
+		try {
+			if (!restarted) {
+				if (file.exists()) {
+					Assert.state(overwriteOutputFile, "File already exists: ["
+							+ file.getAbsolutePath() + "]");
+					file.delete();
+				}
+
+				if (file.getParent() != null ) {
+					new File(file.getParent()).mkdirs();
+				}
+				file.createNewFile();
+			}
+		} catch (IOException ioe) {
+			throw new DataAccessResourceFailureException(
+					"Unable to create file: [" + file.getAbsolutePath() + "]",
+					ioe);
+		}
+
+		if (!file.canWrite()) {
+			throw new DataAccessResourceFailureException(
+					"File is not writable: [" + file.getAbsolutePath() + "]");
+		}
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java

@@ -0,0 +1,128 @@
+/*
+ * 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.oxm;
+
+import java.io.IOException;
+
+import javax.xml.namespace.NamespaceContext;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLEventWriter;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.XMLEvent;
+import javax.xml.transform.Result;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.oxm.Marshaller;
+import org.springframework.oxm.XmlMappingException;
+
+import junit.framework.TestCase;
+
+/**
+ *
+ *
+ * @author Lucas Ward
+ *
+ */
+public class MarshallingObjectToXmlSerializerTests extends TestCase {
+
+	MarshallingObjectToXmlSerializer xmlSerializer;
+
+	MockMarshaller mockMarshaller = new MockMarshaller();
+
+	protected void setUp() throws Exception {
+		super.setUp();
+
+		xmlSerializer = new MarshallingObjectToXmlSerializer(mockMarshaller);
+		xmlSerializer.setEventWriter(new StubXmlEventWriter());
+	}
+
+	public void testSuccessfulWrite(){
+
+		Object objectToOutput = new Object();
+		xmlSerializer.serializeObject(objectToOutput);
+		assertEquals(objectToOutput, mockMarshaller.getMarshalledObject());
+	}
+
+	public void testUnsucessfulWrite(){
+
+		mockMarshaller.setThrowException(true);
+		try{
+			xmlSerializer.serializeObject(new Object());
+		}catch(DataAccessResourceFailureException ex){
+			//expected
+		}
+	}
+
+	private class MockMarshaller implements Marshaller{
+
+		private Object marshalledObject;
+		private boolean throwException = false;
+
+		public void marshal(Object arg0, Result arg1)
+				throws XmlMappingException, IOException {
+			if(throwException){
+				throw new IOException();
+			}
+			marshalledObject = arg0;
+		}
+
+		public boolean supports(Class arg0) {
+			return false;
+		}
+
+		public Object getMarshalledObject() {
+			return marshalledObject;
+		}
+
+		public void setThrowException(boolean throwException) {
+			this.throwException = throwException;
+		}
+	}
+
+	private class StubXmlEventWriter implements XMLEventWriter{
+
+		public void add(XMLEvent arg0) throws XMLStreamException {		}
+
+		public void add(XMLEventReader arg0) throws XMLStreamException {		}
+
+		public void close() throws XMLStreamException {
+		}
+
+		public void flush() throws XMLStreamException {
+		}
+
+		public NamespaceContext getNamespaceContext() {
+			return null;
+		}
+
+		public String getPrefix(String arg0) throws XMLStreamException {
+			return null;
+		}
+
+		public void setDefaultNamespace(String arg0) throws XMLStreamException {
+		}
+
+		public void setNamespaceContext(NamespaceContext arg0)
+				throws XMLStreamException {
+		}
+
+		public void setPrefix(String arg0, String arg1)
+				throws XMLStreamException {
+		}
+
+	}
+}
+

infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java

@@ -0,0 +1,119 @@
+package org.springframework.batch.io.stax;
+
+import java.util.NoSuchElementException;
+
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.EndElement;
+import javax.xml.stream.events.StartElement;
+import javax.xml.stream.events.XMLEvent;
+
+import junit.framework.TestCase;
+
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+
+/**
+ * Tests for {@link DefaultFragmentEventReader}.
+ * 
+ * @author Robert Kasanicky
+ */
+public class DefaultFragmentEventReaderTests extends TestCase {
+
+	// object under test
+	private FragmentEventReader fragmentReader;
+
+	// wrapped event fragmentReader
+	private XMLEventReader eventReader;
+
+	// test xml input
+	private String xml = "<root> <fragment> <misc1/> </fragment> <misc2/> <fragment> </fragment> </root>";
+
+	/**
+	 * Setup the fragmentReader to read the test input.
+	 */
+	protected void setUp() throws Exception {
+		Resource input = new ByteArrayResource(xml.getBytes());
+		eventReader = new DefaultTransactionalEventReader(XMLInputFactory.newInstance().createXMLEventReader(
+				input.getInputStream()));
+		fragmentReader = new DefaultFragmentEventReader(eventReader);
+	}
+
+	/**
+	 * Marked element should be wrapped with StartDocument and EndDocument
+	 * events. 
+	 * Test uses redundant peek() calls before nextEvent() in important moments to assure
+	 * peek() has no side effects on the inner state of reader.
+	 */
+	public void testFragmentWrapping() throws XMLStreamException {
+
+		moveCursorToNextElementEvent(); // move to root start
+		fragmentReader.nextEvent(); // skip root
+		moveCursorToNextElementEvent(); // move to fragment start
+
+		fragmentReader.markStartFragment(); // mark the fragment
+		assertTrue(EventHelper.startElementName(eventReader.peek()).equals("fragment"));
+		
+		// StartDocument inserted before StartElement
+		assertTrue(fragmentReader.peek().isStartDocument());
+		assertTrue(fragmentReader.nextEvent().isStartDocument());
+		// StartElement follows in the next step
+		assertTrue(EventHelper.startElementName(fragmentReader.nextEvent()).equals("fragment")); 
+
+		moveCursorToNextElementEvent(); // misc1 start
+		fragmentReader.nextEvent(); // skip it
+		moveCursorToNextElementEvent(); // misc1 end
+		fragmentReader.nextEvent(); // skip it
+		moveCursorToNextElementEvent(); // move to end of fragment
+
+		// expected EndElement, peek first which should have no side effect
+		assertTrue(EventHelper.endElementName(fragmentReader.nextEvent()).equals("fragment")); 
+		// inserted EndDocument
+		assertTrue(fragmentReader.peek().isEndDocument());
+		assertTrue(fragmentReader.nextEvent().isEndDocument()); 
+		
+		// now the reader should behave like the document has finished
+		assertTrue(fragmentReader.peek() == null);
+		
+		try{
+			fragmentReader.nextEvent();
+			fail("nextEvent should simulate behavior as if document ended");
+		}
+		catch (NoSuchElementException expected) {
+			//expected
+		}
+
+	}
+
+	/**
+	 * When fragment is marked as processed the cursor is moved after the end of
+	 * the fragment.
+	 */
+	public void testMarkFragmentProcessed() throws XMLStreamException {
+		moveCursorToNextElementEvent(); // move to root start
+		fragmentReader.nextEvent(); // skip root
+		moveCursorToNextElementEvent(); // move to fragment start
+
+		fragmentReader.markStartFragment(); // mark the fragment start
+		
+		// read only one event to move inside the fragment
+		XMLEvent startFragment = fragmentReader.nextEvent(); 
+		assertTrue(startFragment.isStartDocument());
+		fragmentReader.markFragmentProcessed(); // mark fragment as processed
+
+		fragmentReader.nextEvent(); // skip whitespace
+		// the next element after fragment end is <misc2/>
+		XMLEvent misc2 = fragmentReader.nextEvent(); 
+		assertTrue(EventHelper.startElementName(misc2).equals("misc2"));
+	}
+
+	private void moveCursorToNextElementEvent() throws XMLStreamException {
+		XMLEvent event = eventReader.peek();
+		while (!(event instanceof StartElement) && !(event instanceof EndElement)) {
+			eventReader.nextEvent();
+			event = eventReader.peek();
+		}
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java

@@ -0,0 +1,51 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLInputFactory;
+
+import junit.framework.TestCase;
+
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+
+/**
+ * Tests for {@link DefaultTransactionalEventReader}.
+ * 
+ * @author Robert Kasanicky
+ */
+public class DefaultTransactionalEventReaderTests extends TestCase {
+
+	// object under test
+	private TransactionalEventReader reader;
+
+	// test xml input
+	private String xml = "<root> <fragment> <misc1/> </fragment> <misc2/> <fragment> </fragment> </root>";
+
+	
+	protected void setUp() throws Exception {
+		Resource resource = new ByteArrayResource(xml.getBytes());
+		XMLEventReader wrappedReader = XMLInputFactory.newInstance().createXMLEventReader(resource.getInputStream());
+		reader = new DefaultTransactionalEventReader(wrappedReader);
+	}
+
+	/**
+	 * Rollback scenario.
+	 */
+	public void testRollback() throws Exception {
+		reader.nextEvent(); //start document
+		reader.nextEvent(); //start root element
+		reader.nextEvent(); //whitespace
+		
+		reader.onCommit(); // commit point
+		
+		assertTrue(EventHelper.startElementName(reader.nextEvent()).equals("fragment"));
+		reader.nextEvent(); //whitespace
+		assertTrue(EventHelper.startElementName(reader.nextEvent()).equals("misc1"));
+		assertTrue(EventHelper.endElementName(reader.peek()).equals("misc1"));
+		
+		reader.onRollback(); // now we should be at the last commit point
+		assertTrue(EventHelper.startElementName(reader.nextEvent()).equals("fragment"));
+		reader.nextEvent();
+		assertTrue(EventHelper.startElementName(reader.nextEvent()).equals("misc1"));
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java

@@ -0,0 +1,30 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.events.EndElement;
+import javax.xml.stream.events.StartElement;
+import javax.xml.stream.events.XMLEvent;
+
+/**
+ * Helper methods for working with XML Events.
+ * 
+ * @author Robert Kasanicky
+ */
+class EventHelper {
+
+	//utility class
+	private EventHelper() {}
+	
+	/**
+	 * @return element name assuming the event is instance of StartElement
+	 */
+	public static String startElementName(XMLEvent event) {
+		return ((StartElement) event).getName().getLocalPart();
+	}
+	
+	/**
+	 * @return element name assuming the event is instance of EndElement
+	 */
+	public static String endElementName(XMLEvent event) {
+		return ((EndElement) event).getName().getLocalPart();
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java

@@ -0,0 +1,37 @@
+package org.springframework.batch.io.stax;
+
+import javax.xml.stream.XMLEventFactory;
+import javax.xml.stream.events.XMLEvent;
+
+import junit.framework.TestCase;
+
+/**
+ * Tests for {@link EventSequence}
+ * 
+ * @author Robert Kasanicky
+ */
+public class EventSequenceTests extends TestCase {
+
+	// object under test
+	private EventSequence seq = new EventSequence();
+	
+	private XMLEventFactory factory = XMLEventFactory.newInstance();
+
+	/**
+	 * Common usage scenario.
+	 */
+	public void testCommonUse() {
+		XMLEvent event1 = factory.createComment("testString1");
+		XMLEvent event2 = factory.createCData("testString2");
+		seq.addEvent(event1);
+		seq.addEvent(event2);
+		
+		assertTrue(seq.hasNext());
+		assertSame(event1, seq.nextEvent());
+		assertTrue(seq.hasNext());
+		assertSame(event2, seq.nextEvent());
+		assertFalse(seq.hasNext());
+		assertNull(seq.nextEvent());
+		
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventReaderInputSourceTests.java

@@ -0,0 +1,237 @@
+package org.springframework.batch.io.stax;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import javax.xml.stream.FactoryConfigurationError;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.EndElement;
+import javax.xml.stream.events.XMLEvent;
+
+import junit.framework.TestCase;
+
+import org.springframework.batch.restart.RestartData;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+import org.springframework.transaction.support.TransactionSynchronization;
+
+/**
+ * Tests for {@link StaxEventReaderInputSource}.
+ * 
+ * @author Robert Kasanicky
+ */
+public class StaxEventReaderInputSourceTests extends TestCase {
+
+	// object under test
+	private StaxEventReaderInputSource source;
+
+	// test xml input
+	private String xml = "<root> <fragment> <misc1/> </fragment> <misc2/> <fragment> testString </fragment> </root>";
+
+	private FragmentDeserializer deserializer = new FragmentDeserializerMock();
+
+	private static final String FRAGMENT_ROOT_ELEMENT = "fragment";
+
+	protected void setUp() throws Exception {
+		source = createNewInputSouce();
+	}
+
+	/**
+	 * InputSource should pass XML fragments to deserializer wrapped with
+	 * StartDocument and EndDocument events.
+	 */
+	public void testFragmentWrapping() {
+		// see asserts in the mock deserializer
+		assertNotNull(source.read());
+		assertNotNull(source.read());
+		assertNull(source.read()); // there are only two fragments
+	}
+
+	/**
+	 * Cursor is moved before beginning of next fragment.
+	 */
+	public void testMoveCursorToNextFragment() throws XMLStreamException, FactoryConfigurationError, IOException {
+		Resource resource = new ByteArrayResource(xml.getBytes());
+		XMLEventReader reader = XMLInputFactory.newInstance().createXMLEventReader(resource.getInputStream());
+
+		final int EXPECTED_NUMBER_OF_FRAGMENTS = 2;
+		for (int i = 0; i < EXPECTED_NUMBER_OF_FRAGMENTS; i++) {
+			assertTrue(source.moveCursorToNextFragment(reader));
+			assertTrue(EventHelper.startElementName(reader.peek()).equals("fragment"));
+			reader.nextEvent(); // move away from beginning of fragment
+		}
+		assertFalse(source.moveCursorToNextFragment(reader));
+	}
+
+	/**
+	 * Save restart data and restore from it.
+	 */
+	public void testRestart() {
+		source.read();
+		RestartData restartData = source.getRestartData();
+		List expectedAfterRestart = (List) source.read();
+
+		source = createNewInputSouce();
+		source.restoreFrom(restartData);
+		List afterRestart = (List) source.read();
+		assertEquals(expectedAfterRestart.size(), afterRestart.size());
+	}
+
+	/**
+	 * Skipping marked records after rollback.
+	 */
+	public void testSkip() {
+		List first = (List) source.read();
+		source.skip();
+		List second = (List) source.read();
+		assertFalse(first.equals(second));
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+
+		assertEquals(second, source.read());
+	}
+
+	/**
+	 * Rollback to last commited record. 
+	 */
+	public void testRollback() {
+		
+		//rollback between deserializing records
+		List first = (List) source.read();
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		List second = (List) source.read();
+		assertFalse(first.equals(second));
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+
+		assertEquals(second, source.read());
+		
+		
+		//rollback while deserializing record
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		source.setFragmentDeserializer(new ExceptionFragmentDeserializer());
+		try {
+			source.read();
+		}
+		catch (Exception expected) {
+			source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		}
+		source.setFragmentDeserializer(deserializer);
+	
+		assertEquals(second, source.read());
+	}
+
+	/**
+	 * Statistics return the current record count. Calling read after end of
+	 * input does not increase the counter.
+	 */
+	public void testStatistics() {
+		final int NUMBER_OF_RECORDS = 2;
+
+		for (int i = 0; i < NUMBER_OF_RECORDS; i++) {
+			int recordCount = extractRecordCountFrom(source.getStatistics());
+			assertEquals(i, recordCount);
+			source.read();
+		}
+
+		assertEquals(NUMBER_OF_RECORDS, extractRecordCountFrom(source.getStatistics()));
+		source.read();
+		assertEquals(NUMBER_OF_RECORDS, extractRecordCountFrom(source.getStatistics()));
+	}
+
+	private int extractRecordCountFrom(Properties statistics) {
+		return Integer.valueOf(
+				source.getStatistics().getProperty(StaxEventReaderInputSource.READ_COUNT_STATISTICS_NAME)).intValue();
+	}
+
+	private StaxEventReaderInputSource createNewInputSouce() {
+		Resource resource = new ByteArrayResource(xml.getBytes());
+
+		StaxEventReaderInputSource newSource = new StaxEventReaderInputSource();
+		newSource.setResource(resource);
+
+		newSource.setFragmentRootElementName(FRAGMENT_ROOT_ELEMENT);
+		newSource.setFragmentDeserializer(deserializer);
+
+		return newSource;
+	}
+
+	/**
+	 * A simple XMLEvent deserializer mock - check for the start and end
+	 * document events for the fragment root & end tags + skips the fragment
+	 * contents.
+	 */
+	private static class FragmentDeserializerMock implements FragmentDeserializer {
+
+		/**
+		 * A simple mapFragment implementation checking the
+		 * StaxEventReaderInputSource basic read functionality.
+		 * @param eventReader
+		 * @return list of the events from fragment body
+		 */
+		public Object deserializeFragment(XMLEventReader eventReader) {
+			List fragmentContent;
+			try {
+				// first event should be StartDocument
+				XMLEvent event1 = eventReader.nextEvent();
+				assertTrue(event1.isStartDocument());
+
+				// second should be StartElement of the fragment
+				XMLEvent event2 = eventReader.nextEvent();
+				assertTrue(event2.isStartElement());
+				assertTrue(EventHelper.startElementName(event2).equals(FRAGMENT_ROOT_ELEMENT));
+
+				// jump before the end of fragment
+				fragmentContent = readRecordsInsideFragment(eventReader);
+
+				// end of fragment
+				XMLEvent event3 = eventReader.nextEvent();
+				assertTrue(event3.isEndElement());
+				assertTrue(EventHelper.endElementName(event3).equals(FRAGMENT_ROOT_ELEMENT));
+
+				// EndDocument should follow the end of fragment
+				XMLEvent event4 = eventReader.nextEvent();
+				assertTrue(event4.isEndDocument());
+
+			}
+			catch (XMLStreamException e) {
+				throw new RuntimeException("Error occured in FragmentDeserializer", e);
+			}
+			return fragmentContent;
+		}
+
+		/**
+		 * Skips the XML fragment contents.
+		 */
+		private List readRecordsInsideFragment(XMLEventReader eventReader) throws XMLStreamException {
+			XMLEvent eventInsideFragment;
+			List events = new ArrayList();
+			do {
+				eventInsideFragment = eventReader.peek();
+				if (eventInsideFragment instanceof EndElement
+						&& ((EndElement) eventInsideFragment).getName().getLocalPart().equals(FRAGMENT_ROOT_ELEMENT)) {
+					break;
+				}
+				events.add(eventReader.nextEvent());
+			} while (eventInsideFragment != null);
+
+			return events;
+		}
+
+	}
+
+	/**
+	 * Moves cursor inside the fragment body and causes rollback.
+	 */
+	private class ExceptionFragmentDeserializer implements FragmentDeserializer {
+
+		public Object deserializeFragment(XMLEventReader eventReader) {
+			eventReader.next();
+			throw new RuntimeException();
+		}
+
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java

@@ -0,0 +1,200 @@
+package org.springframework.batch.io.stax;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+
+import javax.xml.stream.XMLEventFactory;
+import javax.xml.stream.XMLStreamException;
+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.restart.RestartData;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.core.io.Resource;
+import org.springframework.oxm.Marshaller;
+import org.springframework.oxm.XmlMappingException;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.util.Assert;
+import org.springframework.xml.transform.StaxResult;
+
+/**
+ * Tests for {@link StaxStreamWriterOutputSource}.
+ */
+public class StaxEventWriterOutputSourceTests extends TestCase {
+
+	// object under test
+	private StaxEventWriterOutputSource source;
+
+	// output file
+	private Resource resource;
+
+	// test record for writing to output
+	private Object record = new Object() {
+		public String toString() {
+			return TEST_STRING;
+		}
+	};
+
+	private static final String TEST_STRING = "StaxEventWriterOutputSourceTests-testString";
+
+	private static final int NOT_FOUND = -1;
+
+
+	protected void setUp() throws Exception {
+		resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", "xml"));
+		source = newOutputSource();
+	}
+
+	/**
+	 * Write should pass its argument and StaxResult object to Serializer
+	 */
+	public void testWrite() throws Exception {
+		Marshaller marshaller = new InputCheckMarshaller();
+		MarshallingObjectToXmlSerializer serializer = new MarshallingObjectToXmlSerializer(marshaller);
+		source.setSerializer(serializer);
+
+		// see asserts in the marshaller
+		source.write(record);
+
+	}
+
+	/**
+	 * Rolled back records should not be written to output file.
+	 */
+	public void testRollback() throws Exception {
+		source.write(record);
+
+		// rollback
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		assertEquals("", outputFileContent());
+	}
+
+	/**
+	 * Commited output is written to the output file.
+	 */
+	public void testCommit() throws Exception {
+		source.write(record);
+
+		// commit
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		assertTrue(outputFileContent().indexOf(TEST_STRING) != NOT_FOUND);
+	}
+
+	/**
+	 * Restart scenario - content is appended to the output file after restart.
+	 */
+	public void testRestart() throws Exception {
+		// write records
+		source.write(record);
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		RestartData restartData = source.getRestartData();
+
+		// create new output source from saved restart data and continue writing
+		source = newOutputSource();
+		source.restoreFrom(restartData);
+		source.write(record);
+		source.close();
+
+		// check the output is concatenation of 'before restart' and 'after restart' writes.
+		String outputFile = outputFileContent();
+		int firstRecord = outputFile.indexOf(TEST_STRING);
+		int secondRecord = outputFile.indexOf(TEST_STRING, firstRecord + TEST_STRING.length());
+		int thirdRecord = outputFile.indexOf(TEST_STRING, secondRecord + TEST_STRING.length());
+
+		// (two records should be written)
+		assertTrue(firstRecord != NOT_FOUND);
+		assertTrue(secondRecord != NOT_FOUND);
+		assertEquals(NOT_FOUND, thirdRecord);
+	}
+
+	/**
+	 * Count of 'records written so far' is returned as statistics.
+	 */
+	public void testStatistics() throws Exception {
+		final int NUMBER_OF_RECORDS = 10;
+		for (int i = 0; i < NUMBER_OF_RECORDS; i++) {
+			String writeStatistics =
+				source.getStatistics().getProperty(StaxEventWriterOutputSource.WRITE_STATISTICS_NAME);
+
+			assertEquals(String.valueOf(i), writeStatistics);
+			source.write(record);
+		}
+	}
+
+	/**
+	 * Open method writes the root tag, close method adds corresponding end tag.
+	 */
+	public void testOpenAndClose() throws IOException {
+		source.setRootTagName("testroot");
+		source.setRootElementAttributes(new HashMap() {{
+			put("attribute", "value");
+		}});
+		source.open();
+		source.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+
+		assertTrue(outputFileContent().indexOf("<testroot attribute=\"value\"") != NOT_FOUND);
+
+		source.close();
+		assertTrue(outputFileContent().endsWith("</testroot>"));
+	}
+
+	/**
+	 * Checks the received parameters.
+	 */
+	private class InputCheckMarshaller implements Marshaller {
+		public void marshal(Object graph, Result result) {
+			assertTrue(result instanceof StaxResult);
+			assertSame(record, graph);
+		}
+
+		public boolean supports(Class clazz) {
+			return true;
+		}
+	}
+
+	/**
+	 * Writes object's toString representation as XML comment.
+	 */
+	private static class SimpleMarshaller implements Marshaller {
+		public void marshal(Object graph, Result result) throws XmlMappingException, IOException {
+			Assert.isInstanceOf(StaxResult.class, result);
+
+			StaxResult staxResult = (StaxResult) result;
+			try {
+				staxResult.getXMLEventWriter().add(XMLEventFactory.newInstance().createComment(graph.toString()));
+			}
+			catch (XMLStreamException e) {
+				throw new RuntimeException("Exception while writing to output file", e);
+			}
+		}
+
+		public boolean supports(Class clazz) {
+			return true;
+		}
+	}
+
+	/**
+	 * @return output file content as String
+	 */
+	private String outputFileContent() throws IOException {
+		return FileUtils.readFileToString(resource.getFile(), null);
+	}
+
+	/**
+	 * @return new instance of fully configured output source
+	 */
+	private StaxEventWriterOutputSource newOutputSource() {
+		StaxEventWriterOutputSource source = new StaxEventWriterOutputSource();
+		source.setResource(resource);
+
+		Marshaller marshaller = new SimpleMarshaller();
+		MarshallingObjectToXmlSerializer serializer = new MarshallingObjectToXmlSerializer(marshaller);
+		source.setSerializer(serializer);
+
+		return source;
+	}
+}