Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - issue BATCH-433: FlatFileItemWriter should buffer output

Browse Source 
http://jira.springframework.org/browse/BATCH-433
This commit is contained in:
robokaso
2008-03-25 16:08:59 +00:00
parent 44c478cd0e
commit f888d44336
2 changed files with 152 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
View File

@@ -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.<br/>
* 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.<br/>
*
* Uses buffered writer to improve performance.<br/>
*
* Use {@link #write(String)} method to output a line to an item writer.
*
* <p>
* 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
* </p>
*
* @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.<br/>
*
* @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
*/

121
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java
View File

@@ -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();
}
}