Add tap and tapAsView operators and document how to use them for …

…logging (#170)

core/src/main/scala/ox/channels/SourceOps.scala

@@ -28,6 +28,19 @@ trait SourceOps[+T] { outer: Source[T] =>
     override val delegate: JSource[Any] = outer.delegate.asInstanceOf[JSource[T]].collectAsView(t => f(t))
   }
 
+  /** Lazily-evaluated tap: creates a view of this source, where the results of [[receive]] will be applied to the given function `f` on the
+    * consumer's thread. Useful for side-effects without result values, like logging and debugging. For an eager, asynchronous version, see
+    * [[tap]].
+    *
+    * The same logic applies to receive clauses created using this source, which can be used in [[select]].
+    *
+    * @param f
+    *   The consumer function.
+    * @return
+    *   A source which is a view of this source, with the consumer function applied.
+    */
+  def tapAsView(f: T => Unit): Source[T] = mapAsView(t => { f(t); t })
+
   /** Lazily-evaluated filter: Creates a view of this source, where the results of [[receive]] will be filtered on the consumer's thread
     * using the given predicate `p`. For an eager, asynchronous version, see [[filter]].
     *
@@ -70,7 +83,7 @@ trait SourceOps[+T] { outer: Source[T] =>
   /** Applies the given mapping function `f` to each element received from this source, and sends the results to the returned channel.
     *
     * Errors from this channel are propagated to the returned channel. Any exceptions that occur when invoking `f` are propagated as errors
-    * to the returned channel as wel.
+    * to the returned channel as well.
     *
     * Must be run within a scope, as a child fork is created, which receives from this source and sends the mapped values to the resulting
     * one.
@@ -101,6 +114,23 @@ trait SourceOps[+T] { outer: Source[T] =>
     }
     c2
 
+  /** Applies the given consumer function `f` to each element received from this source.
+    *
+    * Errors from this channel are propagated to the returned channel. Any exceptions that occur when invoking `f` are propagated as errors
+    * to the returned channel as well.
+    *
+    * Must be run within a scope, as a child fork is created, which receives from this source and sends the mapped values to the resulting
+    * one.
+    *
+    * Useful for side-effects without result values, like logging and debugging. For a lazily-evaluated version, see [[tapAsView]].
+    *
+    * @param f
+    *   The consumer function.
+    * @return
+    *   A source, which the elements from the input source are passed to.
+    */
+  def tap(f: T => Unit)(using Ox, StageCapacity): Source[T] = map(t => { f(t); t })
+
   /** Intersperses this source with provided element and forwards it to the returned channel.
     *
     * @param inject

core/src/test/scala/ox/channels/SourceOpsAsViewTest.scala

@@ -5,6 +5,7 @@ import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import ox.*
 
+import java.util.concurrent.atomic.AtomicInteger
 import scala.jdk.CollectionConverters.*
 import scala.util.{Failure, Try}
 
@@ -111,6 +112,27 @@ class SourceOpsAsViewTest extends AnyFlatSpec with Matchers with Eventually {
     }
   }
 
+  it should "tap over a source as a view" in {
+    val c: Channel[Int] = Channel.rendezvous
+    val sum = new AtomicInteger()
+
+    supervised {
+      fork {
+        c.send(1)
+        c.send(2)
+        c.send(3)
+        c.done()
+      }
+
+      val s2 = c.tapAsView(v => sum.addAndGet(v).discard)
+      s2.receive() shouldBe 1
+      s2.receive() shouldBe 2
+      s2.receive() shouldBe 3
+      s2.receiveOrClosed() shouldBe ChannelClosed.Done
+      sum.get() shouldBe 6
+    }
+  }
+
   it should "propagate exceptions to the calling select" in {
     val c: Channel[Int] = Channel.rendezvous
 

core/src/test/scala/ox/channels/SourceOpsTest.scala

@@ -5,9 +5,11 @@ import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import ox.*
 
+import java.util.concurrent.atomic.AtomicInteger
 import scala.concurrent.duration.DurationInt
 
 class SourceOpsTest extends AnyFlatSpec with Matchers with Eventually {
+
   it should "timeout" in {
     supervised {
       val c = Source.timeout(100.millis)
@@ -65,4 +67,12 @@ class SourceOpsTest extends AnyFlatSpec with Matchers with Eventually {
       s.toList shouldBe List("a", "b", "c", "d", "e", "f", "g", "h", "i")
     }
   }
+
+  it should "tap over a source" in {
+    supervised {
+      val sum = new AtomicInteger()
+      Source.fromValues(1, 2, 3).tap(v => sum.addAndGet(v).discard).toList shouldBe List(1, 2, 3)
+      sum.get() shouldBe 6
+    }
+  }
 }

doc/channels/transforming-sources.md

@@ -80,3 +80,23 @@ When dealing with Sources with chunks of bytes or Strings, you can leverage foll
 * `decodeStringUtf8` to decode a `Source[Chunk[Byte]]` into a `Source[String]`, without handling line breaks, just processing input bytes as UTF-8 characters, even if a multi-byte character is divided into two chunks.
 
 Such operations may be useful when dealing with I/O like files, `InputStream`, etc.. See [examples here](io.md).
+
+## Logging
+
+Ox does not have any integrations with logging libraries, but it provides a simple way to log elements flowing through channels
+using the `.tap` (eagerly evaluated) or `.tapAsView` (lazily evaluated) methods.
+
+```scala mdoc:compile-only
+import ox.supervised
+import ox.channels.Source
+
+supervised {
+  Source.fromValues(1, 2, 3)
+    .tap(n => println(s"Received: $n")) // prints as soon as the element is sent from the source
+    .toList
+
+  Source.fromValues(1, 2, 3)
+    .tapAsView(n => println(s"Received: $n")) // prints when the element is consumed by `toList`
+    .toList
+}
+```