diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index 9b4468fae..14257516d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -24,9 +24,11 @@ import java.io.OutputStreamWriter; import java.io.UnsupportedEncodingException; import java.io.Writer; import java.nio.channels.FileChannel; +import java.util.Collections; import java.util.List; import java.util.Map; +import javax.xml.namespace.QName; import javax.xml.stream.FactoryConfigurationError; import javax.xml.stream.XMLEventFactory; import javax.xml.stream.XMLEventWriter; @@ -44,6 +46,8 @@ import org.springframework.batch.item.file.ResourceAwareItemWriterItemStream; import org.springframework.batch.item.support.AbstractItemStreamItemWriter; import org.springframework.batch.item.util.FileUtils; import org.springframework.batch.item.xml.stax.NoStartEndDocumentStreamWriter; +import org.springframework.batch.item.xml.stax.UnclosedElementCollectingEventWriter; +import org.springframework.batch.item.xml.stax.UnopenedElementClosingEventWriter; import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; @@ -86,6 +90,9 @@ ResourceAwareItemWriterItemStream, InitializingBean { // restart data property name private static final String RESTART_DATA_NAME = "position"; + // unclosed header callback elements property name + private static final String UNCLOSED_HEADER_CALLBACK_ELEMENTS_NAME = "unclosedHeaderCallbackElements"; + // restart data property name private static final String WRITE_STATISTICS_NAME = "record.count"; @@ -143,6 +150,11 @@ ResourceAwareItemWriterItemStream, InitializingBean { private boolean forceSync; private boolean shouldDeleteIfEmpty = false; + + private boolean restarted = false; + + // List holding the QName of elements that were opened in the header callback, but not closed + private List unclosedHeaderCallbackElements = Collections.EMPTY_LIST; public StaxEventItemWriter() { setExecutionContextName(ClassUtils.getShortName(StaxEventItemWriter.class)); @@ -353,6 +365,7 @@ ResourceAwareItemWriterItemStream, InitializingBean { * * @see org.springframework.batch.item.ItemStream#open(ExecutionContext) */ + @SuppressWarnings("unchecked") @Override public void open(ExecutionContext executionContext) { super.open(executionContext); @@ -360,13 +373,17 @@ ResourceAwareItemWriterItemStream, InitializingBean { Assert.notNull(resource, "The resource must be set"); long startAtPosition = 0; - boolean restarted = false; - + // if restart data is provided, restart from provided offset // otherwise start from beginning if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) { startAtPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME)); currentRecordCount = executionContext.getLong(getExecutionContextKey(WRITE_STATISTICS_NAME)); + if (executionContext.containsKey(getExecutionContextKey(UNCLOSED_HEADER_CALLBACK_ELEMENTS_NAME))) { + unclosedHeaderCallbackElements = (List) executionContext + .get(getExecutionContextKey(UNCLOSED_HEADER_CALLBACK_ELEMENTS_NAME)); + } + restarted = true; if (shouldDeleteIfEmpty && currentRecordCount == 0) { // previous execution deleted the output file because no items were written @@ -375,14 +392,19 @@ ResourceAwareItemWriterItemStream, InitializingBean { } else { restarted = true; } + } else { + currentRecordCount = 0; + restarted = false; } - open(startAtPosition, restarted); + open(startAtPosition); if (startAtPosition == 0) { try { if (headerCallback != null) { - headerCallback.write(delegateEventWriter); + UnclosedElementCollectingEventWriter headerCallbackWriter = new UnclosedElementCollectingEventWriter(delegateEventWriter); + headerCallback.write(headerCallbackWriter); + unclosedHeaderCallbackElements = headerCallbackWriter.getUnclosedElements(); } } catch (IOException e) { @@ -395,7 +417,8 @@ ResourceAwareItemWriterItemStream, InitializingBean { /** * Helper method for opening output source at given file position */ - private void open(long position, boolean restarted) { + @SuppressWarnings("resource") + private void open(long position) { File file; FileOutputStream os = null; @@ -640,7 +663,12 @@ ResourceAwareItemWriterItemStream, InitializingBean { try { if (footerCallback != null) { - footerCallback.write(delegateEventWriter); + XMLEventWriter footerCallbackWriter = delegateEventWriter; + if (restarted && !unclosedHeaderCallbackElements.isEmpty()) { + footerCallbackWriter = new UnopenedElementClosingEventWriter( + delegateEventWriter, bufferedWriter, unclosedHeaderCallbackElements); + } + footerCallback.write(footerCallbackWriter); } delegateEventWriter.flush(); endDocument(delegateEventWriter); @@ -737,6 +765,10 @@ ResourceAwareItemWriterItemStream, InitializingBean { Assert.notNull(executionContext, "ExecutionContext must not be null"); executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), getPosition()); executionContext.putLong(getExecutionContextKey(WRITE_STATISTICS_NAME), currentRecordCount); + if (!unclosedHeaderCallbackElements.isEmpty()) { + executionContext.put(getExecutionContextKey(UNCLOSED_HEADER_CALLBACK_ELEMENTS_NAME), + unclosedHeaderCallbackElements); + } } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriter.java new file mode 100644 index 000000000..5230d5ffd --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriter.java @@ -0,0 +1,54 @@ +/* + * Copyright 2006-2013 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.item.xml.stax; + +import java.util.LinkedList; +import java.util.List; + +import javax.xml.namespace.QName; +import javax.xml.stream.XMLEventWriter; +import javax.xml.stream.XMLStreamException; +import javax.xml.stream.events.XMLEvent; + +/** + * Delegating XMLEventWriter, which collects the QNames of elements that were opened but not closed. + * + * @author Jimmy Praet + */ +public class UnclosedElementCollectingEventWriter extends AbstractEventWriterWrapper { + + private LinkedList unclosedElements = new LinkedList(); + + public UnclosedElementCollectingEventWriter(XMLEventWriter wrappedEventWriter) { + super(wrappedEventWriter); + } + + @Override + public void add(XMLEvent event) throws XMLStreamException { + if (event.isStartElement()) { + unclosedElements.addLast(event.asStartElement().getName()); + } else if (event.isEndElement()) { + unclosedElements.removeLast(); + } + super.add(event); + } + + public List getUnclosedElements() { + return unclosedElements; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriter.java new file mode 100644 index 000000000..f380f7150 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriter.java @@ -0,0 +1,81 @@ +/* + * Copyright 2006-2013 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.item.xml.stax; + +import java.io.IOException; +import java.io.Writer; +import java.util.LinkedList; +import java.util.List; + +import javax.xml.namespace.QName; +import javax.xml.stream.XMLEventWriter; +import javax.xml.stream.XMLStreamException; +import javax.xml.stream.events.EndElement; +import javax.xml.stream.events.XMLEvent; + +import org.springframework.dao.DataAccessResourceFailureException; +import org.springframework.util.StringUtils; + +/** + * Delegating XMLEventWriter, which writes EndElement events that match a given collection of QNames directly + * to the underlying java.io.Writer instead of to the delegate XMLEventWriter. + * + * @author Jimmy Praet + */ +public class UnopenedElementClosingEventWriter extends AbstractEventWriterWrapper { + + private LinkedList unopenedElements; + + private Writer ioWriter; + + public UnopenedElementClosingEventWriter(XMLEventWriter wrappedEventWriter, Writer ioWriter, List unopenedElements) { + super(wrappedEventWriter); + this.unopenedElements = new LinkedList(unopenedElements); + this.ioWriter = ioWriter; + } + + @Override + public void add(XMLEvent event) throws XMLStreamException { + if (isUnopenedElementCloseEvent(event)) { + EndElement endElement = event.asEndElement(); + QName name = endElement.getName(); + String nsPrefix = !StringUtils.hasText(name.getPrefix()) ? "" : name.getPrefix() + ":"; + try { + super.flush(); + ioWriter.write(""); + ioWriter.flush(); + } + catch (IOException ioe) { + throw new DataAccessResourceFailureException("Unable to close tag: " + name, ioe); + } + unopenedElements.removeLast(); + } else { + super.add(event); + } + } + + private boolean isUnopenedElementCloseEvent(XMLEvent event) { + if (unopenedElements.isEmpty()) { + return false; + } else if (!event.isEndElement()) { + return false; + } else { + return unopenedElements.getLast().equals(event.asEndElement().getName()); + } + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java index aa5a67c23..1246f8d54 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java @@ -766,6 +766,153 @@ public class StaxEventItemWriterTests { writer.write(items); } + /** + * Test opening and closing corresponding tags in header- and footer callback. + */ + @Test + public void testOpenAndCloseTagsInCallbacks() throws Exception { + initWriterForSimpleCallbackTests(); + writer.open(executionContext); + writer.write(items); + writer.close(); + String content = getOutputFileContent(); + + assertEquals("Wrong content: " + content, + "", content); + } + + /** + * Test opening and closing corresponding tags in header- and footer callback (restart). + */ + @Test + public void testOpenAndCloseTagsInCallbacksRestart() throws Exception { + initWriterForSimpleCallbackTests(); + writer.open(executionContext); + writer.write(items); + writer.update(executionContext); + + initWriterForSimpleCallbackTests(); + + writer.open(executionContext); + writer.write(items); + writer.close(); + String content = getOutputFileContent(); + + assertEquals("Wrong content: " + content, + "" + + "", content); + } + + /** + * Test opening and closing corresponding tags in complex header- and footer callback (restart). + */ + @Test + public void testOpenAndCloseTagsInComplexCallbacksRestart() throws Exception { + initWriterForComplexCallbackTests(); + writer.open(executionContext); + writer.write(items); + writer.update(executionContext); + + initWriterForComplexCallbackTests(); + + writer.open(executionContext); + writer.write(items); + writer.close(); + String content = getOutputFileContent(); + + assertEquals("Wrong content: " + content, + "" + + "PRE-HEADERPOST-HEADER" + + "" + + "PRE-FOOTERPOST-FOOTER" + + "", content); + } + + private void initWriterForSimpleCallbackTests() throws Exception { + writer = createItemWriter(); + writer.setHeaderCallback(new StaxWriterCallback() { + + @Override + public void write(XMLEventWriter writer) throws IOException { + XMLEventFactory factory = XMLEventFactory.newInstance(); + try { + writer.add(factory.createStartElement("ns", "http://www.springframework.org/test", "group")); + } + catch (XMLStreamException e) { + throw new RuntimeException(e); + } + } + + }); + writer.setFooterCallback(new StaxWriterCallback() { + + @Override + public void write(XMLEventWriter writer) throws IOException { + XMLEventFactory factory = XMLEventFactory.newInstance(); + try { + writer.add(factory.createEndElement("ns", "http://www.springframework.org/test", "group")); + } + catch (XMLStreamException e) { + throw new RuntimeException(e); + } + + } + + }); + writer.setRootTagName("{http://www.springframework.org/test}ns:testroot"); + writer.afterPropertiesSet(); + } + + // more complex callbacks, writing element before and after the multiple corresponding header- and footer elements + private void initWriterForComplexCallbackTests() throws Exception { + writer = createItemWriter(); + writer.setHeaderCallback(new StaxWriterCallback() { + + @Override + public void write(XMLEventWriter writer) throws IOException { + XMLEventFactory factory = XMLEventFactory.newInstance(); + try { + writer.add(factory.createStartElement("", "", "preHeader")); + writer.add(factory.createCharacters("PRE-HEADER")); + writer.add(factory.createEndElement("", "", "preHeader")); + writer.add(factory.createStartElement("ns", "http://www.springframework.org/test", "group")); + writer.add(factory.createStartElement("", "", "subGroup")); + writer.add(factory.createStartElement("", "", "postHeader")); + writer.add(factory.createCharacters("POST-HEADER")); + writer.add(factory.createEndElement("", "", "postHeader")); + } + catch (XMLStreamException e) { + throw new RuntimeException(e); + } + } + + }); + writer.setFooterCallback(new StaxWriterCallback() { + + @Override + public void write(XMLEventWriter writer) throws IOException { + XMLEventFactory factory = XMLEventFactory.newInstance(); + try { + writer.add(factory.createStartElement("", "", "preFooter")); + writer.add(factory.createCharacters("PRE-FOOTER")); + writer.add(factory.createEndElement("", "", "preFooter")); + writer.add(factory.createEndElement("", "", "subGroup")); + writer.add(factory.createEndElement("ns", "http://www.springframework.org/test", "group")); + writer.add(factory.createStartElement("", "", "postFooter")); + writer.add(factory.createCharacters("POST-FOOTER")); + writer.add(factory.createEndElement("", "", "postFooter")); + } + catch (XMLStreamException e) { + throw new RuntimeException(e); + } + + } + + }); + writer.setRootTagName("{http://www.springframework.org/test}ns:testroot"); + writer.afterPropertiesSet(); + } + /** * Writes object's toString representation as XML comment. */ diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriterTests.java new file mode 100644 index 000000000..d85c7dc7f --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnclosedElementCollectingEventWriterTests.java @@ -0,0 +1,85 @@ +package org.springframework.batch.item.xml.stax; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import javax.xml.namespace.QName; +import javax.xml.stream.XMLEventFactory; +import javax.xml.stream.XMLEventWriter; +import javax.xml.stream.events.XMLEvent; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +/** + * Tests for {@link UnclosedElementCollectingEventWriter} + * + * @author Jimmy Praet + */ +public class UnclosedElementCollectingEventWriterTests { + + private UnclosedElementCollectingEventWriter writer; + + private XMLEventWriter wrappedWriter; + + private XMLEventFactory eventFactory = XMLEventFactory.newInstance(); + + private QName elementA = new QName("elementA"); + + private QName elementB = new QName("elementB"); + + private QName elementC = new QName("elementC"); + + @Before + public void setUp() throws Exception { + wrappedWriter = mock(XMLEventWriter.class); + writer = new UnclosedElementCollectingEventWriter(wrappedWriter); + } + + @Test + public void testNoUnclosedElements() throws Exception { + writer.add(eventFactory.createStartElement(elementA, null, null)); + writer.add(eventFactory.createEndElement(elementA, null)); + + assertEquals(0, writer.getUnclosedElements().size()); + verify(wrappedWriter, Mockito.times(2)).add(Mockito.any(XMLEvent.class)); + } + + @Test + public void testSingleUnclosedElement() throws Exception { + writer.add(eventFactory.createStartElement(elementA, null, null)); + writer.add(eventFactory.createEndElement(elementA, null)); + writer.add(eventFactory.createStartElement(elementB, null, null)); + + assertEquals(1, writer.getUnclosedElements().size()); + assertEquals(elementB, writer.getUnclosedElements().get(0)); + verify(wrappedWriter, Mockito.times(3)).add(Mockito.any(XMLEvent.class)); + } + + @Test + public void testMultipleUnclosedElements() throws Exception { + writer.add(eventFactory.createStartElement(elementA, null, null)); + writer.add(eventFactory.createStartElement(elementB, null, null)); + writer.add(eventFactory.createStartElement(elementC, null, null)); + writer.add(eventFactory.createEndElement(elementC, null)); + + assertEquals(2, writer.getUnclosedElements().size()); + assertEquals(elementA, writer.getUnclosedElements().get(0)); + assertEquals(elementB, writer.getUnclosedElements().get(1)); + verify(wrappedWriter, Mockito.times(4)).add(Mockito.any(XMLEvent.class)); + } + + @Test + public void testMultipleIdenticalUnclosedElement() throws Exception { + writer.add(eventFactory.createStartElement(elementA, null, null)); + writer.add(eventFactory.createStartElement(elementA, null, null)); + + assertEquals(2, writer.getUnclosedElements().size()); + assertEquals(elementA, writer.getUnclosedElements().get(0)); + assertEquals(elementA, writer.getUnclosedElements().get(1)); + verify(wrappedWriter, Mockito.times(2)).add(Mockito.any(XMLEvent.class)); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriterTests.java new file mode 100644 index 000000000..5d36a7182 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/stax/UnopenedElementClosingEventWriterTests.java @@ -0,0 +1,118 @@ +package org.springframework.batch.item.xml.stax; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.io.Writer; +import java.util.LinkedList; +import java.util.List; + +import javax.xml.namespace.QName; +import javax.xml.stream.XMLEventFactory; +import javax.xml.stream.XMLEventWriter; +import javax.xml.stream.events.EndElement; +import javax.xml.stream.events.StartElement; +import javax.xml.stream.events.XMLEvent; + +import junit.framework.Assert; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.dao.DataAccessResourceFailureException; + +/** + * Tests for {@link UnopenedElementClosingEventWriter} + * + * @author Jimmy Praet + */ +public class UnopenedElementClosingEventWriterTests { + + private UnopenedElementClosingEventWriter writer; + + private XMLEventWriter wrappedWriter; + + private Writer ioWriter; + + private XMLEventFactory eventFactory = XMLEventFactory.newInstance(); + + private List unopenedElements = new LinkedList(); + + private QName unopenedA = new QName("http://test", "unopened-a", "t"); + + private QName unopenedB = new QName("", "unopened-b", ""); + + private QName other = new QName("http://test", "other", "t"); + + @Before + public void setUp() throws Exception { + wrappedWriter = mock(XMLEventWriter.class); + ioWriter = mock(Writer.class); + unopenedElements.add(unopenedA); + unopenedElements.add(unopenedB); + writer = new UnopenedElementClosingEventWriter(wrappedWriter, ioWriter, unopenedElements); + } + + @Test + public void testEndUnopenedElements() throws Exception { + EndElement endElementB = eventFactory.createEndElement(unopenedB, null); + writer.add(endElementB); + + EndElement endElementA = eventFactory.createEndElement(unopenedA, null); + writer.add(endElementA); + + verify(wrappedWriter, Mockito.never()).add(endElementB); + verify(wrappedWriter, Mockito.never()).add(endElementA); + verify(wrappedWriter, Mockito.times(2)).flush(); + verify(ioWriter).write(""); + verify(ioWriter).write(""); + verify(ioWriter, Mockito.times(2)).flush(); + } + + @Test + public void testEndUnopenedElementRemovesFromList() throws Exception { + EndElement endElement = eventFactory.createEndElement(unopenedB, null); + writer.add(endElement); + + verify(wrappedWriter, Mockito.never()).add(endElement); + verify(wrappedWriter).flush(); + verify(ioWriter).write(""); + verify(ioWriter).flush(); + + StartElement startElement = eventFactory.createStartElement(unopenedB, null, null); + writer.add(startElement); + endElement = eventFactory.createEndElement(unopenedB, null); + writer.add(endElement); + + verify(wrappedWriter).add(startElement); + verify(wrappedWriter).add(endElement); + + // only internal list should be modified + Assert.assertEquals(2, unopenedElements.size()); + } + + @Test + public void testOtherEndElement() throws Exception { + EndElement endElement = eventFactory.createEndElement(other, null); + writer.add(endElement); + + verify(wrappedWriter).add(endElement); + } + + @Test + public void testOtherEvent() throws Exception { + XMLEvent event = eventFactory.createCharacters("foo"); + writer.add(event); + + verify(wrappedWriter).add(event); + } + + @Test (expected = DataAccessResourceFailureException.class) + public void testIOException() throws Exception { + EndElement endElementB = eventFactory.createEndElement(unopenedB, null); + Mockito.doThrow(new IOException("Simulated IOException")).when(ioWriter).write(""); + writer.add(endElementB); + } + +}