Stream Gatherers: Custom Intermediate Operations for Infinite Flexibility

Content

• Introduction
• The Problem: Fixed Intermediate Operations
• The Solution: Stream Gatherers
• Technical Deep Dive: Defining Custom Gatherers
• Parallel Evaluation and Combiners
• Practical Examples
• Composing Gatherers
• Performance Considerations
• Migration Path and Best Practices
• Conclusions
• References

Introduction

The Stream API revolutionized Java programming when it arrived in Java 8. With operations like map, filter, reduce, and collect, developers gained a powerful toolkit for processing collections declaratively. Yet despite its expressiveness, the Stream API has always been limited by a fixed set of intermediate operations. If you need something beyond filter or flatMap, you’re forced into uncomfortable workarounds: breaking out of the stream paradigm, writing imperative loops, or contorting your logic to fit existing operations.

JEP 485 changes this fundamentally by introducing Stream Gatherers — custom intermediate operations that transform data in ways previously impossible. Think of Collector for terminal operations; now we have Gatherer for intermediate operations. This single extension point unlocks:

• Fixed-size windowing: Group elements into batches of N
• Sliding windows: Detect patterns across consecutive elements
• Concurrent mapping: Apply transformations using virtual threads
• Prefix scans: Cumulative operations (running totals, string concatenation)
• Custom stateful logic: Track previous elements to influence future transformations

The built-in gatherers (windowFixed, windowSliding, scan, fold, mapConcurrent) handle common use cases, but the real power comes from defining your own. The API is designed for both simplicity (ad-hoc gatherers via factory methods) and flexibility (full Gatherer interface implementation with parallel support).

For developers tired of abandoning streams when complexity increases, JEP 485 delivers the missing piece: infinite flexibility without sacrificing declarative style.

The Problem: Fixed Intermediate Operations

The Stream API provides intermediate operations like filter, map, flatMap, distinct, sorted, limit, and skip. These cover many common transformations, but real-world problems often require something slightly different.

Example 1: Distinctness by Attribute

Suppose you have a stream of strings and want distinctness based on string length rather than content. At most one string of length 1, one of length 2, etc. Ideally:

var result = Stream.of("foo", "bar", "baz", "quux")
                   .distinctBy(String::length)  // Hypothetical!
                   .toList();

// result => [foo, quux]  (lengths 3 and 4)

But distinctBy doesn’t exist. The built-in distinct() uses object equality (equals() and hashCode()), not a custom key extractor. The workaround? Wrap each string in a record with custom equality:

record DistinctByLength(String str) {
    @Override public boolean equals(Object obj) {
        return obj instanceof DistinctByLength(String other)
               && str.length() == other.length();
    }
    
    @Override public int hashCode() {
        return str == null ? 0 : Integer.hashCode(str.length());
    }
}

var result = Stream.of("foo", "bar", "baz", "quux")
                   .map(DistinctByLength::new)
                   .distinct()
                   .map(DistinctByLength::str)
                   .toList();

This works but obscures intent. Readers must decode the wrapper record to understand the business logic.

Example 2: Fixed-Size Grouping with Short-Circuiting

Group integers into batches of 3, but stop after 2 batches:

var result = Stream.iterate(0, i -> i + 1)
                   .windowFixed(3)         // Hypothetical!
                   .limit(2)
                   .toList();

// result => [[0, 1, 2], [3, 4, 5]]

No built-in operation supports fixed-size windowing. The best workaround is a custom Collector, but collectors can’t signal completion. For infinite streams, you must use limit before collecting, calculating the exact number of elements needed:

var result = Stream.iterate(0, i -> i + 1)
                   .limit(3 * 2)  // Pre-calculate: 3 elements × 2 windows
                   .collect(Collector.of(
                       () -> new ArrayList<ArrayList<Integer>>(),
                       (groups, element) -> {
                           if (groups.isEmpty() || groups.getLast().size() == 3) {
                               var current = new ArrayList<Integer>();
                               current.add(element);
                               groups.addLast(current);
                           } else {
                               groups.getLast().add(element);
                           }
                       },
                       (left, right) -> {
                           throw new UnsupportedOperationException("Cannot parallelize");
                       }
                   ));

