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 fe795700d..023356dda 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
@@ -24,6 +24,9 @@ import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.charset.UnsupportedCharsetException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.ExecutionContext;
@@ -45,17 +48,20 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
- * This class is an output target that writes data to a file or stream. The writer also provides restart, statistics and
- * transaction features by implementing corresponding interfaces where possible (with a file). The location of the file
- * is defined by a {@link Resource} and must represent a writable file.
+ * This class is an output target that writes data to a file or stream. The
+ * writer also provides restart, statistics and transaction features by
+ * implementing corresponding interfaces where possible (with a file). The
+ * location of the file is defined by a {@link Resource} and must represent a
+ * writable file.
*
* Uses buffered writer to improve performance.
*
* Use {@link #write(String)} method to output a line to an item writer.
*
*
- * This class will be updated in the future to use a buffering approach to handling transactions, rather than outputting
- * directly to the file and truncating on rollback
+ * This class will be updated in the future to use a buffering approach to
+ * handling transactions, rather than outputting directly to the file and
+ * truncating on rollback
*
*
* @author Waseem Malik
@@ -80,7 +86,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
private LineAggregator lineAggregator = new DelimitedLineAggregator();
private FieldSetCreator fieldSetCreator;
-
+
private boolean saveState = false;
private boolean shouldDeleteIfExists = true;
@@ -89,6 +95,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
private int bufferSize = OutputState.DEFAULT_BUFFER_SIZE;
+ private List lineBuffer = new ArrayList();
+
public FlatFileItemWriter() {
setName(ClassUtils.getShortName(FlatFileItemWriter.class));
}
@@ -106,8 +114,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
/**
- * Public setter for the {@link LineAggregator}. This will be used to translate a {@link FieldSet} into a line for
- * output.
+ * Public setter for the {@link LineAggregator}. This will be used to
+ * translate a {@link FieldSet} into a line for output.
*
* @param lineAggregator the {@link LineAggregator} to set
*/
@@ -116,8 +124,9 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
/**
- * Public setter for the {@link FieldSetCreator}. This will be used to transform the item into a {@link FieldSet}
- * before it is aggregated by the {@link LineAggregator}.
+ * Public setter for the {@link FieldSetCreator}. This will be used to
+ * transform the item into a {@link FieldSet} before it is aggregated by the
+ * {@link LineAggregator}.
*
* @param fieldSetCreator the {@link FieldSetCreator} to set
*/
@@ -135,18 +144,21 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
/**
- * Writes out a string followed by a "new line", where the format of the new line separator is determined by the
- * underlying operating system. If the input is not a String and a converter is available the converter will be
- * applied and then this method recursively called with the result. If the input is an array or collection each
- * value will be written to a separate line (recursively calling this method for each value). If no converter is
+ * Writes out a string followed by a "new line", where the format of the new
+ * line separator is determined by the underlying operating system. If the
+ * input is not a String and a converter is available the converter will be
+ * applied and then this method recursively called with the result. If the
+ * input is an array or collection each value will be written to a separate
+ * line (recursively calling this method for each value). If no converter is
* supplied the input object's toString method will be used.
*
- * @param data Object (a String or Object that can be converted) to be written to output stream
+ * @param data Object (a String or Object that can be converted) to be
+ * written to output stream
* @throws Exception if the transformer or file output fail
*/
public void write(Object data) throws Exception {
FieldSet fieldSet = fieldSetCreator.mapItem(data);
- getOutputState().write(lineAggregator.aggregate(fieldSet) + LINE_SEPARATOR);
+ lineBuffer.add(lineAggregator.aggregate(fieldSet) + LINE_SEPARATOR);
}
/**
@@ -190,6 +202,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) {
outputState.restoreFrom(executionContext);
}
+ outputState.initializeBufferedWriter();
}
/**
@@ -199,18 +212,18 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
if (state == null) {
throw new ItemStreamException("ItemStream not open or already closed.");
}
-
Assert.notNull(executionContext, "ExecutionContext must not be null");
-
- if(saveState){
-
+
+ if (saveState) {
+
try {
executionContext.putLong(getKey(RESTART_DATA_NAME), state.position());
- } catch (IOException e) {
+ }
+ catch (IOException e) {
throw new ItemStreamException("ItemStream does not return current position properly", e);
}
-
+
executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
executionContext.putLong(getKey(RESTART_COUNT_STATISTICS_NAME), state.restartCount);
}
@@ -228,7 +241,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
/**
- * Encapsulates the runtime state of the writer. All state changing operations on the writer go through this class.
+ * Encapsulates the runtime state of the writer. All state changing
+ * operations on the writer go through this class.
*/
private class OutputState {
// default encoding for writing to output files - set to UTF-8.
@@ -261,7 +275,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
boolean shouldDeleteIfExists = true;
/**
- * Return the byte offset position of the cursor in the output file as a long integer.
+ * Return the byte offset position of the cursor in the output file as a
+ * long integer.
*/
public long position() throws IOException {
long pos = 0;
@@ -318,7 +333,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
outputBufferedWriter.close();
fileChannel.close();
- } catch (IOException ioe) {
+ }
+ catch (IOException ioe) {
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
}
}
@@ -355,13 +371,15 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
public void mark() {
try {
lastMarkedByteOffsetPosition = this.position();
- } catch (IOException e) {
+ }
+ catch (IOException e) {
throw new MarkFailedException("Unable to get position for mark", e);
}
}
/**
- * Creates the buffered writer for the output file channel based on configuration information.
+ * Creates the buffered writer for the output file channel based on
+ * configuration information.
*/
private void initializeBufferedWriter() {
File file;
@@ -380,7 +398,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
if (file.exists()) {
if (shouldDeleteIfExists) {
file.delete();
- } else {
+ }
+ else {
throw new ItemStreamException("Resource already exists: " + resource);
}
}
@@ -391,14 +410,16 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
file.createNewFile();
}
- } catch (IOException ioe) {
+ }
+ catch (IOException ioe) {
throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]",
- ioe);
+ ioe);
}
try {
fileChannel = (new FileOutputStream(file.getAbsolutePath(), true)).getChannel();
- } catch (FileNotFoundException fnfe) {
+ }
+ catch (FileNotFoundException fnfe) {
throw new ItemStreamException("Bad filename property parameter " + file, fnfe);
}
@@ -414,8 +435,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
/**
- * Returns the buffered writer opened to the beginning of the file specified by the absolute path name contained
- * in absoluteFileName.
+ * Returns the buffered writer opened to the beginning of the file
+ * specified by the absolute path name contained in absoluteFileName.
*/
private BufferedWriter getBufferedWriter(FileChannel fileChannel, String encoding, int bufferSize) {
try {
@@ -425,35 +446,41 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
// If a buffer was requested, allocate.
if (bufferSize > 0) {
outputBufferedWriter = new BufferedWriter(Channels.newWriter(fileChannel, encoding), bufferSize);
- } else {
+ }
+ else {
outputBufferedWriter = new BufferedWriter(Channels.newWriter(fileChannel, encoding));
}
return outputBufferedWriter;
- } catch (UnsupportedCharsetException ucse) {
+ }
+ catch (UnsupportedCharsetException ucse) {
throw new ItemStreamException("Bad encoding configuration for output file " + fileChannel, ucse);
}
}
/**
- * Resets the file writer's current position to the point stored in the last marked byte offset position
- * variable. It first checks to make sure the current size of the file is not less than the byte position to be
- * moved to (if it is, throws an environment exception), then it truncates the file to that reset position, and
- * set the cursor to start writing at that point.
+ * Resets the file writer's current position to the point stored in the
+ * last marked byte offset position variable. It first checks to make
+ * sure the current size of the file is not less than the byte position
+ * to be moved to (if it is, throws an environment exception), then it
+ * truncates the file to that reset position, and set the cursor to
+ * start writing at that point.
*/
public void reset() throws ResetFailedException {
checkFileSize();
try {
getOutputState().truncate();
- } catch (IOException e) {
+ }
+ catch (IOException e) {
throw new ResetFailedException("Unable to truncate file", e);
}
}
/**
- * Checks (on setState) to make sure that the current output file's size is not smaller than the last saved
- * commit point. If it is, then the file has been damaged in some way and whole task must be started over again
- * from the beginning.
+ * Checks (on setState) to make sure that the current output file's size
+ * is not smaller than the last saved commit point. If it is, then the
+ * file has been damaged in some way and whole task must be started over
+ * again from the beginning.
*/
private void checkFileSize() {
long size = -1;
@@ -461,7 +488,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
try {
outputBufferedWriter.flush();
size = fileChannel.size();
- } catch (IOException e) {
+ }
+ catch (IOException e) {
throw new ResetFailedException("An Error occured while checking file size", e);
}
@@ -473,22 +501,29 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
}
public void clear() throws ClearFailedException {
- try {
- getOutputState().reset();
- } catch (Exception e) {
- throw new ClearFailedException("Could not reset the state of the writer", e);
- }
+ lineBuffer.clear();
}
public void flush() throws FlushFailedException {
- getOutputState().mark();
+ OutputState state = getOutputState();
+ for (Iterator iterator = lineBuffer.listIterator(); iterator.hasNext();) {
+ String line = (String) iterator.next();
+ try {
+ state.write(line);
+ }
+ catch (IOException e) {
+ throw new FlushFailedException("Failed to write line to output file: " + line, e);
+ }
+ }
+ lineBuffer.clear();
+ state.mark();
}
-
+
/**
- * Set the boolean indicating whether or not state should be saved
- * in the provided {@link ExecutionContext} during the {@link ItemStream}
- * call to update. Setting this to false means that it will always start
- * at the beginning.
+ * Set the boolean indicating whether or not state should be saved in the
+ * provided {@link ExecutionContext} during the {@link ItemStream} call to
+ * update. Setting this to false means that it will always start at the
+ * beginning.
*
* @param saveState
*/
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 9d5edecfa..5c39309da 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
@@ -46,7 +46,7 @@ import org.springframework.util.ClassUtils;
public class FlatFileItemWriterTests extends TestCase {
// object under test
- private FlatFileItemWriter inputSource = new FlatFileItemWriter();
+ private FlatFileItemWriter writer = new FlatFileItemWriter();
// String to be written into file by the FlatFileInputTemplate
private static final String TEST_STRING = "FlatFileOutputTemplateTest-OutputData";
@@ -72,10 +72,10 @@ public class FlatFileItemWriterTests extends TestCase {
outputFile = File.createTempFile("flatfile-output-", ".tmp");
- inputSource.setResource(new FileSystemResource(outputFile));
- inputSource.setFieldSetCreator(new PassThroughFieldSetMapper());
- inputSource.afterPropertiesSet();
- inputSource.setSaveState(true);
+ writer.setResource(new FileSystemResource(outputFile));
+ writer.setFieldSetCreator(new PassThroughFieldSetMapper());
+ writer.afterPropertiesSet();
+ writer.setSaveState(true);
executionContext = new ExecutionContext();
}
@@ -86,7 +86,7 @@ public class FlatFileItemWriterTests extends TestCase {
if (reader != null) {
reader.close();
}
- inputSource.close(null);
+ writer.close(null);
outputFile.delete();
}
@@ -110,9 +110,10 @@ public class FlatFileItemWriterTests extends TestCase {
* @throws Exception
*/
public void testWriteString() throws Exception {
- inputSource.open(executionContext);
- inputSource.write(TEST_STRING);
- inputSource.close(null);
+ writer.open(executionContext);
+ writer.write(TEST_STRING);
+ writer.flush();
+ writer.close(null);
String lineFromFile = readLine();
assertEquals(TEST_STRING, lineFromFile);
@@ -124,14 +125,14 @@ public class FlatFileItemWriterTests extends TestCase {
* @throws Exception
*/
public void testWriteWithConverter() throws Exception {
- inputSource.setFieldSetCreator(new FieldSetCreator() {
+ writer.setFieldSetCreator(new FieldSetCreator() {
public FieldSet mapItem(Object data) {
return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
Object data = new Object();
- inputSource.write(data);
- inputSource.close(null);
+ writer.write(data);
+ writer.flush();
String lineFromFile = readLine();
// converter not used if input is String
assertEquals("FOO:" + data.toString(), lineFromFile);
@@ -143,14 +144,14 @@ public class FlatFileItemWriterTests extends TestCase {
* @throws Exception
*/
public void testWriteWithConverterAndInfiniteLoop() throws Exception {
- inputSource.setFieldSetCreator(new FieldSetCreator() {
+ writer.setFieldSetCreator(new FieldSetCreator() {
public FieldSet mapItem(Object data) {
return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
Object data = new Object();
- inputSource.write(data);
- inputSource.close(null);
+ writer.write(data);
+ writer.flush();
String lineFromFile = readLine();
// converter not used if input is String
assertEquals("FOO:" + data.toString(), lineFromFile);
@@ -162,13 +163,13 @@ public class FlatFileItemWriterTests extends TestCase {
* @throws Exception
*/
public void testWriteWithConverterAndString() throws Exception {
- inputSource.setFieldSetCreator(new FieldSetCreator() {
+ writer.setFieldSetCreator(new FieldSetCreator() {
public FieldSet mapItem(Object data) {
return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
- inputSource.write(TEST_STRING);
- inputSource.close(null);
+ writer.write(TEST_STRING);
+ writer.flush();
String lineFromFile = readLine();
assertEquals("FOO:" + TEST_STRING, lineFromFile);
}
@@ -182,71 +183,74 @@ public class FlatFileItemWriterTests extends TestCase {
String args = "1";
// AggregatorStub ignores the LineDescriptor, so we pass null
- inputSource.write(args);
- inputSource.close(null);
+ writer.write(args);
+ writer.flush();
String lineFromFile = readLine();
assertEquals(args, lineFromFile);
}
public void testRollback() throws Exception {
- inputSource.write("testLine1");
+ writer.write("testLine1");
// rollback
rollback();
- inputSource.close(null);
+ writer.flush();
+ writer.close(null);
String lineFromFile = readLine();
assertEquals(null, lineFromFile);
}
public void testCommit() throws Exception {
- inputSource.write("testLine1");
+ writer.write("testLine1");
// rollback
commit();
- inputSource.close(null);
+ writer.close(null);
String lineFromFile = readLine();
assertEquals("testLine1", lineFromFile);
}
public void testRestart() throws Exception {
-
- inputSource.open(executionContext);
+
+ writer.open(executionContext);
// write some lines
- inputSource.write("testLine1");
- inputSource.write("testLine2");
- inputSource.write("testLine3");
+ writer.write("testLine1");
+ writer.write("testLine2");
+ writer.write("testLine3");
// commit
commit();
// this will be rolled back...
- inputSource.write("this will be rolled back");
+ writer.write("this will be rolled back");
// rollback
rollback();
// write more lines
- inputSource.write("testLine4");
- inputSource.write("testLine5");
+ writer.write("testLine4");
+ writer.write("testLine5");
// commit
commit();
// get restart data
- inputSource.update(executionContext);
+ writer.update(executionContext);
// close template
- inputSource.close(executionContext);
+ writer.close(executionContext);
// init with correct data
- inputSource.open(executionContext);
+ writer.open(executionContext);
// write more lines
- inputSource.write("testLine6");
- inputSource.write("testLine7");
- inputSource.write("testLine8");
+ writer.write("testLine6");
+ writer.write("testLine7");
+ writer.write("testLine8");
+
+ commit();
// get statistics
- inputSource.update(executionContext);
+ writer.update(executionContext);
// close template
- inputSource.close(executionContext);
+ writer.close(executionContext);
// verify what was written to the file
for (int i = 1; i < 9; i++) {
@@ -259,9 +263,9 @@ public class FlatFileItemWriterTests extends TestCase {
}
public void testAfterPropertiesSetChecksMandatory() throws Exception {
- inputSource = new FlatFileItemWriter();
+ writer = new FlatFileItemWriter();
try {
- inputSource.afterPropertiesSet();
+ writer.afterPropertiesSet();
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {
@@ -270,13 +274,13 @@ public class FlatFileItemWriterTests extends TestCase {
}
public void testDefaultStreamContext() throws Exception {
- inputSource = new FlatFileItemWriter();
- inputSource.setResource(new FileSystemResource(outputFile));
- inputSource.setFieldSetCreator(new PassThroughFieldSetMapper());
- inputSource.afterPropertiesSet();
- inputSource.setSaveState(true);
- inputSource.open(executionContext);
- inputSource.update(executionContext);
+ writer = new FlatFileItemWriter();
+ writer.setResource(new FileSystemResource(outputFile));
+ writer.setFieldSetCreator(new PassThroughFieldSetMapper());
+ writer.afterPropertiesSet();
+ writer.setSaveState(true);
+ writer.open(executionContext);
+ writer.update(executionContext);
assertNotNull(executionContext);
assertEquals(3, executionContext.entrySet().size());
assertEquals(0, executionContext.getLong(ClassUtils.getShortName(FlatFileItemWriter.class) + ".current.count"));
@@ -288,16 +292,15 @@ public class FlatFileItemWriterTests extends TestCase {
* @throws Exception
*/
public void testWriteStringWithBogusEncoding() throws Exception {
- inputSource.setEncoding("BOGUS");
- inputSource.open(executionContext);
+ writer.setEncoding("BOGUS");
try {
- inputSource.write(TEST_STRING);
+ writer.open(executionContext);
fail("Expecyted ItemStreamException");
}
catch (ItemStreamException e) {
assertTrue(e.getCause() instanceof UnsupportedCharsetException);
}
- inputSource.close(null);
+ writer.close(null);
}
/**
@@ -307,21 +310,21 @@ public class FlatFileItemWriterTests extends TestCase {
*/
public void testWriteStringWithEncodingAfterClose() throws Exception {
testWriteStringWithBogusEncoding();
- inputSource.setEncoding("UTF-8");
- inputSource.open(executionContext);
- inputSource.write(TEST_STRING);
- inputSource.close(null);
+ writer.setEncoding("UTF-8");
+ writer.open(executionContext);
+ writer.write(TEST_STRING);
+ writer.flush();
String lineFromFile = readLine();
assertEquals(TEST_STRING, lineFromFile);
}
private void commit() throws Exception {
- inputSource.flush();
+ writer.flush();
}
private void rollback() throws Exception {
- inputSource.clear();
+ writer.clear();
}
}