grouped

Accumulate incoming events until the specified number of elements have been accumulated and then pass the collection of elements downstream.

Simple operators

Signature

Source.groupedSource.grouped Flow.groupedFlow.grouped

Description

Accumulate incoming events until the specified number of elements have been accumulated and then pass the collection of elements downstream.

See also:

• groupedWeighted for a variant that groups based on element weight
• groupedWithin for a variant that groups based on number of elements and a time window
• groupedWeightedWithin for a variant that groups based on element weight and a time window

Examples

The below example demonstrates how grouped groups the accumulated elements into Seq List and maps with other operation.

Scala

copysourceSource(1 to 7).grouped(3).runForeach(println)
// Vector(1, 2, 3)
// Vector(4, 5, 6)
// Vector(7)

Source(1 to 7).grouped(3).map(_.sum).runForeach(println)
// 6   (= 1 + 2 + 3)
// 15  (= 4 + 5 + 6)
// 7   (= 7)

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7))
    .grouped(3)
    .runForeach(System.out::println, system);
// [1, 2, 3]
// [4, 5, 6]
// [7]

Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7))
    .grouped(3)
    .map(g -> g.stream().reduce(0, Integer::sum))
    .runForeach(System.out::println, system);
// 6   (= 1 + 2 + 3)
// 15  (= 4 + 5 + 6)
// 7   (= 7)

Reactive Streams semantics