diff --git a/infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java b/infrastructure/src/main/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializer.java new file mode 100644 index 000000000..c42c8fd36 --- /dev/null +++ b/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); + } + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java b/infrastructure/src/main/java/org/springframework/batch/io/oxm/UnmarshallingFragmentDeserializer.java new file mode 100644 index 000000000..9079f2559 --- /dev/null +++ b/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; + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/AbstractEventReaderWrapper.java new file mode 100644 index 000000000..782790c21 --- /dev/null +++ b/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(); + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultFragmentEventReader.java new file mode 100644 index 000000000..43829e51c --- /dev/null +++ b/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; + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/DefaultTransactionalEventReader.java new file mode 100644 index 000000000..80fb169f1 --- /dev/null +++ b/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(); + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/EventSequence.java new file mode 100644 index 000000000..05a84b0ec --- /dev/null +++ b/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(); + } + + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentDeserializer.java new file mode 100644 index 000000000..7db2506c9 --- /dev/null +++ b/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); +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/FragmentEventReader.java new file mode 100644 index 000000000..d9e4c09ee --- /dev/null +++ b/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(); + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/NoStartEndDocumentStreamWriter.java new file mode 100644 index 000000000..a522c2291 --- /dev/null +++ b/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); + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/ObjectToXmlSerializer.java new file mode 100644 index 000000000..5ac416078 --- /dev/null +++ b/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); +} 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 new file mode 100644 index 000000000..9aa8f7af7 --- /dev/null +++ b/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 true if next fragment was found, false 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); + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/StaxEventWriterOutputSource.java new file mode 100644 index 000000000..79deb14ed --- /dev/null +++ b/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: + * + * 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(("").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; + } + + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java b/infrastructure/src/main/java/org/springframework/batch/io/stax/TransactionalEventReader.java new file mode 100644 index 000000000..2f75b67fd --- /dev/null +++ b/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(); + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/support/FileUtils.java b/infrastructure/src/main/java/org/springframework/batch/io/support/FileUtils.java new file mode 100644 index 000000000..7d8deda65 --- /dev/null +++ b/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. + *

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() + "]"); + } + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java b/infrastructure/src/test/java/org/springframework/batch/io/oxm/MarshallingObjectToXmlSerializerTests.java new file mode 100644 index 000000000..450c73ac1 --- /dev/null +++ b/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 { + } + + } +} + diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultFragmentEventReaderTests.java new file mode 100644 index 000000000..5438fb98a --- /dev/null +++ b/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 = " "; + + /** + * 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 + 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(); + } + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/DefaultTransactionalEventReaderTests.java new file mode 100644 index 000000000..6a7ca743c --- /dev/null +++ b/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 = " "; + + + 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")); + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/EventHelper.java new file mode 100644 index 000000000..14223c326 --- /dev/null +++ b/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(); + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/EventSequenceTests.java new file mode 100644 index 000000000..7a96d6552 --- /dev/null +++ b/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()); + + } +} 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 new file mode 100644 index 000000000..f95cb37be --- /dev/null +++ b/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 = " testString "; + + 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(); + } + + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java b/infrastructure/src/test/java/org/springframework/batch/io/stax/StaxEventWriterOutputSourceTests.java new file mode 100644 index 000000000..529adb878 --- /dev/null +++ b/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("")); + } + + /** + * 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; + } +}