From 8f29cafe4756454d93dc16f8ea652341d9bb40bc Mon Sep 17 00:00:00 2001 From: robokaso Date: Mon, 22 Sep 2008 13:30:42 +0000 Subject: [PATCH] OPEN - BATCH-806: Resurrect ResourceLineReader RLR is back on stage - TODO find a sensible way to encapsulate duplications with FFIR --- .../batch/item/file/ResourceLineReader.java | 176 ++++++++++++++++++ .../item/file/ResourceLineReaderTests.java | 168 +++++++++++++++++ 2 files changed, 344 insertions(+) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceLineReader.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourceLineReaderTests.java diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceLineReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceLineReader.java new file mode 100644 index 000000000..d4b68954b --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceLineReader.java @@ -0,0 +1,176 @@ +package org.springframework.batch.item.file; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ReaderNotOpenException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.item.file.separator.RecordSeparatorPolicy; +import org.springframework.batch.item.file.separator.SimpleRecordSeparatorPolicy; +import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * Restartable {@link ItemReader} that reads lines from input + * {@link #setResource(Resource)}. Line is defined by the + * {@link #setRecordSeparatorPolicy(RecordSeparatorPolicy)}. + * + * @author Robert Kasanicky + */ +public class ResourceLineReader extends AbstractItemCountingItemStreamItemReader implements + ResourceAwareItemReaderItemStream { + + private static final Log logger = LogFactory.getLog(ResourceLineReader.class); + + // default encoding for input files + public static final String DEFAULT_CHARSET = "ISO-8859-1"; + + private RecordSeparatorPolicy recordSeparatorPolicy = new SimpleRecordSeparatorPolicy(); + + private Resource resource; + + private BufferedReader reader; + + private int lineCount = 0; + + private String[] comments = new String[] { "#" }; + + private boolean noInput = false; + + private String encoding = DEFAULT_CHARSET; + + public ResourceLineReader() { + setName(ClassUtils.getShortName(FlatFileItemReader.class)); + } + + /** + * Setter for the encoding for this input source. Default value is + * {@link #DEFAULT_CHARSET}. + * + * @param encoding a properties object which possibly contains the encoding + * for this input file; + */ + public void setEncoding(String encoding) { + this.encoding = encoding; + } + + /** + * Setter for comment prefixes. Can be used to ignore header lines as well + * by using e.g. the first couple of column names as a prefix. + * + * @param comments an array of comment line prefixes. + */ + public void setComments(String[] comments) { + this.comments = new String[comments.length]; + System.arraycopy(comments, 0, this.comments, 0, comments.length); + } + + /** + * Public setter for the input resource. + */ + public void setResource(Resource resource) { + this.resource = resource; + } + + /** + * Public setter for the recordSeparatorPolicy. Used to determine where the + * line endings are and do things like continue over a line ending if inside + * a quoted string. + * + * @param recordSeparatorPolicy the recordSeparatorPolicy to set + */ + public void setRecordSeparatorPolicy(RecordSeparatorPolicy recordSeparatorPolicy) { + this.recordSeparatorPolicy = recordSeparatorPolicy; + } + + /** + * @return string corresponding to logical record according to + * {@link #setRecordSeparatorPolicy(RecordSeparatorPolicy)} (might span + * multiple lines in file). + */ + @Override + protected String doRead() { + if (noInput) { + return null; + } + String line = readLine(); + String record = line; + if (line != null) { + while (line != null && !recordSeparatorPolicy.isEndOfRecord(record)) { + record = recordSeparatorPolicy.preProcess(record) + (line = readLine()); + } + } + return recordSeparatorPolicy.postProcess(record); + } + + /** + * @return next line (skip comments). + */ + private String readLine() { + + if (reader == null) { + throw new ReaderNotOpenException("Reader must be open before it can be read."); + } + + String line = null; + + try { + line = this.reader.readLine(); + if (line == null) { + return null; + } + lineCount++; + while (isComment(line)) { + line = reader.readLine(); + if (line == null) { + return null; + } + lineCount++; + } + } + catch (IOException e) { + throw new UnexpectedInputException("Unable to read from resource '" + resource + "' at line " + lineCount, + e); + } + return line; + } + + private boolean isComment(String line) { + for (String prefix : comments) { + if (line.startsWith(prefix)) { + return true; + } + } + return false; + } + + @Override + protected void doClose() throws Exception { + lineCount = 0; + reader.close(); + + } + + @Override + protected void doOpen() throws Exception { + Assert.notNull(resource, "Input resource must be set"); + Assert.notNull(recordSeparatorPolicy, "RecordSeparatorPolicy must be set"); + + noInput = false; + if (!resource.exists()) { + noInput = true; + logger.warn("Input resource does not exist"); + return; + } + + reader = new BufferedReader(new InputStreamReader(resource.getInputStream(), encoding)); + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourceLineReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourceLineReaderTests.java new file mode 100644 index 000000000..9a598a14c --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourceLineReaderTests.java @@ -0,0 +1,168 @@ +package org.springframework.batch.item.file; + +import static org.junit.Assert.*; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.file.separator.RecordSeparatorPolicy; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.DescriptiveResource; +import org.springframework.core.io.Resource; + +/** + * Tests for {@link ResourceLineReader}. + */ +public class ResourceLineReaderTests { + + private ResourceLineReader tested; + + private static final Resource resource = new ByteArrayResource("line1\nline2\nline3\nline4\nline5".getBytes()); + + private ExecutionContext executionContext = new ExecutionContext(); + + private static ResourceLineReader getItemReader() { + ResourceLineReader result = new ResourceLineReader(); + result.setResource(resource); + return result; + } + + @Before + public void setUp() throws Exception { + tested = getItemReader(); + tested.open(executionContext); + } + + /** + * Regular scenario - read the input and eventually return null. + */ + @Test + public void testRead() throws Exception { + + assertEquals("line1", tested.read()); + assertEquals("line2", tested.read()); + assertEquals("line3", tested.read()); + assertEquals("line4", tested.read()); + assertEquals("line5", tested.read()); + + assertNull(tested.read()); + } + + /** + * No input should be handled gracefully - null is returned on first + * read. + */ + @Test + public void testNoInput() throws Exception { + tested = getItemReader(); + tested.setResource(new DescriptiveResource("doesn't exist") { + + @Override + public boolean exists() { + return false; + } + + }); + tested.open(executionContext); + assertNull(tested.read()); + } + + /** + * Restart scenario - read items, update execution context, create new + * reader and restore from restart data - the new input source should + * continue where the old one finished. + */ + @Test + public void testRestart() throws Exception { + + tested.update(executionContext); + + assertEquals("line1", tested.read()); + assertEquals("line2", tested.read()); + + tested.update(executionContext); + + // create new input source + tested = getItemReader(); + + tested.open(executionContext); + + assertEquals("line3", tested.read()); + } + + /** + * Restart scenario - read items, rollback to last marked position, update + * execution context, create new reader and restore from restart data - the + * new input source should continue where the old one finished. + */ + @Test + public void testResetAndRestart() throws Exception { + + tested.update(executionContext); + + assertEquals("line1", tested.read()); + + assertEquals("line2", tested.read()); + + tested.update(executionContext); + + assertEquals("line3", tested.read()); + + // create new input source + tested = getItemReader(); + + tested.open(executionContext); + + assertEquals("line3", tested.read()); + } + + @Test + public void testReopen() throws Exception { + tested.update(executionContext); + + assertEquals("line1", tested.read()); + assertEquals("line2", tested.read()); + + tested.update(executionContext); + + + tested.close(executionContext); + tested.open(executionContext); + + assertEquals("line3", tested.read()); + } + + @Test + public void testRestartWithCustomRecordSeparatorPolicy() throws Exception { + + tested.setRecordSeparatorPolicy(new RecordSeparatorPolicy() { + // 1 record = 2 lines + boolean pair = true; + + public boolean isEndOfRecord(String line) { + pair = !pair; + return pair; + } + + public String postProcess(String record) { + return record; + } + + public String preProcess(String record) { + return record; + } + }); + + tested.open(executionContext); + + assertEquals("line1line2", tested.read()); + + tested.update(executionContext); + tested.close(executionContext); + tested.open(executionContext); + + assertEquals("line3line4", tested.read()); + + } + +}