RESOLVED - BATCH-662: simplify FlatFileItemReader
removed redundant ResourceLineReader
This commit is contained in:
robokaso
@@ -30,7 +30,6 @@ import org.springframework.batch.item.file.mapping.FieldSet;
|
||||
import org.springframework.batch.item.file.mapping.FieldSetMapper;
|
||||
import org.springframework.batch.item.file.separator.DefaultRecordSeparatorPolicy;
|
||||
import org.springframework.batch.item.file.separator.RecordSeparatorPolicy;
|
||||
import org.springframework.batch.item.file.separator.ResourceLineReader;
|
||||
import org.springframework.batch.item.file.transform.AbstractLineTokenizer;
|
||||
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
|
||||
import org.springframework.batch.item.file.transform.LineTokenizer;
|
||||
@@ -48,9 +47,7 @@ import org.springframework.util.ClassUtils;
|
||||
* {@link LineTokenizer} is used to parse data obtained from the file. <br/>
|
||||
*
|
||||
* A {@link FlatFileItemReader} is not thread safe because it maintains state in
|
||||
* the form of a {@link ResourceLineReader}. Be careful to configure a
|
||||
* {@link FlatFileItemReader} using an appropriate factory or scope so that it
|
||||
* is not shared between threads.<br/>
|
||||
* instance variables. <br/>
|
||||
*
|
||||
* <p>
|
||||
* This class supports restart, skipping invalid lines and storing statistics.
|
||||
|
||||
@@ -1,35 +0,0 @@
|
||||
/*
|
||||
* Copyright 2006-2007 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.item.file.separator;
|
||||
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public interface LineReader extends ItemReader<String> {
|
||||
|
||||
/**
|
||||
* @return position
|
||||
*/
|
||||
int getPosition();
|
||||
|
||||
void open();
|
||||
|
||||
void close();
|
||||
|
||||
}
|
||||
@@ -1,310 +0,0 @@
|
||||
/*
|
||||
* Copyright 2006-2007 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.file.separator;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStreamReader;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.MarkFailedException;
|
||||
import org.springframework.batch.item.ResetFailedException;
|
||||
import org.springframework.batch.item.UnexpectedInputException;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* An input source that reads lines one by one from a resource. <br/>
|
||||
*
|
||||
* A line can consist of multiple lines in the input resource, according to the {@link RecordSeparatorPolicy} in force.
|
||||
* By default a line is either terminated by a newline (as per {@link BufferedReader#readLine()}), or can be continued
|
||||
* onto the next line if a field surrounded by quotes (\") contains a newline.<br/>
|
||||
*
|
||||
* Comment lines can be indicated using a line prefix (or collection of prefixes) and they will be ignored. The default
|
||||
* is "#", so lines starting with a pound sign will be ignored.<br/>
|
||||
*
|
||||
* All the public methods that interact with the underlying resource (open, close, read etc.) are synchronized on this.<br/>
|
||||
*
|
||||
* Package private because this is not intended to be a public API - used internally by the flat file input sources.
|
||||
* That makes abuses of the fact that it is stateful easier to control.<br/>
|
||||
*
|
||||
* @author Dave Syer
|
||||
* @author Rob Harrop
|
||||
*/
|
||||
public class ResourceLineReader implements LineReader {
|
||||
|
||||
private static final Collection<String> DEFAULT_COMMENTS = Collections.singleton("#");
|
||||
|
||||
private static final String DEFAULT_ENCODING = "ISO-8859-1";
|
||||
|
||||
private static final int READ_AHEAD_LIMIT = 100000;
|
||||
|
||||
private final Resource resource;
|
||||
|
||||
private final String encoding;
|
||||
|
||||
private Collection<String> comments = DEFAULT_COMMENTS;
|
||||
|
||||
// Encapsulates the state of the reader.
|
||||
private State state = null;
|
||||
|
||||
private RecordSeparatorPolicy recordSeparatorPolicy = new DefaultRecordSeparatorPolicy();
|
||||
|
||||
public ResourceLineReader(Resource resource) throws IOException {
|
||||
this(resource, DEFAULT_ENCODING);
|
||||
}
|
||||
|
||||
public ResourceLineReader(Resource resource, String encoding) {
|
||||
Assert.notNull(resource, "'resource' cannot be null.");
|
||||
Assert.notNull(encoding, "'encoding' cannot be null.");
|
||||
this.resource = resource;
|
||||
this.encoding = encoding;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for the {@link RecordSeparatorPolicy}. Default value is a {@link DefaultRecordSeparatorPolicy}. Ideally
|
||||
* should not be changed once a reader is in use, but it would not be fatal if it was.
|
||||
*
|
||||
* @param recordSeparatorPolicy the new {@link RecordSeparatorPolicy}
|
||||
*/
|
||||
public void setRecordSeparatorPolicy(RecordSeparatorPolicy recordSeparatorPolicy) {
|
||||
/*
|
||||
* The rest of the code accesses the policy in synchronized blocks, copying the reference before using it. So in
|
||||
* principle it can be changed in flight - the results might not be what the user expected!
|
||||
*/
|
||||
this.recordSeparatorPolicy = recordSeparatorPolicy;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 HashSet<String>(Arrays.asList(comments));
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the next line from the input resource, ignoring comments, and according to the {@link RecordSeparatorPolicy}.
|
||||
*
|
||||
* @return a String.
|
||||
*
|
||||
* @see org.springframework.batch.item.ItemReader#read()
|
||||
*/
|
||||
public synchronized String read() {
|
||||
// Make a copy of the recordSeparatorPolicy reference, in case it is
|
||||
// changed during a read operation (unlikely, but you never know)...
|
||||
RecordSeparatorPolicy recordSeparatorPolicy = this.recordSeparatorPolicy;
|
||||
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 the next non-comment line
|
||||
*/
|
||||
private String readLine() {
|
||||
return getState().readLine();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return
|
||||
*/
|
||||
private State getState() {
|
||||
if (state == null) {
|
||||
open();
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates internal state object.
|
||||
*/
|
||||
public synchronized void open() {
|
||||
state = new State();
|
||||
state.open();
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the reader associated with this input source.
|
||||
*/
|
||||
public synchronized void close() {
|
||||
if (state == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
state.close();
|
||||
} finally {
|
||||
state = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Getter for current line count (not the current number of lines returned).
|
||||
*
|
||||
* @return the current line count.
|
||||
*/
|
||||
public int getPosition() {
|
||||
return getState().getCurrentLineCount();
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark the state for return later with reset. Uses the read-ahead limit from an underlying {@link BufferedReader},
|
||||
* which means that there is a limit to how much data can be recovered if the mark needs to be reset.<br/>
|
||||
*
|
||||
* 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 #reset()
|
||||
*
|
||||
* @throws MarkFailedException if the mark could not be set.
|
||||
*/
|
||||
public synchronized void mark() throws MarkFailedException {
|
||||
getState().mark();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset the reader to the last mark.
|
||||
*
|
||||
* @see #mark()
|
||||
*
|
||||
* @throws ResetFailedException if the reset is unsuccessful, e.g. if the read-ahead limit was breached.
|
||||
*/
|
||||
public synchronized void reset() throws ResetFailedException {
|
||||
getState().reset();
|
||||
}
|
||||
|
||||
private boolean isComment(String line) {
|
||||
for (String prefix : comments) {
|
||||
if (line.startsWith(prefix)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private class State {
|
||||
private BufferedReader reader;
|
||||
|
||||
private int currentLineCount = 0;
|
||||
|
||||
private int markedLineCount = -1;
|
||||
|
||||
public String readLine() {
|
||||
String line = null;
|
||||
|
||||
try {
|
||||
line = this.reader.readLine();
|
||||
if (line == null) {
|
||||
return null;
|
||||
}
|
||||
currentLineCount++;
|
||||
while (isComment(line)) {
|
||||
line = reader.readLine();
|
||||
if (line == null) {
|
||||
return null;
|
||||
}
|
||||
currentLineCount++;
|
||||
}
|
||||
} catch (IOException e) {
|
||||
throw new UnexpectedInputException("Unable to read from resource '" + resource + "' at line "
|
||||
+ currentLineCount, e);
|
||||
}
|
||||
return line;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public void open() {
|
||||
try {
|
||||
reader = new BufferedReader(new InputStreamReader(resource.getInputStream(), encoding));
|
||||
mark();
|
||||
} catch (IOException e) {
|
||||
throw new ItemStreamException("Could not open resource", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the reader and reset the counters.
|
||||
*/
|
||||
public void close() {
|
||||
|
||||
if (reader == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
reader.close();
|
||||
} catch (IOException e) {
|
||||
throw new ItemStreamException("Could not close reader", e);
|
||||
} finally {
|
||||
currentLineCount = 0;
|
||||
markedLineCount = -1;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the current line count
|
||||
*/
|
||||
public int getCurrentLineCount() {
|
||||
return currentLineCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark the underlying reader and set the line counters.
|
||||
*/
|
||||
public void mark() throws MarkFailedException {
|
||||
try {
|
||||
reader.mark(READ_AHEAD_LIMIT);
|
||||
markedLineCount = currentLineCount;
|
||||
} catch (IOException e) {
|
||||
throw new MarkFailedException("Could not mark reader", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset the reader and line counters to the last marked position if possible.
|
||||
*/
|
||||
public void reset() throws ResetFailedException {
|
||||
|
||||
if (markedLineCount < 0) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
this.reader.reset();
|
||||
currentLineCount = markedLineCount;
|
||||
} catch (IOException e) {
|
||||
throw new ResetFailedException("Could not reset reader", e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,196 +0,0 @@
|
||||
/*
|
||||
* Copyright 2006-2007 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.file;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.springframework.batch.item.UnexpectedInputException;
|
||||
import org.springframework.batch.item.file.separator.ResourceLineReader;
|
||||
import org.springframework.batch.item.file.separator.SuffixRecordSeparatorPolicy;
|
||||
import org.springframework.core.io.ByteArrayResource;
|
||||
import org.springframework.core.io.InputStreamResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
|
||||
/**
|
||||
* @author Rob Harrop
|
||||
*/
|
||||
public class ResourceLineReaderTests extends TestCase {
|
||||
|
||||
public void testBadResource() throws Exception {
|
||||
ResourceLineReader reader = new ResourceLineReader(new InputStreamResource(new InputStream() {
|
||||
public int read() throws IOException {
|
||||
throw new IOException("Foo");
|
||||
}
|
||||
}));
|
||||
try {
|
||||
reader.read();
|
||||
fail("Expected UnexpectedInputException");
|
||||
}
|
||||
catch (UnexpectedInputException e) {
|
||||
// expected
|
||||
assertTrue(e.getMessage().startsWith("Unable to read"));
|
||||
}
|
||||
}
|
||||
|
||||
public void testRead() throws Exception {
|
||||
Resource resource = new ByteArrayResource("a,b,c\n1,2,3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
int count = 0;
|
||||
String line;
|
||||
while ((line = (String) reader.read()) != null) {
|
||||
count++;
|
||||
assertNotNull(line);
|
||||
}
|
||||
|
||||
assertEquals(2, count);
|
||||
}
|
||||
|
||||
public void testCloseTwice() throws Exception {
|
||||
Resource resource = new ByteArrayResource("a,b,c\n1,2,3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.open();
|
||||
reader.close();
|
||||
try {
|
||||
reader.close(); // just closing a BufferedReader twice should be fine
|
||||
} catch (Exception e) {
|
||||
fail("Unexpected Exception "+e);
|
||||
}
|
||||
assertEquals("a,b,c", reader.read());
|
||||
}
|
||||
|
||||
public void testEncoding() throws Exception {
|
||||
Resource resource = new ByteArrayResource("a,b,c\n1,2,3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource, "UTF-8");
|
||||
int count = 0;
|
||||
String line;
|
||||
while ((line = (String) reader.read()) != null) {
|
||||
count++;
|
||||
assertNotNull(line);
|
||||
}
|
||||
|
||||
assertEquals(2, count);
|
||||
}
|
||||
|
||||
public void testLineCount() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1,2,\"3\n4\"\n5,6,7".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
assertEquals(2, reader.getPosition());
|
||||
}
|
||||
|
||||
public void testLineContent() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1,2,3\n4\n5,6,7".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
assertEquals("1,2,3", reader.read());
|
||||
assertEquals("4", reader.read());
|
||||
assertEquals("5,6,7", reader.read());
|
||||
}
|
||||
|
||||
public void testLineContentWhenLineContainsQuotedNewline() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1,2,\"3\n4\"\n5,6,7".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
assertEquals("1,2,\"3\n4\"", reader.read());
|
||||
assertEquals("5,6,7", reader.read());
|
||||
}
|
||||
|
||||
public void testLineEndings() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n2\r\n3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
String line = (String) reader.read();
|
||||
assertEquals("2", line);
|
||||
assertEquals(2, reader.getPosition());
|
||||
line = (String) reader.read();
|
||||
assertEquals("3", line);
|
||||
assertEquals(3, reader.getPosition());
|
||||
}
|
||||
|
||||
public void testDefaultComments() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n# 2\n3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
String line = (String) reader.read();
|
||||
assertEquals("3", line);
|
||||
}
|
||||
|
||||
public void testComments() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n-- 2\n3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.setComments(new String[] {"//", "--"});
|
||||
reader.read();
|
||||
String line = (String) reader.read();
|
||||
assertEquals("3", line);
|
||||
}
|
||||
|
||||
public void testCommentOnTheLastLine() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n#last line".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
String line = (String) reader.read();
|
||||
assertNull(line);
|
||||
}
|
||||
|
||||
public void testResetNewReader() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n4\n5".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.reset();
|
||||
assertEquals(0, reader.getPosition());
|
||||
}
|
||||
|
||||
public void testMarkReset() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n4\n5".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
assertEquals(1, reader.getPosition());
|
||||
reader.mark();
|
||||
reader.read();
|
||||
assertEquals(2, reader.getPosition());
|
||||
reader.reset();
|
||||
reader.read();
|
||||
assertEquals(2, reader.getPosition());
|
||||
}
|
||||
|
||||
public void testMarkOnFirstRead() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n# 2\n3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
// The first read should do a mark() so the reset goes back to the beginning.
|
||||
reader.reset();
|
||||
String line = (String) reader.read();
|
||||
assertEquals("1", line);
|
||||
}
|
||||
|
||||
public void testMarkAfterClose() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n# 2\n3".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.read();
|
||||
reader.close();
|
||||
reader.mark();
|
||||
}
|
||||
|
||||
public void testNonDefaultRecordSeparatorPolicy() throws Exception {
|
||||
Resource resource = new ByteArrayResource("1\n\"4\n5\"; \n6".getBytes());
|
||||
ResourceLineReader reader = new ResourceLineReader(resource);
|
||||
reader.setRecordSeparatorPolicy(new SuffixRecordSeparatorPolicy());
|
||||
assertEquals(0, reader.getPosition());
|
||||
String line = (String) reader.read();
|
||||
assertEquals("1\"4\n5\"", line);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user