Modern Java

108. Stream Gatherers: Custom Intermediate Operations

Stream Gatherers enhance the Java Stream API with user-defined and built-in intermediate operations like windowFixed, windowSliding, fold, and scan, eliminating the historic limitations of the Stream pipeline.

Java 22 / 23 / 24 (JEP 461, 473, 485 Preview)

Explanation

Since Java 8, java.util.stream.Collector allowed customizing the terminal operation of a stream pipeline (stream.collect(...)). However, the set of intermediate operations (map, filter, flatMap, distinct) was completely closed and hardcoded into the JDK. Writing custom stream operations (like sliding windows, grouping adjacent items, or stateful transformations) required either abandoning streams or collecting to an intermediate list.

Stream Gatherers (JEP 485) provide Stream.gather(Gatherer) — the intermediate counterpart to Collector. A Gatherer defines: - Initializer: Creates internal private state. - Integrator: Consumes incoming stream elements, inspects/updates state, and optionally pushes elements downstream. - Finisher: Performs cleanup and pushes final aggregated elements when the stream ends.

The JDK ships with standard built-in gatherers in java.util.stream.Gatherers: - windowFixed(n): Groups elements into non-overlapping lists of size n. - windowSliding(n): Produces overlapping sliding windows of size n. - fold(initial, folder): Incremental reduction into a downstream stream. - scan(initial, scanner): Cumulative running totals or prefix sums.

Example

import java.util.List;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class StreamGatherersDemo {
    public static void main(String[] args) {
        List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7);

        // 1. Fixed-size batching / chunking: windowFixed(3)
        System.out.println("--- Fixed Windows (Chunk size 3) ---");
        Stream.of(1, 2, 3, 4, 5, 6, 7)
            .gather(Gatherers.windowFixed(3))
            .forEach(System.out::println);
        // Prints: [1, 2, 3], [4, 5, 6], [7]

        // 2. Sliding window: windowSliding(3)
        System.out.println("--- Sliding Windows (Window size 3) ---");
        Stream.of(1, 2, 3, 4, 5)
            .gather(Gatherers.windowSliding(3))
            .forEach(System.out::println);
        // Prints: [1, 2, 3], [2, 3, 4], [3, 4, 5]

        // 3. Scan (Cumulative Running Sum)
        System.out.println("--- Running Prefix Sum ---");
        Stream.of(10, 20, 30, 40)
            .gather(Gatherers.scan(() -> 0, (acc, next) -> acc + next))
            .forEach(System.out::println);
        // Prints: 10, 30, 60, 100
    }
}

Sources