From 7cb0e32b28e69fbe1823730e4bd068bc46df19af Mon Sep 17 00:00:00 2001
From: Jacek Centkowski <jacek.centkowski@softwaremill.com>
Date: Fri, 27 Oct 2023 17:42:28 +0200
Subject: [PATCH] feat: implement `fold` operator

Uses the first and the following (if available) elements from this source
and applies function `f` on them. The returned value is used as the next
current value and `f` is applied again with the value received from a source.
The operation is repeated until the source is drained.
This is similar operation to `fold` but it uses the first source element as
`zero` e.g.:

  Source.empty[Int].reduce(_ + _)    // throws NoSuchElementException("cannot reduce over an empty source")
  Source.fromValues(1).reduce(_ + _) // 1
  val s = Source.fromValues(1, 2)
  s.reduce(_ + _)                    // 3
  s.receive()                        // ChannelClosed.Done
---
 .../main/scala/ox/channels/SourceOps.scala    | 30 +++++++++++++
 .../ox/channels/SourceOpsReduceTest.scala     | 45 +++++++++++++++++++
 2 files changed, 75 insertions(+)
 create mode 100644 core/src/test/scala/ox/channels/SourceOpsReduceTest.scala

diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala
index 1c3dcf9c..597bb157 100644
--- a/core/src/main/scala/ox/channels/SourceOps.scala
+++ b/core/src/main/scala/ox/channels/SourceOps.scala
@@ -659,6 +659,36 @@ trait SourceOps[+T] { this: Source[T] =>
       }
       current
     }
+
+  /** Uses the first and the following (if available) elements from this source and applies function `f` on them. The returned value is used
+    * as the next current value and `f` is applied again with the value received from a source. The operation is repeated until the source
+    * is drained. This is similar operation to [[fold]] but it uses the first source element as `zero`.
+    *
+    * @param f
+    *   A binary function (a function that takes two arguments) that is applied to the current and next values received from a source.
+    * @return
+    *   Combined value retrieved from running function `f` on all source elements in a cumulative manner where result of the previous call
+    *   is used as an input value to the next.
+    * @throws NoSuchElementException
+    *   When source is empty.
+    * @throws ChannelClosedException.Error
+    *   When [[receive]] fails.
+    * @example
+    *   {{{
+    *   import ox.*
+    *   import ox.channels.Source
+    *
+    *   supervised {
+    *     Source.empty[Int].reduce(_ + _)    // throws NoSuchElementException("cannot reduce over an empty source")
+    *     Source.fromValues(1).reduce(_ + _) // 1
+    *     val s = Source.fromValues(1, 2)
+    *     s.reduce(_ + _)                    // 3
+    *     s.receive()                        // ChannelClosed.Done
+    *   }
+    *   }}}
+    */
+  def reduce[U >: T](f: (U, U) => U): U =
+    fold(headOption().getOrElse(throw new NoSuchElementException("cannot reduce over an empty source")))(f)
 }
 
 trait SourceCompanionOps:
diff --git a/core/src/test/scala/ox/channels/SourceOpsReduceTest.scala b/core/src/test/scala/ox/channels/SourceOpsReduceTest.scala
new file mode 100644
index 00000000..df23000f
--- /dev/null
+++ b/core/src/test/scala/ox/channels/SourceOpsReduceTest.scala
@@ -0,0 +1,45 @@
+package ox.channels
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.*
+
+class SourceOpsReduceTest extends AnyFlatSpec with Matchers {
+  behavior of "SourceOps.reduce"
+
+  it should "throw NoSuchElementException for reduce over the empty source" in supervised {
+    the[NoSuchElementException] thrownBy {
+      Source.empty[Int].reduce(_ + _)
+    } should have message "cannot reduce over an empty source"
+  }
+
+  it should "throw ChannelClosedException.Error with exception and message that was thrown during retrieval" in supervised {
+    the[ChannelClosedException.Error] thrownBy {
+      Source
+        .failed[Int](new RuntimeException("source is broken"))
+        .reduce(_ + _)
+    } should have message "java.lang.RuntimeException: source is broken"
+  }
+
+  it should "throw ChannelClosedException.Error for source failed without exception" in supervised {
+    the[ChannelClosedException.Error] thrownBy {
+      Source
+        .failedWithoutReason[Int]()
+        .reduce(_ + _)
+    }
+  }
+
+  it should "return first element from reduce over the single element source" in supervised {
+    Source.fromValues(1).reduce(_ + _) shouldBe 1
+  }
+
+  it should "run reduce over on non-empty source" in supervised {
+    Source.fromValues(1, 2).reduce(_ + _) shouldBe 3
+  }
+
+  it should "drain the source" in supervised {
+    val s = Source.fromValues(1)
+    s.reduce(_ + _) shouldBe 1
+    s.receive() shouldBe ChannelClosed.Done
+  }
+}