This code is verbose, fragile (the 3 * 2 calculation), and not parallelizable. For ordered windowing, the combiner must throw an exception.

Example 3: Detecting Patterns Across Consecutive Elements

You have temperature readings and want to detect suspicious changes (> 30°K within 5 seconds):

record Reading(Instant obtainedAt, int kelvins) {}

boolean isSuspicious(Reading prev, Reading next) {
    return next.obtainedAt().isBefore(prev.obtainedAt().plusSeconds(5))
           && Math.abs(next.kelvins() - prev.kelvins()) > 30;
}

This requires comparing consecutive elements, but streams process elements independently. The imperative solution:

List<List<Reading>> findSuspicious(Stream<Reading> source) {
    var suspicious = new ArrayList<List<Reading>>();
    Reading previous = null;
    boolean hasPrevious = false;
    
    for (Reading next : source.toList()) {
        if (!hasPrevious) {
            hasPrevious = true;
            previous = next;
        } else {
            if (isSuspicious(previous, next))
                suspicious.add(List.of(previous, next));
            previous = next;
        }
    }
    return suspicious;
}

You’ve abandoned the stream. No more declarative pipelines, no parallelism, no lazy evaluation. The logic is buried in imperative loops.

The Solution: Stream Gatherers

Stream::gather(Gatherer) is the new intermediate operation that processes elements using a gatherer — a user-defined transformation. Just as Stream::collect(Collector) provides an extension point for terminal operations, gather provides one for intermediate operations.

A gatherer transforms elements in flexible ways:

• One-to-one: Like map (each input → one output)
• One-to-many: Like flatMap (each input → multiple outputs)
• Many-to-one: Like reduce (multiple inputs → one output)
• Many-to-many: Custom windowing, filtering, grouping

Built-In Gatherers

The Gatherers class provides five built-in gatherers:

1. windowFixed(int windowSize)

Group elements into fixed-size lists:

var windows = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                    .gather(Gatherers.windowFixed(3))
                    .toList();

// windows => [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

The last window may contain fewer elements if the stream doesn’t divide evenly.

2. windowSliding(int windowSize)

Group elements into overlapping windows:

var windows = Stream.of(1, 2, 3, 4, 5)
                    .gather(Gatherers.windowSliding(3))
                    .toList();

// windows => [[1, 2, 3], [2, 3, 4], [3, 4, 5]]

Each window slides one element forward. Perfect for detecting patterns across consecutive elements.

3. scan(Supplier<R> initial, BiFunction<R, T, R> scanner)

Produce a cumulative result for each element (prefix scan):

var prefixSums = Stream.of(1, 2, 3, 4, 5)
                       .gather(Gatherers.scan(() -> 0, Integer::sum))
                       .toList();

// prefixSums => [1, 3, 6, 10, 15]  (running totals)

Each output is the result of applying the scanner to the previous output and the current input. Equivalent to:

output[0] = scanner(initial, input[0])
output[1] = scanner(output[0], input[1])
output[2] = scanner(output[1], input[2])
...

4. fold(Supplier<R> initial, BiFunction<R, T, R> folder)

Accumulate all elements into a single result, emitted when the stream ends (many-to-one):

var sum = Stream.of(1, 2, 3, 4, 5)
                .gather(Gatherers.fold(() -> 0, Integer::sum))
                .findFirst();

// sum => Optional[15]

Similar to reduce, but usable as an intermediate operation. Downstream operations receive the final accumulated value.

5. mapConcurrent(int maxConcurrency, Function<T, R> mapper)

Apply a function to each element concurrently using virtual threads:

var results = Stream.of(urls)
                    .gather(Gatherers.mapConcurrent(10, url -> fetch(url)))
                    .toList();

Up to 10 virtual threads execute fetch(url) concurrently. Results are delivered in encounter order (not completion order), preserving stream semantics.

Solving Previous Problems

Distinctness by length (using windowSliding + filter):

