diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index c2a62240..fcd634cc 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -565,6 +565,44 @@ trait SourceOps[+T] { this: Source[T] => case ChannelClosed.Error(r) => throw r.getOrElse(new NoSuchElementException("getting head failed")) case t: T @unchecked => t } + + /** Uses `zero` as the current value and applies function `f` on it and a value received from a source. 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. + * + * @param zero + * An initial value to be used as the first argument to function `f` call. + * @param f + * A binary function (a function that takes two arguments) that is applied to the current value and value 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 `receive()` failed without error. + * @throws exception + * When `receive()` failed with exception then this exception is re-thrown. + * @example + * {{{ + * import ox.* + * import ox.channels.Source + * + * scoped { + * Source.empty[Int].fold(0)((acc, n) => acc + n) // 0 + * Source.fromValues(2, 3).fold(5)((acc, n) => acc - n) // 0 + * } + * }}} + */ + def fold[U](zero: U)(f: (U, T) => U): U = + supervised { + var current = zero + repeatWhile { + receive() match + case ChannelClosed.Done => false + case ChannelClosed.Error(r) => throw r.getOrElse(new NoSuchElementException("folding failed")) + case t: T @unchecked => current = f(current, t); true + } + current + } } trait SourceCompanionOps: diff --git a/core/src/test/scala/ox/channels/SourceOpsFoldTest.scala b/core/src/test/scala/ox/channels/SourceOpsFoldTest.scala new file mode 100644 index 00000000..a890f5dc --- /dev/null +++ b/core/src/test/scala/ox/channels/SourceOpsFoldTest.scala @@ -0,0 +1,39 @@ +package ox.channels + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import ox.* + +class SourceOpsFoldTest extends AnyFlatSpec with Matchers { + behavior of "Source.fold" + + it should "re-throw exception that was thrown during fold performance" in supervised { + the[RuntimeException] thrownBy { + Source + .failed[Int](new RuntimeException("source is broken")) + .fold(0)((acc, n) => acc + n) + } should have message "source is broken" + } + + it should "throw NoSuchElementException for source failed without exception" in supervised { + the[NoSuchElementException] thrownBy { + Source + .failedWithoutReason[Int]() + .fold(0)((acc, n) => acc + n) + } should have message "folding failed" + } + + it should "return `zero` value from fold on the empty source" in supervised { + Source.empty[Int].fold(0)((acc, n) => acc + n) shouldBe 0 + } + + it should "return fold on non-empty source" in supervised { + Source.fromValues(1, 2).fold(0)((acc, n) => acc + n) shouldBe 3 + } + + it should "drain the source" in supervised { + val s = Source.fromValues(1) + s.fold(0)((acc, n) => acc + n) shouldBe 1 + s.receive() shouldBe ChannelClosed.Done + } +}