Fix max line length not always being checked

modules/core/src/main/scala/ceesvee/CsvParser.scala

@@ -142,6 +142,11 @@ object CsvParser {
         } else {
           val str = in.next()
           val (newState, lines) = splitStrings(List(str), state)
+          lines.foreach { line =>
+            if (line.length > options.maximumLineLength) {
+              throw Error.LineTooLong(options.maximumLineLength)
+            }
+          }
           val _ = toOutput.enqueueAll(lines)
           state = newState
           next()

modules/core/src/main/scala/ceesvee/CsvParserVector.scala

@@ -60,6 +60,11 @@ object CsvParserVector {
         } else {
           val bytes = in.next()
           val (newState, lines) = splitBytes(bytes, state)
+          lines.foreach { line =>
+            if (line.length > options.maximumLineLength) {
+              throw Error.LineTooLong(options.maximumLineLength)
+            }
+          }
           val _ = toOutput.enqueueAll(lines)
           state = newState
           next()

modules/core/src/test/scala/ceesvee/CsvParserSpec.scala

@@ -19,7 +19,7 @@ object CsvParserSpec extends ZIOSpecDefault
   override protected def parse(lines: Iterable[String], options: CsvParser.Options) = {
     val input = lines.mkString("\n").grouped(8192)
     val result = CsvParser.parse[List](input, options)
-    ZIO.succeed(Chunk.fromIterator(result))
+    ZIO.attempt(Chunk.fromIterator(result)).refineOrDie { case e: CsvParser.Error => e }
   }
 
   override protected def parseLine(line: String, options: CsvParser.Options) = {
@@ -37,7 +37,7 @@ trait CsvParserParserSuite { self: ZIOSpecDefault =>
   protected def parse(
     lines: Iterable[String],
     options: CsvParser.Options,
-  ): ZIO[Any, Throwable, Chunk[List[String]]]
+  ): ZIO[Any, CsvParser.Error, Chunk[List[String]]]
 
   protected def parserSuite = suite("parser")(
     test("lots") {
@@ -48,6 +48,20 @@ trait CsvParserParserSuite { self: ZIOSpecDefault =>
         assertTrue(result.length == 10)
       }
     },
+    suite("maximum line length")(
+      test("oversized line") {
+        val options = CsvParser.Options.Defaults.copy(maximumLineLength = 3)
+        parse(List("abcd", "ok"), options).either.map { result =>
+          assertTrue(result == Left(CsvParser.Error.LineTooLong(3)))
+        }
+      },
+      test("oversized state") {
+        val options = CsvParser.Options.Defaults.copy(maximumLineLength = 3)
+        parse(List("abcd"), options).either.map { result =>
+          assertTrue(result == Left(CsvParser.Error.LineTooLong(3)))
+        }
+      },
+    ),
     suite("comment prefix")({
       val lines = List(
         "a,b,c",

modules/core/src/test/scala/ceesvee/CsvParserVectorSpec.scala

@@ -22,7 +22,7 @@ object CsvParserVectorSpec extends ZIOSpecDefault
   override protected def parse(lines: Iterable[String], options: CsvParser.Options) = {
     val input = lines.mkString("\n").grouped(8192).map(_.getBytes(charset))
     val result = CsvParserVector.parse[List](input, options)
-    ZIO.succeed(Chunk.fromIterator(result))
+    ZIO.attempt(Chunk.fromIterator(result)).refineOrDie { case e: CsvParser.Error => e }
   }
 
   override protected def parseLine(line: String, options: CsvParser.Options) = {

modules/fs2/src/main/scala/ceesvee/fs2/Fs2CsvParser.scala

@@ -1,7 +1,7 @@
 package ceesvee.fs2
 
 import _root_.fs2.Chunk
-import _root_.fs2.Pipe
+import _root_.fs2.Pipe as fs2Pipe
 import _root_.fs2.Pull
 import _root_.fs2.RaiseThrowable
 import _root_.fs2.Stream
@@ -19,12 +19,12 @@ object Fs2CsvParser {
   /**
    * Turns a stream of strings into a stream of CSV records.
    *
-   * Raises a [[Error.LineTooLong]] if a line is longer than
+   * Raises a [[CsvParser.Error.LineTooLong]] if a line is longer than
    * `maximumLineLength`.
    */
   def parse[F[_]: RaiseThrowable](
     options: CsvParser.Options,
-  ): Pipe[F, String, ArraySeq[String]] = {
+  ): fs2Pipe[F, String, ArraySeq[String]] = {
     _.through(splitLines(options))
       .filter(str => !ignoreLine(str, options))
       .map(parseLine[ArraySeq](_, options))
@@ -35,12 +35,12 @@ object Fs2CsvParser {
    *
    * Delimiters within double-quotes are ignored.
    *
-   * Raises a [[Error.LineTooLong]] if a line is longer than
+   * Raises a [[CsvParser.Error.LineTooLong]] if a line is longer than
    * `maximumLineLength`.
    */
   def splitLines[F[_]: RaiseThrowable](
     options: CsvParser.Options,
-  ): Pipe[F, String, String] = {
+  ): fs2Pipe[F, String, String] = {
 
     @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
     def go(stream: Stream[F, String], state: State, first: Boolean): Pull[F, String, Unit] =
@@ -54,7 +54,7 @@ object Fs2CsvParser {
         case Some((chunk, stream)) =>
           val (newState, lines) = splitStrings(chunk.toArraySeq, state)
 
-          if (newState.leftover.length > options.maximumLineLength) {
+          if (newState.leftover.length > options.maximumLineLength || lines.exists(_.length > options.maximumLineLength)) {
             Pull.raiseError[F](Error.LineTooLong(options.maximumLineLength))
           } else {
             Pull.output(Chunk.from(lines)) >> go(stream, newState, first = false)

modules/fs2/src/test/scala/ceesvee/fs2/CsvParserSpec.scala

@@ -21,5 +21,6 @@ object CsvParserSpec extends ZIOSpecDefault with ceesvee.CsvParserParserSuite {
       .compile
       .toList
       .map(Chunk.fromIterable(_))
+      .refineOrDie { case e: CsvParser.Error => e }
   }
 }

modules/zio/src/main/scala/ceesvee/zio/ZioCsvParser.scala

@@ -2,7 +2,7 @@ package ceesvee.zio
 
 import _root_.zio.Chunk
 import _root_.zio.Ref
-import _root_.zio.Trace
+import _root_.zio.Trace as ZIOTrace
 import _root_.zio.ZIO
 import _root_.zio.stream.ZPipeline
 import ceesvee.CsvParser
@@ -19,12 +19,8 @@ object ZioCsvParser {
    */
   def parse(
     options: CsvParser.Options,
-  )(implicit trace: Trace): ZPipeline[Any, Error, String, Chunk[String]] = {
-    _parse(State.initial, options)
-  }
-
-  private[ceesvee] def _parse(state: State, options: CsvParser.Options)(implicit trace: Trace) = {
-    _splitLines(state, options) >>>
+  )(implicit trace: ZIOTrace): ZPipeline[Any, Error, String, Chunk[String]] = {
+    splitLines(options) >>>
       ZPipeline.filter[String](str => !ignoreLine(str, options)) >>>
       ZPipeline.map(parseLine[Chunk](_, options))
   }
@@ -36,27 +32,26 @@ object ZioCsvParser {
    */
   def splitLines(
     options: CsvParser.Options,
-  )(implicit trace: Trace): ZPipeline[Any, CsvParser.Error, String, String] = {
-    _splitLines(State.initial, options)
-  }
-
-  private def _splitLines(
-    state: State,
-    options: CsvParser.Options,
-  )(implicit trace: Trace) = ZPipeline.fromPush {
-    Ref.make(state).map { stateRef => (chunk: Option[Chunk[String]]) =>
-      chunk match {
+  )(implicit trace: ZIOTrace): ZPipeline[Any, CsvParser.Error, String, String] = ZPipeline.fromPush {
+    Ref.make(State.initial).map { stateRef => (chunk: Option[Chunk[String]]) =>
+      (chunk match {
         case None =>
           stateRef.getAndSet(State.initial).map { case State(leftover, _, _) =>
             if (leftover.isEmpty) Chunk.empty else Chunk(leftover)
           }
 
         case Some(strings) =>
           stateRef.get.flatMap { case State(leftover, _, _) =>
-            ZIO.fail(Error.LineTooLong(options.maximumLineLength))
-              .when(leftover.length > options.maximumLineLength)
+            failWhenLineTooLong(leftover, options)
           } *> stateRef.modify(splitStrings(strings, _).swap)
+
+      }).tap { lines =>
+        ZIO.foreachDiscard(lines)(failWhenLineTooLong(_, options))
       }
     }
   }
+
+  private def failWhenLineTooLong(line: String, options: CsvParser.Options) =
+    ZIO.fail(Error.LineTooLong(options.maximumLineLength))
+      .when(line.length > options.maximumLineLength)
 }

modules/zio/src/main/scala/ceesvee/zio/ZioCsvParserVector.scala

@@ -11,6 +11,7 @@ import ceesvee.CsvParserVector
 import java.nio.charset.StandardCharsets
 
 object ZioCsvParserVector {
+  import CsvParser.Error
   import CsvParserVector.State
   import CsvParserVector.parseLine
   import CsvParserVector.splitBytes
@@ -20,12 +21,8 @@ object ZioCsvParserVector {
    */
   def parse(
     options: CsvParser.Options,
-  )(implicit trace: ZIOTrace): ZPipeline[Any, CsvParser.Error, Byte, Chunk[String]] = {
-    _parse(State.initial, options)
-  }
-
-  private[ceesvee] def _parse(state: State, options: CsvParser.Options)(implicit trace: ZIOTrace) = {
-    _splitLines(state, options) >>>
+  )(implicit trace: ZIOTrace): ZPipeline[Any, Error, Byte, Chunk[String]] = {
+    _splitLines(options) >>>
       ZPipeline.filter[Array[Byte]](bytes => !CsvParser.ignoreLine(new String(bytes, StandardCharsets.UTF_8), options)) >>>
       ZPipeline.map(parseLine[Chunk](_, options))
   }
@@ -37,27 +34,32 @@ object ZioCsvParserVector {
    */
   def splitLines(
     options: CsvParser.Options,
-  )(implicit trace: ZIOTrace): ZPipeline[Any, CsvParser.Error, Byte, String] = {
-    _splitLines(State.initial, options).map(new String(_, StandardCharsets.UTF_8))
+  )(implicit trace: ZIOTrace): ZPipeline[Any, Error, Byte, String] = {
+    _splitLines(options).map(new String(_, StandardCharsets.UTF_8))
   }
 
   private def _splitLines(
-    state: State,
     options: CsvParser.Options,
   )(implicit trace: ZIOTrace) = ZPipeline.fromPush {
-    Ref.make(state).map { stateRef => (chunk: Option[Chunk[Byte]]) =>
-      chunk match {
+    Ref.make(State.initial).map { stateRef => (chunk: Option[Chunk[Byte]]) =>
+      (chunk match {
         case None =>
           stateRef.getAndSet(State.initial).map { s =>
             if (s.leftover.isEmpty) Chunk.empty else Chunk(s.leftover)
           }
 
         case Some(bytes) =>
           stateRef.get.flatMap { s =>
-            ZIO.fail(CsvParser.Error.LineTooLong(options.maximumLineLength))
-              .when(s.leftover.length > options.maximumLineLength)
+            failWhenLineTooLong(s.leftover, options)
           } *> stateRef.modify(splitBytes[Chunk](bytes.toArray, _).swap)
+
+      }).tap { bytes =>
+        ZIO.foreachDiscard(bytes)(failWhenLineTooLong(_, options))
       }
     }
   }
+
+  private def failWhenLineTooLong(bytes: Array[Byte], options: CsvParser.Options) =
+    ZIO.fail(Error.LineTooLong(options.maximumLineLength))
+      .when(bytes.length > options.maximumLineLength)
 }

modules/zio/src/main/scala/ceesvee/zio/ZioCsvReader.scala

@@ -2,7 +2,7 @@ package ceesvee.zio
 
 import _root_.zio.Cause
 import _root_.zio.Chunk
-import _root_.zio.Trace
+import _root_.zio.Trace as ZIOTrace
 import _root_.zio.stream.ZPipeline
 import _root_.zio.stream.ZStream
 import ceesvee.CsvHeader
@@ -38,7 +38,7 @@ object ZioCsvReader {
     header: CsvHeader[T],
     options: CsvReader.Options,
   )(implicit
-    trace: Trace,
+    trace: ZIOTrace,
   ): ZStream[R, Either[E, Error], Either[CsvHeader.Errors, T]] = {
     decodeWithHeader_(stream, header)(ZioCsvParser.parse(options))
   }
@@ -50,7 +50,7 @@ object ZioCsvReader {
   )(
     parse: ZPipeline[Any, CsvParser.Error, A, Chunk[String]],
   )(implicit
-    trace: Trace,
+    trace: ZIOTrace,
   ): ZStream[R, Either[E, Error], Either[CsvHeader.Errors, T]] = ZStream.suspend {
     var decoder: CsvHeader.Decoder[T] = null
 
@@ -79,7 +79,7 @@ object ZioCsvReader {
     options: CsvReader.Options,
   )(implicit
     D: CsvRecordDecoder[T],
-    trace: Trace,
+    trace: ZIOTrace,
   ): ZPipeline[Any, CsvParser.Error, String, Either[CsvRecordDecoder.Errors, T]] = {
     ZioCsvParser.parse(options) >>> ZPipeline.map(D.decode(_))
   }

modules/zio/src/main/scala/ceesvee/zio/ZioCsvWriter.scala

@@ -1,6 +1,6 @@
 package ceesvee.zio
 
-import _root_.zio.Trace
+import _root_.zio.Trace as ZIOTrace
 import _root_.zio.stream.ZStream
 import ceesvee.CsvRecordEncoder
 
@@ -13,7 +13,7 @@ object ZioCsvWriter {
   def encodeWithHeader[R, E, A](
     header: Iterable[String],
     rows: ZStream[R, E, A],
-  )(implicit E: CsvRecordEncoder[A], trace: Trace): ZStream[R, E, String] = {
+  )(implicit E: CsvRecordEncoder[A], trace: ZIOTrace): ZStream[R, E, String] = {
     ZStream.succeed(fieldsToLine(header)) ++ encode(rows)
   }
 
@@ -22,7 +22,7 @@ object ZioCsvWriter {
    */
   def encode[R, E, A](
     rows: ZStream[R, E, A],
-  )(implicit E: CsvRecordEncoder[A], trace: Trace): ZStream[R, E, String] = {
+  )(implicit E: CsvRecordEncoder[A], trace: ZIOTrace): ZStream[R, E, String] = {
     rows.map(E.encode(_)).map(fieldsToLine(_))
   }
 }

modules/zio/src/test/scala/ceesvee/zio/ZioCsvParserSpec.scala

@@ -16,6 +16,6 @@ object ZioCsvParserSpec extends ZIOSpecDefault with ceesvee.CsvParserParserSuite
       .via(ZioCsvParser.parse(options))
       .map(_.toList)
       .runCollect
-      .mapError(e => new RuntimeException(s"failed to parse: $e"))
+      .mapError { case e: CsvParser.Error.LineTooLong => e }
   }
 }

modules/zio/src/test/scala/ceesvee/zio/ZioCsvParserVectorSpec.scala

@@ -18,6 +18,5 @@ object ZioCsvParserVectorSpec extends ZIOSpecDefault with ceesvee.CsvParserParse
       .via(ZioCsvParserVector.parse(options))
       .map(_.toList)
       .runCollect
-      .mapError(e => new RuntimeException(s"failed to parse: $e"))
   }
 }