// Not the most direct, but demonstrates sliding windows
var result = Stream.of("foo", "bar", "baz", "quux")
                   .sorted(Comparator.comparingInt(String::length))
                   .gather(Gatherers.windowSliding(2))
                   .filter(window -> window.size() < 2 
                                     || window.get(0).length() != window.get(1).length())
                   .map(window -> window.get(0))
                   .toList();

Or better, define a custom distinctBy gatherer (shown later).

Fixed-size windowing with limit:

var result = Stream.iterate(0, i -> i + 1)
                   .gather(Gatherers.windowFixed(3))
                   .limit(2)
                   .toList();

// result => [[0, 1, 2], [3, 4, 5]]

Clean, declarative, handles infinite streams correctly.

Suspicious temperature changes (using windowSliding):

List<List<Reading>> findSuspicious(Stream<Reading> source) {
    return source.gather(Gatherers.windowSliding(2))
                 .filter(window -> window.size() == 2 
                                   && isSuspicious(window.get(0), window.get(1)))
                 .toList();
}

The stream paradigm preserved. Logic is clear, composable, and potentially parallelizable.

Technical Deep Dive: Defining Custom Gatherers

The Gatherer<T, A, R> interface defines a transformation from input elements of type T to output elements of type R, using intermediate state of type A. Four functions compose a gatherer:

1. Initializer (Supplier<A>): Create the initial state object. Invoked once per evaluation (or per partition in parallel streams).

2. Integrator (Integrator<A, T, R>): Process each input element. Receives the state, the element, and a Downstream object for emitting results. Returns boolean: true to continue processing, false to short-circuit.

3. Combiner (BinaryOperator<A>): Merge two state objects during parallel evaluation. Optional; if omitted, the gatherer is evaluated sequentially even in parallel streams.

4. Finisher (BiConsumer<A, Downstream<? super R>>): Perform final actions when input is exhausted. Can emit additional results or perform cleanup.

Example: Fixed-Size Windows (Full Implementation)

Let’s implement windowFixed from scratch:

record WindowFixed<TR>(int windowSize)
    implements Gatherer<TR, ArrayList<TR>, List<TR>>
{
    public WindowFixed {
        if (windowSize < 1)
            throw new IllegalArgumentException("window size must be positive");
    }

    @Override
    public Supplier<ArrayList<TR>> initializer() {
        return () -> new ArrayList<>(windowSize);
    }

    @Override
    public Integrator<ArrayList<TR>, TR, List<TR>> integrator() {
        return Integrator.ofGreedy((window, element, downstream) -> {
            window.add(element);
            
            if (window.size() < windowSize)
                return true;  // Keep accepting elements
            
            // Window is full - emit a copy and clear
            var result = new ArrayList<>(window);
            window.clear();
            
            return downstream.push(result);
        });
    }

    @Override
    public BiConsumer<ArrayList<TR>, Downstream<? super List<TR>>> finisher() {
        return (window, downstream) -> {
            if (!downstream.isRejecting() && !window.isEmpty()) {
                downstream.push(new ArrayList<>(window));
                window.clear();
            }
        };
    }
}

Key observations:

• State: ArrayList<TR> accumulates elements for the current window.
• Integrator: Uses Integrator.ofGreedy to signal it never initiates short-circuiting (always consumes all input unless downstream rejects). Adds elements until the window is full, then emits a copy (preserving immutability) and clears the state.
• Finisher: Emits the final partial window (if any) when the stream ends.
• No combiner: Fixed windowing is inherently sequential (order matters), so parallel evaluation is not supported.

Usage:

Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .gather(new WindowFixed<>(3))
      .toList();

// => [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

Simplified Ad-Hoc Gatherers

For simple cases, use factory methods instead of implementing the full interface:

static <TR> Gatherer<TR, ?, List<TR>> fixedWindow(int windowSize) {
    if (windowSize < 1)
        throw new IllegalArgumentException("window size must be positive");

    return Gatherer.ofSequential(
        // Initializer: create state
        () -> new ArrayList<TR>(windowSize),
        
        // Integrator: process each element
        Integrator.ofGreedy((window, element, downstream) -> {
            window.add(element);
            
            if (window.size() < windowSize)
                return true;
            
            var result = new ArrayList<>(window);
            window.clear();
            
            return downstream.push(result);
        }),
        
        // Finisher: emit partial window
        (window, downstream) -> {
            if (!downstream.isRejecting() && !window.isEmpty()) {
                downstream.push(new ArrayList<>(window));
                window.clear();
            }
        }
    );
}

