Source.unfold
Stream the result of a function as long as it returns a Some
non empty Optional
.
Signature
Description
Stream the result of a function as long as it returns a Some
non empty Optional
. The value inside the option consists of a tuple pair where the first value is a state passed back into the next call to the function allowing to pass a state. The first invocation of the provided fold function will receive the zero
state.
The same zero
state object will be used for every materialization of the Source
so it is mandatory that the state is immutable. For example a java.util.Iterator
, Array
or Java standard library collection would not be safe as the fold operation could mutate the value. If you must use a mutable value, combining with Source.lazySource to make sure a new mutable zero
value is created for each materialization is one solution.
Note that for unfolding a source of elements through a blocking API, such as a network or filesystem resource you should prefer using unfoldResource.
Examples
This first sample starts at a user provided integer and counts down to zero using unfold
:
- Scala
-
source
def countDown(from: Int): Source[Int, NotUsed] = Source.unfold(from) { current => if (current == 0) None else Some((current - 1, current)) }
- Java
-
source
public static Source<Integer, NotUsed> countDown(Integer from) { return Source.unfold( from, current -> { if (current == 0) return Optional.empty(); else return Optional.of(Pair.create(current - 1, current)); }); }
It is also possible to express unfolds that don’t have an end, which will never return None
Optional.empty
and must be combined with for example .take(n)
to not produce infinite streams. Here we have implemented the Fibonacci numbers (0, 1, 1, 2, 3, 5, 8, 13, etc.) with unfold
:
- Scala
-
source
def fibonacci: Source[BigInt, NotUsed] = Source.unfold((BigInt(0), BigInt(1))) { case (a, b) => Some(((b, a + b), a)) }
- Java
-
source
public static Source<BigInteger, NotUsed> fibonacci() { return Source.unfold( Pair.create(BigInteger.ZERO, BigInteger.ONE), current -> { BigInteger a = current.first(); BigInteger b = current.second(); Pair<BigInteger, BigInteger> next = Pair.create(b, a.add(b)); return Optional.of(Pair.create(next, a)); }); }
Reactive Streams semantics
emits when there is demand and the unfold function over the previous state returns non empty value
completes when the unfold function returns an empty value