CSV: stream may end with delimiter (#1711)

CSV: stream may end with delimiter
akka · May 22, 2019 · 459a684 · 459a684
2 parents f1ec152 + efa7999
commit 459a684
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 16 deletions.
diff --git a/csv/src/main/scala/akka/stream/alpakka/csv/impl/CsvParser.scala b/csv/src/main/scala/akka/stream/alpakka/csv/impl/CsvParser.scala
@@ -114,19 +114,19 @@ import scala.collection.mutable
       current = next.iterator
     }
 
-  def poll(requireLineEnd: Boolean): Option[List[ByteString]] =
-    if (buffer.nonEmpty) {
-      val line = parseLine(requireLineEnd)
-      if (line.nonEmpty) {
-        currentLineNo += 1
-        if (state == LineEnd) {
-          state = LineStart
-        }
-        resetLine()
-        columns.clear()
+  def poll(requireLineEnd: Boolean): Option[List[ByteString]] = {
+    if (buffer.nonEmpty) parseLine()
+    val line = maybeExtractLine(requireLineEnd)
+    if (line.nonEmpty) {
+      currentLineNo += 1
+      if (state == LineEnd || !requireLineEnd) {
+        state = LineStart
       }
-      line
-    } else None
+      resetLine()
+      columns.clear()
+    }
+    line
+  }
 
   private[this] def advance(n: Int = 1): Unit = {
     pos += n
@@ -198,14 +198,12 @@ import scala.collection.mutable
       }
     }
 
-  protected def parseLine(requireLineEnd: Boolean): Option[List[ByteString]] = {
+  private[this] def parseLine(): Unit = {
     if (firstData) {
       checkForByteOrderMark()
       firstData = false
     }
-
     churn()
-    maybeExtractLine(requireLineEnd)
   }
 
   private[this] def churn(): Unit = {
@@ -446,8 +444,10 @@ import scala.collection.mutable
           Some(columns.toList)
         case WithinFieldEscaped | WithinQuotedFieldEscaped =>
           noCharEscaped()
-        case _ =>
+        case _ if columns.nonEmpty =>
           Some(columns.toList)
+        case _ =>
+          None
       }
     }
 

diff --git a/csv/src/test/scala/akka/stream/alpakka/csv/CsvParserSpec.scala b/csv/src/test/scala/akka/stream/alpakka/csv/CsvParserSpec.scala
@@ -238,6 +238,19 @@ class CsvParserSpec extends WordSpec with Matchers with OptionValues {
       exception.getMessage should be("unclosed quote at end of input 1:6, no matching quote found")
     }
 
+    "accept delimiter as last input" in {
+      val parser = new CsvParser(delimiter = ',', quoteChar = '"', escapeChar = '\\', maximumLineLength)
+      parser.offer(ByteString("A,B\nA,"))
+      parser.poll(requireLineEnd = false).value.map(_.utf8String) should be(List("A", "B"))
+      parser.poll(requireLineEnd = false).value shouldBe List(ByteString("A"), ByteString.empty)
+    }
+
+    "accept delimiter as last input on first line" in {
+      val parser = new CsvParser(delimiter = ',', quoteChar = '"', escapeChar = '\\', maximumLineLength)
+      parser.offer(ByteString("A,"))
+      parser.poll(requireLineEnd = false).value shouldBe List(ByteString("A"), ByteString.empty)
+    }
+
     "detect line ending correctly if input is split between CR & LF" in {
       val parser = new CsvParser(delimiter = ',', quoteChar = '"', escapeChar = '\\', maximumLineLength)
       parser.offer(ByteString("A,D\r"))

diff --git a/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala b/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala
@@ -135,6 +135,20 @@ class CsvParsingSpec extends CsvSpec {
       sink.expectComplete()
     }
 
+    "read all lines without final line end and last column empty" in {
+      val result = Source
+        .single(ByteString("""eins,zwei,drei
+          |uno,""".stripMargin))
+        .via(CsvParsing.lineScanner())
+        .map(_.map(_.utf8String))
+        .runWith(Sink.seq)
+        .futureValue
+
+      result should have size 2
+      result.head should be(List("eins", "zwei", "drei"))
+      result(1) should be(List("uno", ""))
+    }
+
     "parse Apple Numbers exported file" in assertAllStagesStopped {
       val fut =
         FileIO