Skip to content

Commit

Permalink
feat: implement fold operator
Browse files Browse the repository at this point in the history
The `fold` operation returns 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 e.g.:

  Source.empty[Int].fold(0)((acc, n) => acc + n)       // 0
  Source.fromValues(2, 3).fold(5)((acc, n) => acc - n) // 0

Note that in case when `receive()` operation fails then:
* the original exception is re-thrown
* `NoSuchElement` exception is thrown when source fails without error
  • Loading branch information
geminicaprograms committed Oct 26, 2023
1 parent e28a268 commit 4bef471
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 0 deletions.
38 changes: 38 additions & 0 deletions core/src/main/scala/ox/channels/SourceOps.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
39 changes: 39 additions & 0 deletions core/src/test/scala/ox/channels/SourceOpsFoldTest.scala
Original file line number Diff line number Diff line change
@@ -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
}
}

0 comments on commit 4bef471

Please sign in to comment.