text.lines enhancements

core/shared/src/main/scala/fs2/text.scala

@@ -316,14 +316,34 @@ object text {
   def utf8EncodeC[F[_]]: Pipe[F, String, Chunk[Byte]] =
     utf8.encodeC
 
+  /** Transforms a stream of `String` such that each emitted `String` is a line from the input
+    * @param maxLineLength maximum size to accumulate a line to; throw an error if a line is larger
+    * @param crsOnly separate lines that are delimited only by '\r'
+    * @tparam F
+    */
+  def linesFor[F[_]: RaiseThrowable](
+      maxLineLength: Option[Int] = None,
+      crsOnly: Boolean = false
+  ): Pipe[F, String, String] =
+    linesImpl[F](
+      maxLineLength = maxLineLength.map((_, implicitly[RaiseThrowable[F]])),
+      crsOnly = crsOnly
+    )
+
   /** Transforms a stream of `String` such that each emitted `String` is a line from the input. */
-  def lines[F[_]]: Pipe[F, String, String] = {
+  def lines[F[_]]: Pipe[F, String, String] = linesImpl[F]()
+
+  private def linesImpl[F[_]](
+      maxLineLength: Option[(Int, RaiseThrowable[F])] = None,
+      crsOnly: Boolean = false
+  ): Pipe[F, String, String] = {
     def fillBuffers(
         stringBuilder: StringBuilder,
         linesBuffer: ArrayBuffer[String],
         string: String
     ): Unit = {
       val l = stringBuilder.length
+
       var i =
         if (l > 0 && stringBuilder(l - 1) == '\r' && string.nonEmpty && string(0) == '\n') {
           stringBuilder.deleteCharAt(l - 1)
@@ -341,6 +361,9 @@ object text {
             linesBuffer += stringBuilder.result()
             stringBuilder.clear()
             i += 1
+          case '\r' if crsOnly =>
+            linesBuffer += stringBuilder.result()
+            stringBuilder.clear()
           case other =>
             stringBuilder.append(other)
         }
@@ -360,12 +383,25 @@ object text {
           chunk.foreach { string =>
             fillBuffers(stringBuilder, linesBuffer, string)
           }
-          Pull.output(Chunk.buffer(linesBuffer)) >> go(stream, stringBuilder, first = false)
+
+          maxLineLength match {
+            case Some((max, raiseThrowable)) if stringBuilder.length > max =>
+              Pull.raiseError[F](
+                new LineTooLongException(stringBuilder.length, max)
+              )(raiseThrowable)
+            case _ =>
+              Pull.output(Chunk.indexedSeq(linesBuffer)) >> go(stream, stringBuilder, first = false)
+          }
       }
 
     s => Stream.suspend(go(s, new StringBuilder(), first = true).stream)
   }
 
+  class LineTooLongException(val length: Int, val max: Int)
+      extends RuntimeException(
+        s"Max line size is $max but $length chars have been accumulated"
+      )
+
   /** Functions for working with base 64. */
   object base64 {
 

core/shared/src/test/scala/fs2/TextSuite.scala

@@ -247,7 +247,7 @@ class TextSuite extends Fs2Suite {
     }
   }
 
-  group("lines") {
+  group("lines / linesFor") {
     def escapeCrLf(s: String): String =
       s.replaceAll("\r\n", "<CRLF>").replaceAll("\n", "<LF>").replaceAll("\r", "<CR>")
 
@@ -256,6 +256,16 @@ class TextSuite extends Fs2Suite {
         val lines = lines0.map(escapeCrLf)
         assertEquals(lines.intersperse("\n").through(text.lines).toList, lines.toList)
         assertEquals(lines.intersperse("\r\n").through(text.lines).toList, lines.toList)
+        assertEquals(
+          lines
+            .intersperse("\r")
+            .covary[Fallible]
+            .through(
+              text.linesFor[Fallible](crsOnly = true)
+            )
+            .toList,
+          Right(lines.toList)
+        )
       }
     }
 
@@ -269,7 +279,7 @@ class TextSuite extends Fs2Suite {
       }
     }
 
-    property("grouped in 3 characater chunks") {
+    property("grouped in 3 character chunks") {
       forAll { (lines0: Stream[Pure, String]) =>
         val lines = lines0.map(escapeCrLf)
         val s = lines.intersperse("\r\n").toList.mkString.grouped(3).toList
@@ -284,6 +294,23 @@ class TextSuite extends Fs2Suite {
         }
       }
     }
+
+    property("linesFor with maxLineLength") {
+      val line = "foo" * 100
+      (1 to line.length).foreach { i =>
+        val stream = Stream
+          .emits(line.toCharArray)
+          .chunkN(i)
+          .map(c => new String(c.toArray))
+          .covary[Fallible]
+
+        assert(stream.through(text.linesFor(maxLineLength = Some(10))).toList.isLeft)
+        assertEquals(
+          stream.through(text.linesFor(maxLineLength = Some(line.length))).toList,
+          Right(List(line))
+        )
+      }
+    }
   }
 
   property("base64.encode") {