Gatherer.ofSequential constructs a gatherer from initializer, integrator, and finisher. The combiner is omitted (sequential-only). This approach is concise and suitable for most custom gatherers.

Implementation Details: Greedy vs. Non-Greedy Integrators

The Integrator.ofGreedy method signals that the integrator will never initiate short-circuiting based on its own logic. It may return false only to relay that downstream doesn’t want more elements. This distinction enables optimizations.

Greedy integrator (never short-circuits):

Integrator.ofGreedy((state, element, downstream) -> {
    // Always processes element
    state.add(element);
    return downstream.push(element);  // Relay downstream's decision
});

Non-greedy integrator (can short-circuit):

Integrator.of((state, element, downstream) -> {
    if (state.count >= limit) {
        return false;  // Initiate short-circuit
    }
    state.count++;
    return downstream.push(element);
});

For greedy integrators, the stream library can optimize evaluation by avoiding unnecessary checks for early termination.

Actual Implementation: windowFixed Internals

The real windowFixed implementation in Gatherers.java uses an Object[] array for efficiency:

public static <TR> Gatherer<TR, ?, List<TR>> windowFixed(int windowSize) {
    if (windowSize < 1)
        throw new IllegalArgumentException("'windowSize' must be greater than zero");

    class FixedWindow {
        Object[] window;
        int at;

        FixedWindow() {
            at = 0;
            window = new Object[windowSize];
        }

        boolean integrate(TR element, Downstream<? super List<TR>> downstream) {
            window[at++] = element;
            
            if (at < windowSize) {
                return true;
            } else {
                final var oldWindow = window;
                window = new Object[windowSize];
                at = 0;
                
                return downstream.push(
                    SharedSecrets.getJavaUtilCollectionAccess()
                                 .listFromTrustedArrayNullsAllowed(oldWindow)
                );
            }
        }

        void finish(Downstream<? super List<TR>> downstream) {
            if (!downstream.isRejecting() && at > 0) {
                final var oldWindow = window;
                window = new Object[windowSize];
                at = 0;
                
                downstream.push(
                    SharedSecrets.getJavaUtilCollectionAccess()
                                 .listFromTrustedArrayNullsAllowed(
                                     Arrays.copyOf(oldWindow, at)
                                 )
                );
            }
        }
    }

    return Gatherer.ofSequential(
        FixedWindow::new,
        Integrator.ofGreedy(FixedWindow::integrate),
        FixedWindow::finish
    );
}

Optimizations:

• Object[] instead of ArrayList: Avoids allocation overhead and type checks. The array is reused across windows (just reset the index).
• SharedSecrets.getJavaUtilCollectionAccess(): Internal API to create lists directly from arrays without copying (the array is trusted not to be mutated).
• Copy only on partial window: The finisher uses Arrays.copyOf to trim the final partial window to actual size.

These micro-optimizations matter for high-throughput scenarios (millions of elements, small window sizes).

Parallel Evaluation and Combiners

Gatherers can be parallelized if they provide a combiner function. Parallel evaluation works in two modes:

1. Combiner Not Provided (Sequential Evaluation)

Even in a parallel stream, the gatherer is evaluated sequentially. However, upstream and downstream operations can still run in parallel. This is analogous to parallel().forEachOrdered().

Example:

Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .parallel()
      .gather(Gatherers.windowFixed(3))  // Sequential
      .forEach(window -> System.out.println(window));

The windowing happens sequentially (order matters), but upstream element generation and downstream consumption can be parallel.

2. Combiner Provided (Parallel Evaluation)

If a combiner is provided, the gatherer can be split across partitions, evaluated independently, and combined. This is analogous to parallel().reduce().

Example: Parallelizable selectOne (pick the maximum element):

