diff --git a/spring-batch-core/.settings/org.eclipse.jdt.core.prefs b/spring-batch-core/.settings/org.eclipse.jdt.core.prefs index ad9685860..e7a358223 100644 --- a/spring-batch-core/.settings/org.eclipse.jdt.core.prefs +++ b/spring-batch-core/.settings/org.eclipse.jdt.core.prefs @@ -1,4 +1,4 @@ -#Wed Jul 16 09:54:03 CEST 2008 +#Tue Aug 19 18:12:49 BST 2008 eclipse.preferences.version=1 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 @@ -13,7 +13,7 @@ org.eclipse.jdt.core.compiler.problem.assertIdentifier=error org.eclipse.jdt.core.compiler.problem.autoboxing=ignore org.eclipse.jdt.core.compiler.problem.deprecation=warning org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled -org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled +org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=enabled org.eclipse.jdt.core.compiler.problem.discouragedReference=warning org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore org.eclipse.jdt.core.compiler.problem.enumIdentifier=error @@ -66,6 +66,7 @@ org.eclipse.jdt.core.compiler.problem.unnecessaryElse=ignore org.eclipse.jdt.core.compiler.problem.unnecessaryTypeCheck=ignore org.eclipse.jdt.core.compiler.problem.unqualifiedFieldAccess=ignore org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownException=ignore +org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionExemptExceptionAndThrowable=enabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionIncludeDocCommentReference=enabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionIncludeUncheckedExceptions=disabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionWhenOverriding=disabled diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java index e665f01cc..d454fbbbf 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java @@ -144,18 +144,14 @@ public class ItemOrientedStepHandler implements StepHandler { /** * @throws ClearFailedException - * @see org.springframework.batch.item.ItemWriter#clear() */ public void clear() throws ClearFailedException { - itemWriter.clear(); } /** * @throws FlushFailedException - * @see org.springframework.batch.item.ItemWriter#flush() */ public void flush() throws FlushFailedException { - itemWriter.flush(); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java index b53ea7438..0e0f86077 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java @@ -63,15 +63,13 @@ public interface StepHandler { /** * Implementations should delegate to an {@link ItemWriter}. - * - * @see org.springframework.batch.item.ItemWriter#flush() + * @deprecated */ public void flush() throws FlushFailedException; /** * Implementations should delegate to an {@link ItemWriter}. - * - * @see org.springframework.batch.item.ItemWriter#clear() + * @deprecated */ public void clear() throws ClearFailedException; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index 99e2b937b..0842a28b0 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -442,9 +442,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { protected final Log logger = LogFactory.getLog(getClass()); - private List written = new ArrayList(); - - private int flushIndex = -1; + private List written = TransactionAwareProxyFactory.createTransactionalList(); private final Collection failures; @@ -461,13 +459,9 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { } public void clear() throws ClearFailedException { - for (int i = flushIndex + 1; i < written.size(); i++) { - written.remove(written.size() - 1); - } } public void flush() throws FlushFailedException { - flushIndex = written.size() - 1; } public void write(List items) throws Exception { diff --git a/spring-batch-infrastructure/.settings/org.eclipse.jdt.core.prefs b/spring-batch-infrastructure/.settings/org.eclipse.jdt.core.prefs index 65b674997..fba517e3a 100644 --- a/spring-batch-infrastructure/.settings/org.eclipse.jdt.core.prefs +++ b/spring-batch-infrastructure/.settings/org.eclipse.jdt.core.prefs @@ -1,4 +1,4 @@ -#Tue Jul 15 10:10:09 CEST 2008 +#Tue Aug 19 17:57:24 BST 2008 eclipse.preferences.version=1 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 @@ -13,7 +13,7 @@ org.eclipse.jdt.core.compiler.problem.assertIdentifier=error org.eclipse.jdt.core.compiler.problem.autoboxing=ignore org.eclipse.jdt.core.compiler.problem.deprecation=warning org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled -org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled +org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=enabled org.eclipse.jdt.core.compiler.problem.discouragedReference=warning org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore org.eclipse.jdt.core.compiler.problem.enumIdentifier=error @@ -66,6 +66,7 @@ org.eclipse.jdt.core.compiler.problem.unnecessaryElse=ignore org.eclipse.jdt.core.compiler.problem.unnecessaryTypeCheck=ignore org.eclipse.jdt.core.compiler.problem.unqualifiedFieldAccess=ignore org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownException=ignore +org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionExemptExceptionAndThrowable=enabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionIncludeDocCommentReference=enabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionIncludeUncheckedExceptions=disabled org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionWhenOverriding=disabled diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ClearFailedException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ClearFailedException.java index b553a25c1..0c96016f8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ClearFailedException.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ClearFailedException.java @@ -16,7 +16,8 @@ package org.springframework.batch.item; /** - * Unchecked exception indicating that an error has occurred while trying to call {@link ItemWriter#clear()} + * Unchecked exception indicating that an error has occurred while trying to + * clear a buffer on a rollback. * * @author Lucas Ward * @author Ben Hale @@ -24,7 +25,8 @@ package org.springframework.batch.item; public class ClearFailedException extends ItemWriterException { /** - * Create a new {@link ClearFailedException} based on a message and another exception. + * Create a new {@link ClearFailedException} based on a message and another + * exception. * * @param message the message for this exception * @param cause the other exception diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FlushFailedException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FlushFailedException.java index d7a7f738a..f9d0d7b77 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FlushFailedException.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FlushFailedException.java @@ -16,7 +16,7 @@ package org.springframework.batch.item; /** - * Unchecked exception indicating that an error has occurred while trying to call {@link ItemWriter#flush()} + * Unchecked exception indicating that an error has occurred while trying to flush a buffer. * * @author Lucas Ward * @author Ben Hale diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java index fd330c9c6..0c4570249 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java @@ -56,6 +56,7 @@ public interface ItemWriter { * @throws FlushFailedException in case of an error. If this exception is * thrown the writer may be in an inconsistent state and manual intervention * might be required to reconcile the data with persistent output. + * @deprecated */ void flush() throws FlushFailedException; @@ -65,6 +66,7 @@ public interface ItemWriter { * @throws ClearFailedException in case of an error. If this exception is * thrown the writer may be in an inconsistent state and manual intervention * might be required to reconcile the data with persistent output. + * @deprecated */ void clear() throws ClearFailedException; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 3245333cc..c81bcd65f 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -86,8 +86,6 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private String encoding = OutputState.DEFAULT_CHARSET; - private List lineBuffer = new ArrayList(); - private List headerLines = new ArrayList(); private String lineSeparator = DEFAULT_LINE_SEPARATOR; @@ -185,16 +183,24 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement */ public void write(List items) throws Exception { - for (T item : items) { - - if (getOutputState().isInitialized()) { - lineBuffer.add(lineAggregator.aggregate(item) + lineSeparator); - } - else { - throw new WriterNotOpenException("Writer must be open before it can be written to"); - } - + if (!getOutputState().isInitialized()) { + throw new WriterNotOpenException("Writer must be open before it can be written to"); } + + OutputState state = getOutputState(); + + for (T item : items) { + String line = lineAggregator.aggregate(item) + lineSeparator; + try { + state.write(line); + } catch (IOException e) { + throw new FlushFailedException( + "Could not write data. The file may be corrupt.", e); + } + } + + state.mark(); + } /** @@ -222,7 +228,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement } } - private void doOpen(ExecutionContext executionContext) { + private void doOpen(ExecutionContext executionContext) throws ItemStreamException { OutputState outputState = getOutputState(); if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) { outputState.restoreFrom(executionContext); @@ -234,8 +240,14 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement throw new ItemStreamException("Failed to initialize writer", ioe); } if (outputState.lastMarkedByteOffsetPosition == 0) { - for (String line : headerLines) { - lineBuffer.add(line + lineSeparator); + try { + for (String line : headerLines) { + outputState.write(line + lineSeparator); + } + } + catch (IOException e) { + throw new FlushFailedException( + "Could not write headers. The file may be corrupt.", e); } } } @@ -264,17 +276,6 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement } public void flush() throws FlushFailedException { - OutputState state = getOutputState(); - for (String line : lineBuffer) { - try { - state.write(line); - } - catch (IOException e) { - throw new FlushFailedException("Failed to write line to output file: " + line, e); - } - } - lineBuffer.clear(); - state.mark(); } // Returns object representing state. @@ -501,7 +502,6 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement } public void clear() throws ClearFailedException { - lineBuffer.clear(); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemWriter.java index 5f1d9b6e4..1e508b2fd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemWriter.java @@ -20,8 +20,7 @@ import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemWriter; /** - * Abstract {@link ItemWriter} that allows for base classes to only implement - * the {@link #flush()} and {@link #clear()} methods if they need it. + * Abstract {@link ItemWriter}. * * @author Lucas Ward */ diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java index 6d75ae691..c18364dba 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java @@ -3,8 +3,6 @@ package org.springframework.batch.item.support; import java.util.Arrays; import java.util.List; -import org.springframework.batch.item.ClearFailedException; -import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemWriter; /** @@ -15,7 +13,7 @@ import org.springframework.batch.item.ItemWriter; * @author Robert Kasanicky * @author Dave Syer */ -public class CompositeItemWriter implements ItemWriter { +public class CompositeItemWriter extends AbstractItemWriter { private List> delegates; @@ -32,16 +30,4 @@ public class CompositeItemWriter implements ItemWriter { } } - public void clear() throws ClearFailedException { - for (ItemWriter writer : delegates) { - writer.clear(); - } - } - - public void flush() throws FlushFailedException { - for (ItemWriter writer : delegates) { - writer.flush(); - } - } - } 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 a0ddfc537..af8cc4ad1 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 @@ -109,18 +109,11 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen // 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; private boolean saveState = true; - // holds the list of items for writing before they are actually written on - // #flush() - private List buffer = new ArrayList(); - private List headers = new ArrayList(); public StaxEventItemWriter() { @@ -345,6 +338,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen } } + + writer.flush(); } @@ -384,8 +379,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen log.error(e); } - flush(); try { + delegateEventWriter.flush(); endDocument(delegateEventWriter); eventWriter.close(); channel.close(); @@ -399,15 +394,25 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen } /** - * Write the value object to internal buffer. + * Write the value objects and flush them to the file. * - * @param item the value object - * @see #flush() + * @param items the value object */ - public void write(List item) { + public void write(List items) { - currentRecordCount+=item.size(); - buffer.addAll(item); + currentRecordCount+=items.size(); + + for (T item : items) { + serializer.serializeObject(eventWriter, item); + } + try { + eventWriter.flush(); + } + catch (XMLStreamException e) { + throw new FlushFailedException("Failed to flush the events", e); + } + + lastCommitPointPosition = getPosition(); } /** @@ -468,28 +473,12 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen * Writes buffered items to XML stream and marks restore point. */ public void flush() throws FlushFailedException { - - for (T item : buffer) { - serializer.serializeObject(eventWriter, item); - } - try { - eventWriter.flush(); - } - catch (XMLStreamException e) { - throw new FlushFailedException("Failed to flush the events", e); - } - buffer.clear(); - - lastCommitPointPosition = getPosition(); - lastCommitPointRecordCount = currentRecordCount; } /** * Clear the output buffer */ public void clear() throws ClearFailedException { - currentRecordCount = lastCommitPointRecordCount; - buffer.clear(); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java index 00b23e7fc..fd75a3805 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java @@ -198,49 +198,13 @@ public class FlatFileItemWriterTests extends TestCase { assertEquals("1|2|", lineFromFile); } - public void testRollback() throws Exception { - writer.open(executionContext); - writer.write(Collections.singletonList("testLine1")); - // rollback - rollback(); - writer.flush(); - writer.close(null); - String lineFromFile = readLine(); - assertEquals(null, lineFromFile); - } - - public void testCommit() throws Exception { - writer.open(executionContext); - writer.write(Collections.singletonList("testLine1")); - // rollback - commit(); - writer.close(null); - String lineFromFile = readLine(); - assertEquals("testLine1", lineFromFile); - } - public void testRestart() throws Exception { writer.open(executionContext); - // write some lines writer.write(Arrays.asList(new String[] { "testLine1", "testLine2", "testLine3" })); - - // commit - commit(); - - // this will be rolled back... - writer.write(Collections.singletonList("this will be rolled back")); - - // rollback - rollback(); - // write more lines writer.write(Arrays.asList(new String[] {"testLine4", "testLine5"})); - - // commit - commit(); - // get restart data writer.update(executionContext); // close template @@ -248,19 +212,15 @@ public class FlatFileItemWriterTests extends TestCase { // init with correct data writer.open(executionContext); - // write more lines writer.write(Arrays.asList(new String[] {"testLine6","testLine7","testLine8"})); - - commit(); - // get statistics writer.update(executionContext); // close template writer.close(executionContext); // verify what was written to the file - for (int i = 1; i < 9; i++) { + for (int i = 1; i <= 8; i++) { assertEquals("testLine" + i, readLine()); } 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 09ed87032..50a50ba5e 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 @@ -31,6 +31,7 @@ import org.springframework.oxm.Marshaller; import org.springframework.oxm.XmlMappingException; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; +import org.springframework.util.StringUtils; import org.springframework.xml.transform.StaxResult; /** @@ -58,56 +59,13 @@ public class StaxEventItemWriterTests { private static final String TEST_STRING = ""; - private static final int NOT_FOUND = -1; - @Before public void setUp() throws Exception { - resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", "xml")); + resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", ".xml")); writer = createItemWriter(); executionContext = new ExecutionContext(); } - /** - * Flush should pass buffered items to Serializer. - */ - @Test - public void testFlush() throws Exception { - writer.open(executionContext); - InputCheckMarshaller marshaller = new InputCheckMarshaller(); - MarshallingEventWriterSerializer serializer = new MarshallingEventWriterSerializer(marshaller); - writer.setSerializer(serializer); - - // see asserts in the marshaller - writer.write(items); - assertFalse(marshaller.wasCalled); - - writer.flush(); - assertTrue(marshaller.wasCalled); - - } - - @Test - public void testClear() throws Exception { - writer.open(executionContext); - writer.write(Arrays.asList(new Object[] {item, item})); - writer.clear(); - // writer.write(item); - writer.flush(); - assertFalse(contains(outputFileContent(), TEST_STRING)); - } - - /** - * Rolled back records should not be written to output file. - */ - @Test - public void testRollback() throws Exception { - writer.open(executionContext); - writer.write(items); - // rollback - writer.clear(); - assertFalse(outputFileContent().contains(TEST_STRING)); - } - /** * Item is written to the output file only after flush. */ @@ -115,11 +73,9 @@ public class StaxEventItemWriterTests { public void testWriteAndFlush() throws Exception { writer.open(executionContext); writer.write(items); + writer.close(executionContext); String content = outputFileContent(); - assertFalse(content.contains(TEST_STRING)); - writer.flush(); - content = outputFileContent(); - assertTrue("Wrong content: "+content, contains(content, TEST_STRING)); + assertTrue("Wrong content: "+content, content.contains(TEST_STRING)); } /** @@ -130,7 +86,6 @@ public class StaxEventItemWriterTests { writer.open(executionContext); // write item writer.write(items); - writer.flush(); writer.update(executionContext); writer.close(executionContext); @@ -143,14 +98,8 @@ public class StaxEventItemWriterTests { // 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); + + assertEquals(2, StringUtils.countOccurrencesOf(outputFile, TEST_STRING)); } /** @@ -165,9 +114,9 @@ public class StaxEventItemWriterTests { writer.write(items); writer.flush(); String content = outputFileContent(); - assertTrue("Wrong content: "+content, contains(content, "")); - assertTrue("Wrong content: "+content, contains(content, "")); - assertTrue("Wrong content: "+content, contains(content, TEST_STRING)); + assertTrue("Wrong content: "+content, content.contains((""))); + assertTrue("Wrong content: "+content, content.contains((""))); + assertTrue("Wrong content: "+content, content.contains(TEST_STRING)); } /** @@ -179,13 +128,12 @@ public class StaxEventItemWriterTests { writer.setHeaderItems(new Object[] {header}); writer.open(executionContext); writer.write(items); - writer.clear(); writer.open(executionContext); writer.write(items); - writer.flush(); + writer.close(executionContext); String content = outputFileContent(); - assertEquals("Wrong content: "+content, 1, countContains(content, "")); - assertEquals("Wrong content: "+content, 1, countContains(content, TEST_STRING)); + assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, (""))); + assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, TEST_STRING)); } /** @@ -197,16 +145,15 @@ public class StaxEventItemWriterTests { writer.setHeaderItems(new Object[] {header}); writer.open(executionContext); writer.write(items); - writer.flush(); writer.update(executionContext); writer.close(executionContext); writer.open(executionContext); writer.write(items); - writer.clear(); - writer.flush(); + writer.close(executionContext); String content = outputFileContent(); - assertEquals("Wrong content: "+content, 1, countContains(content, "")); - assertEquals("Wrong content: "+content, 1, countContains(content, TEST_STRING)); + assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, (""))); + // THis test is not transactional, so the body gets written twice, but at least there's only one header + assertEquals("Wrong content: "+content, 2, StringUtils.countOccurrencesOf(content, TEST_STRING)); } /** @@ -240,13 +187,10 @@ public class StaxEventItemWriterTests { } }); writer.open(executionContext); - writer.flush(); - - assertTrue(outputFileContent().indexOf("") != NOT_FOUND); - assertTrue(outputFileContent().endsWith("")); + String content = outputFileContent(); + assertTrue(content.contains("")); + assertTrue(content.endsWith("")); } @Test @@ -267,25 +211,6 @@ public class StaxEventItemWriterTests { } } - /** - * 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(item, graph); - } - - @SuppressWarnings("unchecked") - public boolean supports(Class clazz) { - return true; - } - } - /** * Writes object's toString representation as XML comment. */ @@ -336,18 +261,4 @@ public class StaxEventItemWriterTests { return source; } - - private boolean contains(String str, String searchStr) { - return str.indexOf(searchStr) != -1; - } - - private int countContains(String str, String searchStr) { - int begin = -1; - int count = 0; - while (str.indexOf(searchStr, begin+1) > begin) { - count++; - begin = str.indexOf(searchStr, begin); - } - return count; - } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index 540affa70..ccfd91b29 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java @@ -1,6 +1,5 @@ package org.springframework.batch.integration.chunk; -import java.util.ArrayList; import java.util.List; import org.apache.commons.logging.Log; @@ -15,23 +14,16 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatContext; import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageTarget; -import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream { private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class); - /** - * Key for items processed in the current transaction {@link RepeatContext}. - */ - private static final String ITEMS_PROCESSED = ChunkMessageChannelItemWriter.class.getName() + ".ITEMS_PROCESSED"; - static final String ACTUAL = "ACTUAL"; static final String EXPECTED = "EXPECTED"; @@ -65,38 +57,16 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } public void write(List items) throws Exception { - bindTransactionResources(); - for (T item : items) { - getProcessed().add(item); - logger.debug("Added item to chunk: " + item); - } - } - - /** - * Flush the buffer, sending the items as a chunk message to be processed by - * a {@link ChunkHandler}. To avoid overwhelming the receivers, this method - * will block until the number of chunks pending is less than the throttle - * limit. - * - * @see org.springframework.batch.item.ItemWriter#flush() - */ - public void flush() throws FlushFailedException { - - bindTransactionResources(); // in case we are called outside a - // transaction - // Block until expecting <= throttle limit - can Spring // Integration do that for me? while (localState.getExpecting() > throttleLimit) { getNextResult(100); } - List processed = getProcessed(); + if (!items.isEmpty()) { - if (!processed.isEmpty()) { - - logger.debug("Dispatching chunk: " + processed); - ChunkRequest request = new ChunkRequest(processed, localState.getJobId(), localState.getSkipCount()); + logger.debug("Dispatching chunk: " + items); + ChunkRequest request = new ChunkRequest(items, localState.getJobId(), localState.getSkipCount()); GenericMessage> message = new GenericMessage>(request); target.send(message); localState.expected++; @@ -106,8 +76,20 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo // Short little timeout to look for an immediate reply. getNextResult(1); - unbindTransactionResources(); + } + /** + * No-op. + * @see org.springframework.batch.item.ItemWriter#flush() + */ + public void flush() throws FlushFailedException { + } + + /** + * No-op. + * @see org.springframework.batch.item.ItemWriter#clear() + */ + public void clear() throws ClearFailedException { } @Override @@ -179,7 +161,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * otherwise return null. */ private void getNextResult(long timeout) { - // TODO: use the timeout Message message = source.receive(timeout); if (message != null) { ChunkResponse payload = message.getPayload(); @@ -196,50 +177,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } } - /** - * Accessor for the list of processed items in this transaction. - * - * @return the processed - */ - private List getProcessed() { - Assert.state(TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED), - "Processed items not bound to transaction."); - @SuppressWarnings("unchecked") - List processed = (List) TransactionSynchronizationManager.getResource(ITEMS_PROCESSED); - return processed; - } - - /** - * Set up the {@link RepeatContext} as a transaction resource. - * - * @param context the context to set - */ - private void bindTransactionResources() { - if (TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) { - return; - } - TransactionSynchronizationManager.bindResource(ITEMS_PROCESSED, new ArrayList()); - } - - /** - * Remove the transaction resource associated with this context. - */ - private void unbindTransactionResources() { - if (!TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) { - return; - } - TransactionSynchronizationManager.unbindResource(ITEMS_PROCESSED); - } - - /** - * Clear the buffer. - * - * @see org.springframework.batch.item.ItemWriter#clear() - */ - public void clear() throws ClearFailedException { - unbindTransactionResources(); - } - private static class LocalState { private long actual; diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/CustomItemWriterTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/CustomItemWriterTests.java index b2d29ab01..7ee51cd74 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/CustomItemWriterTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/CustomItemWriterTests.java @@ -15,16 +15,15 @@ */ package org.springframework.batch.sample.common; -import java.util.ArrayList; +import static org.junit.Assert.assertEquals; + import java.util.Arrays; import java.util.Collections; import java.util.List; -import junit.framework.TestCase; - -import org.springframework.batch.item.ClearFailedException; -import org.springframework.batch.item.FlushFailedException; -import org.springframework.batch.item.ItemWriter; +import org.junit.Test; +import org.springframework.batch.item.support.AbstractItemWriter; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; /** * Unit test class that was used as part of the Reference Documentation. I'm @@ -34,45 +33,24 @@ import org.springframework.batch.item.ItemWriter; * @author Lucas Ward * */ -public class CustomItemWriterTests extends TestCase { - - /* - * (non-Javadoc) - * - * @see junit.framework.TestCase#setUp() - */ - protected void setUp() throws Exception { - super.setUp(); - } +public class CustomItemWriterTests { + @Test public void testFlush() throws Exception { CustomItemWriter itemWriter = new CustomItemWriter(); itemWriter.write(Collections.singletonList("1")); - assertEquals(0, itemWriter.getOutput().size()); - itemWriter.flush(); assertEquals(1, itemWriter.getOutput().size()); itemWriter.write(Arrays.asList(new String[] {"2","3"})); - itemWriter.clear(); - assertEquals(1, itemWriter.getOutput().size()); + assertEquals(3, itemWriter.getOutput().size()); } - public class CustomItemWriter implements ItemWriter { + public class CustomItemWriter extends AbstractItemWriter { - List output = new ArrayList(); - - List buffer = new ArrayList(); + List output = TransactionAwareProxyFactory.createTransactionalList(); public void write(List items) throws Exception { - buffer.addAll(items); - } - - public void clear() throws ClearFailedException { - buffer.clear(); - } - - public void flush() throws FlushFailedException { - output.addAll(buffer); + output.addAll(items); } public List getOutput() { diff --git a/src/.project b/src/.project index 0496c168d..2e23dfd23 100644 --- a/src/.project +++ b/src/.project @@ -1,6 +1,6 @@ - spring-batch + src