groupBy

Demultiplex the incoming stream into separate output streams.

Nesting and flattening operators

Signature

Source.groupBySource.groupBy Flow.groupByFlow.groupBy

Description

This operation demultiplexes the incoming stream into separate output streams, one for each element key. The key is computed for each element using the given function. When a new key is encountered for the first time a new substream is opened and subsequently fed with all elements belonging to that key.

Note: If allowClosedSubstreamRecreation is set to true substream completion and incoming elements are subject to race-conditions. If elements arrive for a stream that is in the process of closing these elements might get lost.

Example

Scala

copysourceSource(1 to 10)
  .groupBy(maxSubstreams = 2, _ % 2) // create two sub-streams with odd and even numbers
  .reduce(_ + _) // for each sub-stream, sum its elements
  .mergeSubstreams // merge back into a stream
  .runForeach(println)
//30
//25

Java

copysourceSource.range(1, 10)
    .groupBy(2, i -> i % 2 == 0) // create two sub-streams with odd and even numbers
    .reduce(Integer::sum) // for each sub-stream, sum its elements
    .mergeSubstreams() // merge back into a stream
    .runForeach(System.out::println, system);
// 25
// 30

Note: groupBy operator is used to partition the stream based on an identifier which is not same as using async boundary in the stream which is used for running the same stream concurrently. If a stream can be exclusively partitioned, it can be executed efficiently, by maximizing the parallel processing using groupBy operator.

Example with async boundary that introduces running concurrently :

Scala

copysourceSource(1 to 10)
  .groupBy(maxSubstreams = 2, _ % 2) // create two sub-streams with odd and even numbers
  .via(Flow[Int].map(_ => 1).reduce(_ + _).async) // for each sub-stream, sum its elements
  .mergeSubstreams // merge back into a stream
  .runForeach(println)
//30
//25

Java

copysourceSource.range(1, 10)
        .groupBy(2, i -> i % 2 == 0) // create two sub-streams with odd and even numbers
        .via(Flow.of(Integer.class).map(elem-> 1).reduce(Integer::sum).async()) // for each sub-stream, sum its elements
        .mergeSubstreams() // merge back into a stream
        .runForeach(System.out::println, system);
// 25
// 30

Reactive Streams semantics