static <TR> Gatherer<TR, ?, TR> selectOne(BinaryOperator<TR> selector) {
    class State {
        TR value;
        boolean hasValue;
    }

    return Gatherer.of(
        // Initializer
        State::new,
        
        // Integrator (greedy - never short-circuits)
        Integrator.ofGreedy((state, element, downstream) -> {
            if (!state.hasValue) {
                state.value = element;
                state.hasValue = true;
            } else {
                state.value = selector.apply(state.value, element);
            }
            return true;
        }),
        
        // Combiner (merge two states)
        (leftState, rightState) -> {
            if (!leftState.hasValue) {
                return rightState;
            } else if (!rightState.hasValue) {
                return leftState;
            } else {
                leftState.value = selector.apply(leftState.value, rightState.value);
                return leftState;
            }
        },
        
        // Finisher
        (state, downstream) -> {
            if (state.hasValue)
                downstream.push(state.value);
        }
    );
}

Usage:

var max = Stream.generate(() -> ThreadLocalRandom.current().nextInt())
                .limit(1_000_000)
                .parallel()
                .gather(selectOne(Math::max))
                .findFirst();

// max => Optional[2147483581] (or similar large value)

The stream is split into partitions. Each partition finds its local maximum. The combiner merges these maximums to produce the global maximum. Full parallelism achieved.

Practical Examples

Example 1: Running Average (Scan)

Calculate a running average of temperatures:

record State(double sum, int count) {}

var runningAverages = Stream.of(10, 20, 15, 25, 30)
    .gather(
        Gatherers.scan(() -> new State(0, 0),
                      (state, temp) -> new State(state.sum + temp, state.count + 1))
    )
    .map(state -> state.sum / state.count)
    .toList();

// runningAverages => [10.0, 15.0, 15.0, 17.5, 20.0]

Each output is the average of all elements up to that point.

Example 2: Distinct by Key

Implement distinctBy to filter elements by a key extractor:

static <T, K> Gatherer<T, ?, T> distinctBy(Function<T, K> keyExtractor) {
    class State {
        Set<K> seen = new HashSet<>();
    }

    return Gatherer.ofSequential(
        State::new,
        Integrator.ofGreedy((state, element, downstream) -> {
            K key = keyExtractor.apply(element);
            if (state.seen.add(key)) {
                return downstream.push(element);
            }
            return true;  // Skip duplicate, continue
        })
    );
}

Usage:

var result = Stream.of("foo", "bar", "baz", "quux")
                   .gather(distinctBy(String::length))
                   .toList();

// result => [foo, quux]  (lengths 3 and 4, first occurrence of each)

Clean, reusable, composable.

Example 3: Take While with Count

Take elements while a predicate holds, up to a maximum count:

static <T> Gatherer<T, ?, T> takeWhile(Predicate<T> predicate, int maxCount) {
    class State {
        int count = 0;
    }

    return Gatherer.ofSequential(
        State::new,
        Integrator.of((state, element, downstream) -> {
            if (state.count >= maxCount || !predicate.test(element)) {
                return false;  // Short-circuit
            }
            state.count++;
            return downstream.push(element);
        })
    );
}

Usage:

var result = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                   .gather(takeWhile(x -> x < 10, 5))
                   .toList();

// result => [1, 2, 3, 4, 5]  (stopped at count)

Demonstrates short-circuiting: the integrator returns false to stop processing early.

Composing Gatherers

Gatherers support composition via andThen:

var result = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
    .gather(
        Gatherers.windowFixed(3)
                 .andThen(Gatherers.fold(() -> new ArrayList<>(), 
                                         (acc, window) -> { acc.add(window); return acc; }))
    )
    .findFirst();

// result => Optional[[[1, 2, 3], [4, 5, 6], [7, 8, 9]]]

Semantically equivalent to:

Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .gather(Gatherers.windowFixed(3))
      .gather(Gatherers.fold(...))
      .findFirst();

But andThen may enable optimizations (the stream library can fuse the two gatherers into a single pass).

Performance Considerations

Stateless vs. Stateful

