Added debug and debugChunks combinators

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

@@ -656,6 +656,69 @@ final class Stream[+F[_], +O] private (private val free: FreeC[Nothing, O, Unit]
   def metered[F2[x] >: F[x]: Timer](rate: FiniteDuration): Stream[F2, O] =
     Stream.fixedRate[F2](rate).zipRight(this)
 
+  /**
+    * Logs the elements of this stream as they are pulled.
+    *
+    * If a tag is specified, it prefixes each element using the format `$tag: $o`.
+    * By default, log output is sent to standard out. Supply a different logger param
+    * to change this.
+    *
+    * This method does not change the chunk structure of the stream. To debug the
+    * chunk structure, see [[debugChunks]].
+    *
+    * Logging is not done in `F` because this operation is intended for debugging,
+    * including pure streams.
+    *
+    * @example {{{
+    * scala> Stream(1, 2).append(Stream(3, 4)).debug("a").toList
+    * a: 1
+    * a: 2
+    * a: 3
+    * a: 4
+    * res0: List[Int] = List(1, 2, 3, 4)
+    * }}}
+    */
+  def debug(tag: String = "", logger: String => Unit = println(_)): Stream[F, O] =
+    if (tag.isEmpty) debugFormat(_.toString, logger) else debugFormat(o => s"$tag: $o", logger)
+
+  /**
+    * Like [[debug]] but allows custom formatting of each element.
+    */
+  def debugFormat(formatter: O => String, logger: String => Unit = println(_)): Stream[F, O] =
+    map { o =>
+      logger(formatter(o))
+      o
+    }
+
+  /**
+    * Like [[debug]] but logs chunks as they are pulled instead of individual elements.
+    *
+    * @example {{{
+    * scala> Stream(1, 2, 3).append(Stream(4, 5, 6)).debugChunks("a").buffer(2).debugChunks("b").toList
+    * a: Chunk(1, 2, 3)
+    * b: Chunk(1, 2)
+    * a: Chunk(4, 5, 6)
+    * b: Chunk(3, 4)
+    * b: Chunk(5, 6)
+    * res0: List[Int] = List(1, 2, 3, 4, 5, 6)
+    * }}}
+    */
+  def debugChunks(tag: String = "", logger: String => Unit = println(_)): Stream[F, O] =
+    if (tag.isEmpty) debugChunksFormat(_.toString, logger)
+    else debugChunksFormat(o => s"$tag: $o", logger)
+
+  /**
+    * Like [[debugChunks]] but allows custom formatting of each chunk.
+    */
+  def debugChunksFormat(
+      formatter: Chunk[O] => String,
+      logger: String => Unit = println(_)
+  ): Stream[F, O] =
+    chunks.flatMap { os =>
+      logger(formatter(os))
+      Stream.chunk(os)
+    }
+
   /**
     * Returns a stream that when run, sleeps for duration `d` and then pulls from this stream.
     *