javademos/docs/16-gatherers.md

16. Stream Gatherers (JEP 485, final since JDK 24)

Prev: 15. Build files · Next: 17. Scoped values vs ThreadLocal

Compiles and runs on JDK 25+ with no --enable-preview flag -- JEP 485 finalized in JDK 24 and shipped unchanged in 25. Source: gatherers/src/, transcripts 98-basics, 98-mapconcurrent, 98-custom-gatherer, 98-collector-vs-gatherer.

The five built-in gatherers (GathererBasics.java, MapConcurrentDemo.java)

Stream.of(1,2,3,4,5,6,7).gather(Gatherers.windowFixed(3)).toList();   // [[1,2,3],[4,5,6],[7]]
Stream.of(1,2,3,4,5).gather(Gatherers.windowSliding(3)).toList();     // [[1,2,3],[2,3,4],[3,4,5]]
Stream.of(1,2,3,4).gather(Gatherers.fold(() -> "", (a,n) -> a+"["+n+"]")).findFirst();  // "[1][2][3][4]"
Stream.of(1,2,3,4,5).gather(Gatherers.scan(() -> 0, Integer::sum)).toList();            // [1,3,6,10,15]
Stream.of(1,2,3,4).gather(Gatherers.mapConcurrent(4, Worker::call)).toList();           // order preserved

• windowFixed(n) throws IllegalArgumentException for n <= 0 ("'windowSize' must be greater than zero"), not a silent empty result.
• windowSliding(n) does not drop a too-short remainder the way windowFixed shrinks its last window -- on a stream shorter than n, it still emits one window holding whatever is there ([1,2] for windowSliding(3) over [1,2]), and only an empty source stream produces zero windows. Easy to get backwards if you assume it mirrors windowFixed's "shrink the leftover" rule.
• fold and scan share a signature (Supplier<R>, BiFunction<R,T,R>) but not a shape: fold emits exactly one final value (a stream of size 1), scan emits every intermediate accumulation. Neither has a combiner, so neither runs in parallel -- both are sequential-only by design, which is what lets R differ from T in the first place (Stream.reduce can't do that safely in parallel without a combiner).
• mapConcurrent(max, fn) runs each element's mapper on its own virtual thread (confirmed via Thread.currentThread().isVirtual()), bounded to max in flight at once, output in encounter order regardless of completion order. The one behavior worth isolating: a mapper that throws does not cancel its still-running siblings. 98-mapconcurrent.txt has element 3 throw immediately while elements 1/2/4/5 each sleep 900ms -- the exception doesn't surface until ~900ms later, once every already-started mapper has finished. mapConcurrent is not StructuredTaskScope; there's no fail-fast cancellation here.

Writing a custom Gatherer (CustomGatherer.java)

Gatherer<T,A,R> is four composable pieces -- initializer (private state A), integrator (per element: touch state, push 0+ R downstream, return whether to keep pulling), combiner (parallel merge, omit for sequential-only), finisher (flush on exhaustion). Gatherer.ofSequential(...) skips the combiner entirely.

static <T> Gatherer<T, ?, T> dedupeConsecutive() {
    return Gatherer.ofSequential(() -> new ArrayList<Object>(List.of()), (state, element, downstream) -> {
        boolean changed = state.isEmpty() || !state.get(0).equals(element);
        if (changed) { if (state.isEmpty()) state.add(element); else state.set(0, element); return downstream.push(element); }
        return true;
    });
}

takeUntil in the same file is the short-circuiting case a Collector structurally cannot express: the integrator returns false once the stop condition is met, and upstream is never pulled again -- 98-custom-gatherer.txt proves it with a peek() upstream of the gather that only ever sees the 3 elements actually consumed, not all 6 in the source.

Collector vs Gatherer (CollectorVsGatherer.java)

A Collector always reduces to exactly one terminal value and can only sit at the end of a pipeline. A Gatherer is an intermediate op -- it returns Stream<R>, so a gather can be followed by filter, map, another gather, or a collect. groupingBy/summingDouble for an order-totals Map is still the right call in 98-collector-vs-gatherer.txt: the answer really is one Map, and reaching for gather() there would add ceremony with nothing to show for it. scan feeding straight into filter/map in the same lazy pipeline is the case a Collector can't do without materializing an intermediate List and starting a second stream.

Prev: 15. Build files · Next: 17. Scoped values vs ThreadLocal