From 4df44a88fa9d8bb8d2935ec7366ca8968b0b7e40 Mon Sep 17 00:00:00 2001 From: robokaso Date: Thu, 6 Mar 2008 12:22:48 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-382: FlatFileItemWriter and StaxEventItemWriter should buffer output http://jira.springframework.org/browse/BATCH-382 cleaned up stax writer and tests --- .../batch/io/xml/StaxEventItemWriter.java | 43 ++++------------ .../io/xml/StaxEventItemWriterTests.java | 50 +++++++++++++------ 2 files changed, 43 insertions(+), 50 deletions(-) diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index e51c09d7d..57de5b302 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -23,7 +23,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.exception.ClearFailedException; import org.springframework.batch.item.exception.FlushFailedException; -import org.springframework.batch.item.exception.StreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessResourceFailureException; @@ -37,7 +36,7 @@ import org.springframework.util.CollectionUtils; * This item writer also provides restart, statistics and transaction features * by implementing corresponding interfaces. * - * Output is buffered until {@link #flush()} is called - only then the actual + * Output is buffered until {@link #flush()} is wasCalled - only then the actual * writing to file takes place. * * @author Peter Zozom @@ -80,9 +79,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements // root element attributes private Map rootElementAttributes = null; - // signalizes that output source has been initialized - private boolean initialized = false; - // signalizes that marshalling was restarted private boolean restarted = false; @@ -220,10 +216,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements this.overwriteOutput = overwriteOutput; } - protected FileChannel getChannel() { - return channel; - } - /** * @throws Exception * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() @@ -251,7 +243,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements open(startAtPosition); } - /* + /** * Helper method for opening output source at given file position */ private void open(long position) { @@ -283,8 +275,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", xse); } - initialized = true; - } /** @@ -299,7 +289,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements * @param writer XML event writer * @throws XMLStreamException */ - protected void startDocument(XMLEventWriter writer) throws XMLStreamException { + private void startDocument(XMLEventWriter writer) throws XMLStreamException { XMLEventFactory factory = XMLEventFactory.newInstance(); @@ -328,14 +318,14 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements * @param writer XML event writer * @throws XMLStreamException */ - protected void endDocument(XMLEventWriter writer) throws XMLStreamException { + private void endDocument(XMLEventWriter writer) throws XMLStreamException { // writer.writeEndDocument(); <- this doesn't work after restart // we need to write end tag of the root element manually writer.flush(); ByteBuffer bbuf = ByteBuffer.wrap(("").getBytes()); try { - getChannel().write(bbuf); + channel.write(bbuf); } catch (IOException ioe) { throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", ioe); @@ -348,7 +338,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements * @see org.springframework.batch.item.ResourceLifecycle#close(ExecutionContext) */ public void close(ExecutionContext executionContext) { - initialized = false; flush(); try { endDocument(delegateEventWriter); @@ -364,15 +353,13 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements } /** - * Write the value object to XML stream. + * Write the value object to internal buffer. * * @param item the value object - * @see org.springframework.batch.item.ItemWriter#write(java.lang.Object) + * @see #flush() */ public void write(Object item) { - - Assert.state(initialized, "ItemStream must be open before it can be used."); - + currentRecordCount++; buffer.add(item); } @@ -382,9 +369,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements * @see org.springframework.batch.item.ItemStream#update(ExecutionContext) */ public void update(ExecutionContext executionContext) { - if (!initialized) { - throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context."); - } if (saveState) { Assert.notNull(executionContext, "ExecutionContext must not be null"); @@ -393,15 +377,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements } } - /** - * Restore processing from provided restart data. - * @param data the restart data - * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionContext) - */ - public void restoreFrom(ExecutionContext data) { - - } - /* * Get the actual position in file channel. This method flushes any buffered * data before position is read. @@ -443,7 +418,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implements } /** - * Writes buffered items to file and marks restore point. + * Writes buffered items to XML stream and marks restore point. */ public void flush() throws FlushFailedException { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java index cba726a72..9740548f7 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java @@ -33,8 +33,8 @@ public class StaxEventItemWriterTests extends TestCase { private ExecutionContext executionContext; - // test record for writing to output - private Object record = new Object() { + // test item for writing to output + private Object item = new Object() { public String toString() { return TEST_STRING; } @@ -51,37 +51,51 @@ public class StaxEventItemWriterTests extends TestCase { } /** - * Write should pass its argument and StaxResult object to Serializer + * Flush should pass buffered items to Serializer. */ - public void testWrite() throws Exception { + public void testFlush() throws Exception { writer.open(executionContext); - Marshaller marshaller = new InputCheckMarshaller(); + InputCheckMarshaller marshaller = new InputCheckMarshaller(); MarshallingEventWriterSerializer serializer = new MarshallingEventWriterSerializer(marshaller); writer.setSerializer(serializer); // see asserts in the marshaller - writer.write(record); + writer.write(item); + assertFalse(marshaller.wasCalled); + + writer.flush(); + assertTrue(marshaller.wasCalled); } + + public void testClear() throws Exception { + writer.open(executionContext); + writer.write(item); + writer.write(item); + writer.clear(); + //writer.write(item); + writer.flush(); + assertFalse(outputFileContent().contains(TEST_STRING)); + } /** * Rolled back records should not be written to output file. */ public void testRollback() throws Exception { writer.open(executionContext); - writer.write(record); + writer.write(item); // rollback writer.clear(); assertEquals("", outputFileContent()); } /** - * Commited output is written to the output file. + * Item is written to the output file only after flush. */ - public void testCommit() throws Exception { + public void testWriteAndFlush() throws Exception { writer.open(executionContext); - writer.write(record); - // commit + writer.write(item); + assertEquals("", outputFileContent()); writer.flush(); assertTrue(outputFileContent().contains(TEST_STRING)); } @@ -91,8 +105,8 @@ public class StaxEventItemWriterTests extends TestCase { */ public void testRestart() throws Exception { writer.open(executionContext); - // write record - writer.write(record); + // write item + writer.write(item); writer.flush(); writer.update(executionContext); writer.close(executionContext); @@ -100,7 +114,7 @@ public class StaxEventItemWriterTests extends TestCase { // create new writer from saved restart data and continue writing writer = createItemWriter(); writer.open(executionContext); - writer.write(record); + writer.write(item); writer.close(executionContext); // check the output is concatenation of 'before restart' and 'after @@ -123,7 +137,7 @@ public class StaxEventItemWriterTests extends TestCase { writer.open(executionContext); final int NUMBER_OF_RECORDS = 10; for (int i = 1; i <= NUMBER_OF_RECORDS; i++) { - writer.write(record); + writer.write(item); writer.update(executionContext); long writeStatistics = executionContext.getLong(StaxEventItemWriter.class.getSimpleName() + ".record.count"); @@ -154,9 +168,13 @@ public class StaxEventItemWriterTests extends TestCase { * Checks the received parameters. */ private class InputCheckMarshaller implements Marshaller { + + boolean wasCalled = false; + public void marshal(Object graph, Result result) { + wasCalled = true; assertTrue(result instanceof StaxResult); - assertSame(record, graph); + assertSame(item, graph); } public boolean supports(Class clazz) {