Stateless gatherers (no initializer, or initializer returns null) can be optimized:

// Stateless map-like gatherer
var doubler = Gatherer.ofSequential(
    Integrator.ofGreedy((unused, element, downstream) ->
        downstream.push(element * 2)
    )
);

The stream library can avoid state allocation and management overhead. For stateful gatherers, state must be created per partition (in parallel streams) or per evaluation (in sequential streams).

Greedy vs. Non-Greedy

Greedy integrators allow the stream library to skip short-circuit checks:

// Greedy: never initiates short-circuit
Integrator.ofGreedy((state, element, downstream) -> {
    state.add(element);
    return downstream.push(element);
});

// Non-greedy: may short-circuit
Integrator.of((state, element, downstream) -> {
    if (shouldStop(state, element)) {
        return false;
    }
    return downstream.push(element);
});

For high-throughput scenarios (millions of elements), the difference can be measurable (5-10% overhead for non-greedy checks).

Parallel Evaluation Overhead

Providing a combiner enables parallelism but adds overhead:

• Partition splitting: Stream is divided into chunks
• State duplication: Each partition gets its own state object
• Combining: Results must be merged

For small streams (< 1,000 elements), sequential evaluation is often faster. For large streams with expensive per-element operations, parallel evaluation wins.

Benchmark (1,000,000 elements, window size 100):

Benchmark (10,000,000 elements, expensive operation):

Use parallelism when per-element cost is high relative to overhead.

Migration Path and Best Practices

When to Use Gatherers vs. Built-In Operations

Use built-in operations when they fit naturally:

• filter, map, flatMap, distinct, sorted, limit, skip — these are optimized and well-understood

Use gatherers when built-in operations don’t suffice:

• Windowing (fixed or sliding)
• Stateful transformations requiring previous elements
• Custom grouping logic
• Concurrent mapping with virtual threads

Don’t Over-Engineer

Simple transformations don’t need gatherers:

// Bad: overusing gatherers
var doubled = stream.gather(
    Gatherer.ofSequential(
        Integrator.ofGreedy((unused, x, downstream) -> downstream.push(x * 2))
    )
).toList();

// Good: use map
var doubled = stream.map(x -> x * 2).toList();

Gatherers add complexity. Use them when built-in operations can’t express your intent.

Composition Over Complexity

Prefer composing simple gatherers:

// Good: compose simple gatherers
var result = stream
    .gather(Gatherers.windowFixed(10))
    .gather(Gatherers.scan(() -> 0, (sum, window) -> sum + window.size()))
    .toList();

// Bad: one complex gatherer doing everything
var result = stream.gather(customComplexWindowAndScanGatherer()).toList();

Composition aids readability and reusability.

Conclusions

JEP 485 fills a long-standing gap in the Stream API. For eight years, developers have worked around the fixed set of intermediate operations with imperative loops, custom collectors, or abandoning streams entirely. Stream Gatherers restore declarative style for complex transformations.

The design mirrors Collector but adapted for intermediate operations: stateful, short-circuitable, parallelizable. The built-in gatherers (windowFixed, windowSliding, scan, fold, mapConcurrent) handle common patterns, while the Gatherer interface enables infinite customization.

Key strengths:

• Flexibility: One-to-one, one-to-many, many-to-one, many-to-many transformations
• Composability: andThen for building complex pipelines
• Parallel-ready: Provide a combiner for data parallelism
• Familiar: Same vocabulary as Collector (initializer, integrator, combiner, finisher)

For developers building data processing pipelines, framework authors manipulating bytecode streams, or anyone who’s ever thought “I wish streams could do X,” JEP 485 is transformative. The Stream API is no longer a fixed toolkit — it’s an extensible platform.

The future is open: as patterns emerge, new built-in gatherers can be added. But unlike previous expansions of the Stream API, these additions don’t bloat the core interface. gather(Gatherer) is the single extension point. Everything else is library code, composable and reusable.

Java streams just got infinitely more powerful.

References

• JEP 485
• Stream API
• Gatherer interface
• Gatherers class
• Implementation: Gatherer.java, Gatherers.java