feat(tail) Iterant.dump

monix-reactive/shared/src/main/scala/monix/reactive/Observer.scala

@@ -20,8 +20,8 @@ package monix.reactive
 import java.io.PrintStream
 
 import monix.execution.Ack.{Continue, Stop}
-import monix.execution.cancelables.BooleanCancelable
 import monix.execution._
+import monix.execution.cancelables.BooleanCancelable
 import monix.execution.misc.NonFatal
 import monix.reactive.internal.rstreams._
 import monix.reactive.observers.Subscriber
@@ -327,13 +327,13 @@ object Observer {
     private[this] var pos = 0
 
     def onNext(elem: A): Ack = {
-      out.println(s"$pos: $prefix-->$elem")
+      out.println(s"$pos: $prefix --> $elem")
       pos += 1
       Continue
     }
 
     def onError(ex: Throwable) = {
-      out.println(s"$pos: $prefix-->$ex")
+      out.println(s"$pos: $prefix --> $ex")
       pos += 1
     }
 

monix-reactive/shared/src/main/scala/monix/reactive/internal/operators/DumpObservable.scala

@@ -43,7 +43,7 @@ class DumpObservable[A](source: Observable[A], prefix: String, out: PrintStream)
 
       def onNext(elem: A): Future[Ack] = {
         try {
-          out.println(s"$pos: $prefix-->$elem")
+          out.println(s"$pos: $prefix --> $elem")
           pos += 1
         } catch {
           case NonFatal(_) => () // ignore
@@ -54,7 +54,7 @@ class DumpObservable[A](source: Observable[A], prefix: String, out: PrintStream)
 
       def onError(ex: Throwable) = {
         try {
-          out.println(s"$pos: $prefix-->$ex")
+          out.println(s"$pos: $prefix --> $ex")
           pos += 1
         } catch {
           case NonFatal(_) => () // ignore

monix-tail/shared/src/main/scala/monix/tail/Iterant.scala

@@ -17,6 +17,8 @@
 
 package monix.tail
 
+import java.io.PrintStream
+
 import cats.arrow.FunctionK
 import cats.effect.{Effect, Sync}
 import cats.{Applicative, CoflatMap, Eq, Monoid, MonoidK, Order}
@@ -493,6 +495,27 @@ sealed abstract class Iterant[F[_], A] extends Product with Serializable {
   final def dropWhile(p: A => Boolean)(implicit F: Sync[F]): Iterant[F, A] =
     IterantDropWhile(self, p)
 
+  /** Dumps incoming events to standard output with provided prefix.
+    *
+    * Utility that can be used for debugging purposes.
+    *
+    * Example: {{{
+    *   Iterant[Task].range(0, 4)
+    *     .dump("O")
+    *     .completeL.runAsync
+    *
+    *   // Results in:
+    *
+    *   0: O --> 0
+    *   1: O --> 1
+    *   2: O --> 2
+    *   3: O --> 3
+    *   4: O completed
+    * }}}
+    */
+  final def dump(prefix: String, out: PrintStream = System.out)(implicit F: Sync[F]): Iterant[F, A] =
+    IterantDump(this, prefix, out)
+
   /** Returns a computation that should be evaluated in case the
     * streaming must stop before reaching the end.
     *

monix-tail/shared/src/main/scala/monix/tail/internal/IterantDump.scala

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2014-2017 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * Licensed 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
+ *
+ *     http://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 monix.tail.internal
+
+import java.io.PrintStream
+
+import cats.effect.Sync
+import cats.syntax.all._
+import monix.execution.misc.NonFatal
+import monix.tail.Iterant
+import monix.tail.Iterant.{Halt, Last, Next, NextBatch, NextCursor, Suspend}
+import monix.tail.internal.IterantUtils.signalError
+
+private[tail] object IterantDump {
+  /**
+    * Implementation for `Iterant#dump`
+    */
+  def apply[F[_], A](source: Iterant[F, A], prefix: String, out: PrintStream = System.out)
+                    (implicit F: Sync[F]): Iterant[F, A] = {
+    var pos = 0
+
+    def loop(source: Iterant[F, A]): Iterant[F, A] =
+      try source match {
+        case Next(item, rest, stop) =>
+          out.println(s"$pos: $prefix --> $item")
+          pos += 1
+          Next[F, A](item, rest.map(loop), stop)
+
+        case NextCursor(cursor, rest, stop) =>
+          val dumped = cursor.map { el =>
+            out.println(s"$pos: $prefix --> $el")
+            pos += 1
+            el
+          }
+          NextCursor[F, A](dumped, rest.map(loop), stop)
+
+        case NextBatch(batch, rest, stop) =>
+          val dumped = batch.map { el =>
+            out.println(s"$pos: $prefix --> $el")
+            pos += 1
+            el
+          }
+          NextBatch[F, A](dumped, rest.map(loop), stop)
+
+        case Suspend(rest, stop) =>
+          Suspend[F, A](rest.map(loop), stop)
+
+        case Last(item) =>
+          out.println(s"$pos: $prefix --> $item")
+          out.println(s"${pos + 1}: $prefix completed")
+          Last(item)
+
+        case empty@Halt(_) =>
+          out.println(s"$pos: $prefix completed")
+          empty.asInstanceOf[Iterant[F, A]]
+
+      } catch {
+        case NonFatal(ex) =>
+          out.println(s"$pos: $prefix --> $ex")
+          signalError(source, ex)
+      }
+
+    source match {
+      case Suspend(_, _) | Halt(_) => loop(source)
+      case _ =>
+        // Suspending execution in order to preserve laziness and
+        // referential transparency, since the provided PrintStream can
+        // be side effecting and because processing NextBatch and
+        // NextCursor states can have side effects
+        Suspend(F.delay(loop(source)), source.earlyStop)
+    }
+  }
+}

monix-tail/shared/src/test/scala/monix/tail/IterantDumpSuite.scala

@@ -0,0 +1,170 @@
+/*
+ * Copyright (c) 2014-2017 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * Licensed 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
+ *
+ *     http://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 monix.tail
+
+import java.io.{OutputStream, PrintStream}
+
+import monix.eval.{Coeval, Task}
+import monix.execution.atomic.AtomicInt
+import monix.execution.exceptions.DummyException
+import monix.tail.batches.{Batch, BatchCursor}
+
+object IterantDumpSuite extends BaseTestSuite {
+  def dummyOut(count: AtomicInt = null) = {
+    val out = new OutputStream {
+      def write(b: Int) = ()
+    }
+    new PrintStream(out) {
+      override def println(x: String) = {
+        super.println(x)
+        if (count != null) {
+          val c = count.incrementAndGet()
+          if (c == 0) throw DummyException("dummy")
+        }
+      }
+    }
+  }
+
+  def dummyOutException = {
+    val out = new OutputStream {
+      def write(b: Int) = ()
+    }
+    new PrintStream(out) {
+      override def println(x: String) = {
+        throw DummyException("dummy")
+      }
+    }
+  }
+
+  test("Iterant.dump works for Next") { implicit s =>
+    check1 { (el: Int) =>
+      val counter = AtomicInt(0)
+      val out = Iterant[Task].nextS(el, Task.now(Iterant[Task].empty[Int]), Task.unit).dump("O", dummyOut(counter))
+      out.completeL.runAsync
+      s.tick()
+
+      counter.get <-> 2
+    }
+  }
+
+  test("Iterant.dump works for NextCursor") { implicit s =>
+    check1 { (el: Int) =>
+      val counter = AtomicInt(0)
+      val out = Iterant[Task].nextCursorS(BatchCursor(el), Task.now(Iterant[Task].empty[Int]), Task.unit).dump("O", dummyOut(counter))
+      out.completeL.runAsync
+      s.tick()
+
+      counter.get <-> 2
+    }
+  }
+
+  test("Iterant.dump works for NextBatch") { implicit s =>
+    check1 { (el: Int) =>
+      val counter = AtomicInt(0)
+      val out = Iterant[Task].nextBatchS(Batch(el), Task.now(Iterant[Task].empty[Int]), Task.unit).dump("O", dummyOut(counter))
+      out.completeL.runAsync
+      s.tick()
+
+      counter.get <-> 2
+    }
+  }
+
+  test("Iterant.dump works for Suspend") { implicit s =>
+    val counter = AtomicInt(0)
+    val out = Iterant[Task].suspend(Task.now(Iterant[Task].empty[Int])).dump("O", dummyOut(counter))
+    out.completeL.runAsync
+    s.tick()
+
+    assertEquals(counter.get, 1)
+  }
+
+  test("Iterant.dump works for Last") { implicit s =>
+    check1 { (el: Int) =>
+      val counter = AtomicInt(0)
+      val out = Iterant[Task].lastS(el).dump("O", dummyOut(counter))
+      out.completeL.runAsync
+      s.tick()
+
+      counter.get <-> 2
+    }
+  }
+
+  test("Iterant.dump works for Halt") { implicit s =>
+    val dummy = DummyException("dummy")
+    val counter = AtomicInt(0)
+    val out = Iterant[Task].haltS(Some(dummy)).dump("O", dummyOut(counter))
+    out.completeL.runAsync
+    s.tick()
+
+    assertEquals(counter.get, 1)
+  }
+
+  test("Iterant.dump doesn't touch Halt") { _ =>
+    val dummy = DummyException("dummy")
+    val stream: Iterant[Coeval, Int] = Iterant[Coeval].haltS[Int](Some(dummy))
+    val state = stream.dump("O", dummyOut(AtomicInt(0)))
+
+    assertEquals(state, stream)
+  }
+
+  test("Iterant.dump preserves the source earlyStop") { implicit s =>
+    var effect = 0
+    val stop = Coeval.eval(effect += 1)
+    val source = Iterant[Coeval].nextCursorS(BatchCursor(1, 2, 3), Coeval.now(Iterant[Coeval].empty[Int]), stop)
+    val stream = source.dump("O", dummyOut(AtomicInt(0)))
+    stream.earlyStop.value
+
+    assertEquals(effect, 1)
+  }
+
+  test("Iterant[Task].dump can handle errors") { implicit s =>
+    val dummy = DummyException("dummy")
+    val stream = Iterant[Task].raiseError[Int](dummy)
+
+    assertEquals(stream, stream.dump("O", dummyOut(AtomicInt(0))))
+  }
+
+  test("Iterant.dump protects against broken batches") { implicit s =>
+    check1 { (iter: Iterant[Task, Int]) =>
+      val dummy = DummyException("dummy")
+      val suffix = Iterant[Task].nextBatchS[Int](new ThrowExceptionBatch(dummy), Task.now(Iterant[Task].empty), Task.unit)
+      val stream = iter.onErrorIgnore ++ suffix
+
+      stream.dump("O", dummyOut(AtomicInt(0))) <-> Iterant[Task].haltS[Int](Some(dummy))
+    }
+  }
+
+  test("Iterant.dump protects against broken cursors") { implicit s =>
+    check1 { (iter: Iterant[Task, Int]) =>
+      val dummy = DummyException("dummy")
+      val suffix = Iterant[Task].nextCursorS[Int](new ThrowExceptionCursor(dummy), Task.now(Iterant[Task].empty), Task.unit)
+      val stream = iter.onErrorIgnore ++ suffix
+
+      stream.dump("O", dummyOut(AtomicInt(0))) <-> Iterant[Task].haltS[Int](Some(dummy))
+    }
+  }
+
+  test("Iterant.dump protects against user error") { implicit s =>
+    check1 { (stream: Iterant[Task, Int]) =>
+      val dummy = DummyException("dummy")
+      val received = (stream.onErrorIgnore ++ Iterant[Task].now(1)).dump("O", dummyOutException)
+
+      received <-> Iterant[Task].raiseError(dummy)
+    }
+  }
+}