fold

Start with current value zero and then apply the current and next value to the given function. When upstream completes, the current value is emitted downstream.

Simple operators

Signature

def fold[T](zero: T)(f: (T, Out) => T): Repr[T]

Description

Start with current value zero and then apply the current and next value to the given function. When upstream completes, the current value is emitted downstream.

Warning

Note that the zero value must be immutable, because otherwise the same mutable instance would be shared across different threads when running the stream more than once.

Example

fold is typically used to ‘fold up’ the incoming values into an aggregate. For example, you might want to summarize the incoming values into a histogram:

Scala
import akka.actor.ActorSystem
import akka.stream.scaladsl.Source

case class Histogram(low: Long = 0, high: Long = 0) {
  def add(i: Int): Histogram = if (i < 100) copy(low = low + 1) else copy(high = high + 1)
}
Source(1 to 150).fold(Histogram())((acc, n) => acc.add(n)).runForeach(println)

// Prints: Histogram(99,51)
Java
class Histogram {
  final long low;
  final long high;

  private Histogram(long low, long high) {
    this.low = low;
    this.high = high;
  }

  // Immutable start value
  public static Histogram INSTANCE = new Histogram(0L, 0L);

  public Histogram add(int number) {
    if (number < 100) {
      return new Histogram(low + 1L, high);
    } else {
      return new Histogram(low, high + 1L);
    }
  }
}

  // Folding over the numbers from 1 to 150:
  Source.range(1, 150)
      .fold(Histogram.INSTANCE, (acc, n) -> acc.add(n))
      .runForeach(h -> System.out.println("Histogram(" + h.low + ", " + h.high + ")"), system);

  // Prints: Histogram(99, 51)

Reactive Streams semantics

emits when upstream completes

backpressures when downstream backpressures

completes when upstream completes

Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.