Source.unfold

Stream the result of a function as long as it returns a Some non empty Optional.

Source operators

Signature

Source.unfoldSource.unfold

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.

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

copysourcedef countDown(from: Int): Source[Int, NotUsed] =
  Source.unfold(from) { current =>
    if (current == 0) None
    else Some((current - 1, current))
  }

Java

copysourcepublic 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

copysourcedef fibonacci: Source[BigInt, NotUsed] =
  Source.unfold((BigInt(0), BigInt(1))) {
    case (a, b) =>
      Some(((b, a + b), a))
  }

Java

copysourcepublic 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