Merge pull request #625 from saleemno1/chunked-reader-lookahead

fill the lookahead buffer on short reads from a chunked source

Author: garydgregory

src/main/java/org/apache/commons/csv/ExtendedBufferedReader.java

@@ -225,6 +225,37 @@ public void mark(final int readAheadLimit) throws IOException {
         super.mark(readAheadLimit);
     }
 
+    /**
+     * Fills {@code array} with the characters that follow the current position without consuming them.
+     * <p>
+     * Overridden because the inherited implementation stops at the first short read, which leaves the tail of the array holding stale content when the source
+     * delivers data in chunks. Callers compare the whole array against a multi-character delimiter, so a partial fill makes them miss a delimiter that is
+     * really there.
+     * </p>
+     *
+     * @param array the buffer to fill.
+     * @return the number of characters peeked, or {@link IOUtils#EOF} at the end of the stream.
+     * @throws IOException If an I/O error occurs.
+     */
+    @Override
+    public int peek(final char[] array) throws IOException {
+        final int length = array.length;
+        if (length == 0) {
+            return 0;
+        }
+        super.mark(length);
+        int len = 0;
+        while (len < length) {
+            final int more = super.read(array, len, length - len);
+            if (more == EOF) {
+                break;
+            }
+            len += more;
+        }
+        super.reset();
+        return len == 0 ? EOF : len;
+    }
+
     @Override
     public int read() throws IOException {
         final int current = super.read();
@@ -244,7 +275,16 @@ public int read(final char[] buf, final int offset, final int length) throws IOE
         if (length == 0) {
             return 0;
         }
-        final int len = super.read(buf, offset, length);
+        int len = super.read(buf, offset, length);
+        // The underlying buffered reader stops early once the source reports it is not ready, so a stream that delivers data in chunks (a socket or a pipe)
+        // yields a short read. Callers match multi-character sequences against this buffer, so keep reading until it is full or the source is exhausted.
+        while (len > 0 && len < length) {
+            final int more = super.read(buf, offset + len, length - len);
+            if (more == EOF) {
+                break;
+            }
+            len += more;
+        }
         if (encoder != null && len > 0) {
             this.bytesRead += getEncodedCharLength(buf, offset, len);
         }

src/test/java/org/apache/commons/csv/CSVParserTest.java

@@ -1709,6 +1709,19 @@ void testParseWithDelimiterStringWithEscape() throws IOException {
         }
     }
 
+    @Test
+    void testParseWithDelimiterStringFromChunkedReader() throws IOException {
+        final CSVFormat csvFormat = CSVFormat.DEFAULT.builder().setDelimiter("[|]").setEscape('!').get();
+        try (CSVParser csvParser = csvFormat.parse(new ChunkedReader("a[|]b\r\nc![!|!]d[|]e"))) {
+            CSVRecord csvRecord = csvParser.nextRecord();
+            assertEquals("a", csvRecord.get(0));
+            assertEquals("b", csvRecord.get(1));
+            csvRecord = csvParser.nextRecord();
+            assertEquals("c[|]d", csvRecord.get(0));
+            assertEquals("e", csvRecord.get(1));
+        }
+    }
+
     @Test
     void testParseWithDelimiterStringWithQuote() throws IOException {
         final String source = "'a[|]b[|]c'[|]xyz\r\nabc[abc][|]xyz";

src/test/java/org/apache/commons/csv/ChunkedReader.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   https://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.apache.commons.csv;
+
+import java.io.IOException;
+import java.io.StringReader;
+
+/**
+ * A {@link StringReader} that hands out one character per call and never reports itself ready, like a socket or a pipe that delivers data in chunks.
+ */
+final class ChunkedReader extends StringReader {
+
+    ChunkedReader(final String content) {
+        super(content);
+    }
+
+    @Override
+    public int read(final char[] buf, final int offset, final int length) throws IOException {
+        return length <= 0 ? 0 : super.read(buf, offset, 1);
+    }
+
+    @Override
+    public boolean ready() {
+        return false;
+    }
+}

src/test/java/org/apache/commons/csv/ExtendedBufferedReaderTest.java

@@ -240,4 +240,37 @@ void testReadLookahead2() throws Exception {
             assertEquals('d', br.getLastChar());
         }
     }
+
+    @Test
+    void testReadAndPeekArrayFromChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("abcdef"))) {
+            final char[] peeked = new char[3];
+            assertEquals(3, br.peek(peeked));
+            assertArrayEquals(new char[] { 'a', 'b', 'c' }, peeked);
+            final char[] read = new char[3];
+            assertEquals(3, br.read(read, 0, 3));
+            assertArrayEquals(new char[] { 'a', 'b', 'c' }, read);
+        }
+    }
+
+    @Test
+    void testReadArrayPastEndOfChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("ab"))) {
+            final char[] read = new char[4];
+            assertEquals(2, br.read(read, 0, 4));
+            assertEquals('a', read[0]);
+            assertEquals('b', read[1]);
+            assertEquals(EOF, br.read(read, 0, 4));
+        }
+    }
+
+    @Test
+    void testPeekArrayPastEndOfChunkedReader() throws Exception {
+        try (ExtendedBufferedReader br = new ExtendedBufferedReader(new ChunkedReader("ab"))) {
+            final char[] peeked = new char[4];
+            assertEquals(2, br.peek(peeked));
+            assertEquals('a', peeked[0]);
+            assertEquals('b', peeked[1]);
+        }
+    }
 }