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

RESOLVED - BATCH-548: skipping items can lead to an invalidated mark in the ResourceItemReader

Browse Source 
added item buffering to FlatFileItemReader
This commit is contained in:
robokaso
2008-06-11 11:28:56 +00:00
parent 40594d75e7
commit dcca8b74cb
3 changed files with 76 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
View File

@@ -18,11 +18,9 @@ package org.springframework.batch.item.file;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ExecutionContextUserSupport;
import org.springframework.batch.item.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemReaderException;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ReaderNotOpenException;
import org.springframework.batch.item.ResourceAwareItemReaderItemStream;
@@ -62,12 +60,10 @@ import org.springframework.util.ClassUtils;
* @author Robert Kasanicky
* @author Dave Syer
*/
public class FlatFileItemReader extends ExecutionContextUserSupport implements ResourceAwareItemReaderItemStream, InitializingBean {
public class FlatFileItemReader extends AbstractBufferedItemReaderItemStream implements ResourceAwareItemReaderItemStream, InitializingBean {
private static Log log = LogFactory.getLog(FlatFileItemReader.class);
private static final String LINES_READ_COUNT = "lines.read.count";
// default encoding for input files
public static final String DEFAULT_CHARSET = "ISO-8859-1";
@@ -87,8 +83,6 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements R
private FieldSetMapper fieldSetMapper;
private boolean saveState = false;
/**
* Encapsulates the state of the input source. If it is null then we are
* uninitialized.
@@ -99,135 +93,11 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements R
setName(ClassUtils.getShortName(FlatFileItemReader.class));
}
/**
* Initialize the reader if necessary.
*
* @throws IllegalStateException if the resource cannot be opened
*/
public void open(ExecutionContext executionContext) throws ItemStreamException {
Assert.state(resource.exists(), "Resource must exist: [" + resource + "]");
log.debug("Opening flat file for reading: " + resource);
if (this.reader == null) {
ResourceLineReader reader = new ResourceLineReader(resource, encoding);
if (recordSeparatorPolicy != null) {
reader.setRecordSeparatorPolicy(recordSeparatorPolicy);
}
if (comments != null) {
reader.setComments(comments);
}
reader.open();
this.reader = reader;
}
for (int i = 0; i < linesToSkip; i++) {
protected void jumpToItem(int itemIndex) throws Exception {
Object record = "";
while (reader.getPosition() < itemIndex && record != null) {
readLine();
}
if (firstLineIsHeader) {
// skip the header
String firstLine = readLine();
// set names in tokenizer if they haven't been set already
if (tokenizer instanceof AbstractLineTokenizer && !((AbstractLineTokenizer) tokenizer).hasNames()) {
String[] names = tokenizer.tokenize(firstLine).getValues();
((AbstractLineTokenizer) tokenizer).setNames(names);
}
}
if (executionContext.containsKey(getKey(LINES_READ_COUNT))) {
log.debug("Initializing for restart. Restart data is: " + executionContext);
long lineCount = executionContext.getLong(getKey(LINES_READ_COUNT));
LineReader reader = getReader();
Object record = "";
while (reader.getPosition() < lineCount && record != null) {
record = readLine();
}
}
}
/**
* Close and null out the reader.
*
* @throws ItemStreamException
*/
public void close(ExecutionContext executionContext) throws ItemStreamException {
try {
if (reader != null) {
log.debug("Closing flat file for reading: " + resource);
reader.close(null);
}
}
finally {
reader = null;
}
}
/**
* Reads a line from input, tokenizes is it using the
* {@link #setLineTokenizer(LineTokenizer)} and maps to domain object using
* {@link #setFieldSetMapper(FieldSetMapper)}.
*
* @see org.springframework.batch.item.ItemReader#read()
*/
public Object read() throws Exception {
String line = readLine();
if (line != null) {
int lineCount = getReader().getPosition();
try {
FieldSet tokenizedLine = tokenizer.tokenize(line);
return fieldSetMapper.mapLine(tokenizedLine);
}
catch (RuntimeException ex) {
// add current line count to message and re-throw
throw new FlatFileParseException("Parsing error at line: " + lineCount + " in resource="
+ resource.getDescription() + ", input=[" + line + "]", ex, line, lineCount);
}
}
return null;
}
/**
* This method returns the execution attributes for the reader. It returns
* the current Line Count which can be used to reinitialise the batch job in
* case of restart.
*/
public void update(ExecutionContext executionContext) {
if (reader == null) {
throw new ItemStreamException("ItemStream not open or already closed.");
}
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(getKey(LINES_READ_COUNT), reader.getPosition());
}
}
/**
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
getReader().mark();
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
getReader().reset();
}
/**
@@ -345,16 +215,74 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements R
Assert.notNull(fieldSetMapper, "FieldSetMapper must not be null.");
}
protected void doClose() throws Exception {
try {
if (reader != null) {
log.debug("Closing flat file for reading: " + resource);
reader.close(null);
}
}
finally {
reader = null;
}
}
protected void doOpen() throws Exception {
Assert.state(resource.exists(), "Resource must exist: [" + resource + "]");
log.debug("Opening flat file for reading: " + resource);
if (this.reader == null) {
ResourceLineReader reader = new ResourceLineReader(resource, encoding);
if (recordSeparatorPolicy != null) {
reader.setRecordSeparatorPolicy(recordSeparatorPolicy);
}
if (comments != null) {
reader.setComments(comments);
}
reader.open();
this.reader = reader;
}
for (int i = 0; i < linesToSkip; i++) {
readLine();
}
if (firstLineIsHeader) {
// skip the header
String firstLine = readLine();
// set names in tokenizer if they haven't been set already
if (tokenizer instanceof AbstractLineTokenizer && !((AbstractLineTokenizer) tokenizer).hasNames()) {
String[] names = tokenizer.tokenize(firstLine).getValues();
((AbstractLineTokenizer) tokenizer).setNames(names);
}
}
}
/**
* 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.
* Reads a line from input, tokenizes is it using the
* {@link #setLineTokenizer(LineTokenizer)} and maps to domain object using
* {@link #setFieldSetMapper(FieldSetMapper)}.
*
* @param saveState
* @see org.springframework.batch.item.ItemReader#read()
*/
public void setSaveState(boolean saveState) {
this.saveState = saveState;
protected Object doRead() throws Exception {
String line = readLine();
if (line != null) {
int lineCount = getReader().getPosition();
try {
FieldSet tokenizedLine = tokenizer.tokenize(line);
return fieldSetMapper.mapLine(tokenizedLine);
}
catch (RuntimeException ex) {
// add current line count to message and re-throw
throw new FlatFileParseException("Parsing error at line: " + lineCount + " in resource="
+ resource.getDescription() + ", input=[" + line + "]", ex, line, lineCount);
}
}
return null;
}
}

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java
View File

@@ -152,7 +152,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase {
// get restart data
reader.update(executionContext);
assertEquals(4, executionContext.getLong(ClassUtils.getShortName(FlatFileItemReader.class)
+ ".lines.read.count"));
+ ".item.count"));
// close input
reader.close(executionContext);
@@ -167,7 +167,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase {
reader.update(executionContext);
assertEquals(6, executionContext.getLong(ClassUtils.getShortName(FlatFileItemReader.class)
+ ".lines.read.count"));
+ ".item.count"));
}
}

7
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderBasicTests.java
View File

@@ -194,7 +194,7 @@ public class FlatFileItemReaderBasicTests extends TestCase {
try {
itemReader.open(executionContext);
fail("Expected IllegalArgumentException");
} catch (IllegalArgumentException e) {
} catch (ItemStreamException e) {
// expected
}
}
@@ -208,7 +208,8 @@ public class FlatFileItemReaderBasicTests extends TestCase {
fail("Expected BatchEnvironmentException");
} catch (ItemStreamException e) {
// expected
assertEquals("foo", e.getCause().getMessage());
assertEquals("Failed to initialize the reader", e.getMessage());
assertEquals("foo", e.getCause().getCause().getMessage());
}
}
@@ -290,7 +291,7 @@ public class FlatFileItemReaderBasicTests extends TestCase {
try {
testReader.open(executionContext);
fail();
} catch (IllegalStateException ex) {
} catch (ItemStreamException ex) {
// expected
}