diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index 0c165134e..51531a976 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -77,7 +77,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen LineReader reader = getReader(); Object record = ""; - while (reader.getCurrentLineCount() < lineCount && record != null) { + while (reader.getPosition() < lineCount && record != null) { record = readLine(); } @@ -95,7 +95,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen throw new StreamException("ItemStream not open or already closed."); } ExecutionAttributes executionAttributes = new ExecutionAttributes(); - executionAttributes.putLong(READ_STATISTICS_NAME, reader.getCurrentLineCount()); + executionAttributes.putLong(READ_STATISTICS_NAME, reader.getPosition()); executionAttributes.putLong(SKIPPED_STATISTICS_NAME, skippedLines.size()); return executionAttributes; } @@ -130,7 +130,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen * Skip the current line which is being processed. */ public void skip() { - Integer count = new Integer(getReader().getCurrentLineCount()); + Integer count = new Integer(getReader().getPosition()); // we are not really thread safe so we don't need to synchronize skippedLines.add(count); log.debug("Skipping line in template=[" + this + "], line=" + count); @@ -138,7 +138,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen protected String readLine() { String line = super.readLine(); - while (line != null && skippedLines.contains(new Integer(getReader().getCurrentLineCount()))) { + while (line != null && skippedLines.contains(new Integer(getReader().getPosition()))) { line = super.readLine(); } return line; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java index 8ef2fef8d..3297d5d28 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java @@ -223,7 +223,7 @@ public class SimpleFlatFileItemReader extends AbstractItemReader implements Item return fieldSetMapper.mapLine(tokenizedLine); } catch (RuntimeException ex) { // add current line count to message and re-throw - int lineCount = getReader().getCurrentLineCount(); + int lineCount = getReader().getPosition(); throw new FlatFileParsingException("Parsing error at line: "+lineCount+" in resource="+path+", input=["+line+"]", ex, line, lineCount); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/LineReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/LineReader.java index 808f84416..2e02a310d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/LineReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/LineReader.java @@ -16,26 +16,17 @@ package org.springframework.batch.io.file.separator; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; /** * @author Dave Syer * */ -public interface LineReader extends ItemReader { +public interface LineReader extends ItemReader, ItemStream { /** * @return */ - int getCurrentLineCount(); - - /** - * - */ - void mark(); - - /** - * - */ - void reset(); + int getPosition(); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java index e4e81b0e3..9c09251a5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java @@ -27,7 +27,8 @@ import java.util.Iterator; import org.springframework.batch.io.exception.BatchEnvironmentException; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.reader.AbstractItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.beans.factory.DisposableBean; import org.springframework.core.io.Resource; import org.springframework.util.Assert; @@ -55,7 +56,7 @@ import org.springframework.util.Assert; * @author Dave Syer * @author Rob Harrop */ -public class ResourceLineReader extends AbstractItemReader implements LineReader, ItemReader, +public class ResourceLineReader extends ItemStreamAdapter implements LineReader, ItemReader, DisposableBean { private static final Collection DEFAULT_COMMENTS = Collections.singleton("#"); @@ -199,9 +200,21 @@ public class ResourceLineReader extends AbstractItemReader implements LineReader * * @return the current line count. */ - public int getCurrentLineCount() { + public int getPosition() { return getState().getCurrentLineCount(); } + + /** + * 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.ItemStream#isMarkSupported() + */ + public boolean isMarkSupported() { + return true; + } /** * Mark the state for return later with reset. Uses the read-ahead limit diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java index f11b57fcf..ddc7da6dc 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java @@ -51,12 +51,4 @@ public interface ItemReader { */ Object read() throws Exception; - /** - * Close the reader, freeing any resources that may have been allocated - * since the first call to read(). - * - * TODO: this is only used in sandbox? - * - */ - void close() throws StreamException; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/ResourceLineReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/ResourceLineReaderTests.java index a4b58a411..279242d70 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/ResourceLineReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/ResourceLineReaderTests.java @@ -92,7 +92,7 @@ public class ResourceLineReaderTests extends TestCase { Resource resource = new ByteArrayResource("1,2,\"3\n4\"\n5,6,7".getBytes()); ResourceLineReader reader = new ResourceLineReader(resource); reader.read(); - assertEquals(2, reader.getCurrentLineCount()); + assertEquals(2, reader.getPosition()); } public void testLineContent() throws Exception { @@ -116,10 +116,10 @@ public class ResourceLineReaderTests extends TestCase { reader.read(); String line = (String) reader.read(); assertEquals("2", line); - assertEquals(2, reader.getCurrentLineCount()); + assertEquals(2, reader.getPosition()); line = (String) reader.read(); assertEquals("3", line); - assertEquals(3, reader.getCurrentLineCount()); + assertEquals(3, reader.getPosition()); } public void testDefaultComments() throws Exception { @@ -151,20 +151,20 @@ public class ResourceLineReaderTests extends TestCase { Resource resource = new ByteArrayResource("1\n4\n5".getBytes()); ResourceLineReader reader = new ResourceLineReader(resource); reader.reset(); - assertEquals(0, reader.getCurrentLineCount()); + 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.getCurrentLineCount()); + assertEquals(1, reader.getPosition()); reader.mark(); reader.read(); - assertEquals(2, reader.getCurrentLineCount()); + assertEquals(2, reader.getPosition()); reader.reset(); reader.read(); - assertEquals(2, reader.getCurrentLineCount()); + assertEquals(2, reader.getPosition()); } public void testMarkOnFirstRead() throws Exception { @@ -181,7 +181,7 @@ public class ResourceLineReaderTests extends TestCase { Resource resource = new ByteArrayResource("1\n\"4\n5\"; \n6".getBytes()); ResourceLineReader reader = new ResourceLineReader(resource); reader.setRecordSeparatorPolicy(new SuffixRecordSeparatorPolicy()); - assertEquals(0, reader.getCurrentLineCount()); + assertEquals(0, reader.getPosition()); String line = (String) reader.read(); assertEquals("1\"4\n5\"", line); }