-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
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
- Loading branch information
1 parent
1529ba7
commit 7cb0e32
Showing
2 changed files
with
75 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 | ||
} | ||
} |