Streams have had a fixed set of intermediate operations since Java 8 — filter,
map, flatMap, distinct, sorted, and a handful
more. You could add your own terminal operation with a Collector, but there was no way to
add an intermediate one. Gatherers are that missing extension point.
The gap
Some transformations simply could not be expressed. Grouping elements into batches of three,
producing a running total, or removing duplicates by a key rather than by
equals — all of these meant dropping out of the stream, doing it with a loop, and
starting a new stream:
class Demo {
// Batching, the pre-gatherer way
List<List<Integer>> batches(List<Integer> items, int size) {
List<List<Integer>> result = new ArrayList<>();
for (int i = 0; i < items.size(); i += size) {
result.add(new ArrayList<>(items.subList(i, Math.min(i + size, items.size()))));
}
return result;
}
}
The reason for the gap is that these operations are stateful — each element's output
depends on what came before it. map and filter look at one element in
isolation, which is what makes them easy to parallelise and easy to define.
The built-in gatherers
Finalised in Java 24 and available in 25. Four cover most of what people were missing:
class Demo {
void run() {
var numbers = List.of(1, 2, 3, 4, 5);
// Fixed-size batches — the last one is short
System.out.println(numbers.stream().gather(Gatherers.windowFixed(2)).toList());
// [[1, 2], [3, 4], [5]]
// Overlapping windows
System.out.println(numbers.stream().gather(Gatherers.windowSliding(2)).toList());
// [[1, 2], [2, 3], [3, 4], [4, 5]]
// Running totals — one output per input
System.out.println(numbers.stream()
.gather(Gatherers.scan(() -> 0, Integer::sum)).toList());
// [1, 3, 6, 10, 15]
// Fold to a single value, as a stream of one
System.out.println(numbers.stream()
.gather(Gatherers.fold(() -> 0, Integer::sum)).toList());
// [15]
}
}
Each of those four returns a stream, so the pipeline continues afterwards — that is the whole
point. windowFixed is the one you will reach for most — batching records for a bulk insert or
an API that takes a limited number of ids per call is a genuinely common job.
scan versus fold is the distinction worth fixing in your head:
scan emits every intermediate result, fold emits only the final one.
Writing your own
The interface has an initialiser, an integrator and an optional finisher. The integrator receives
the state, an element, and a downstream to push results into — and returns false to stop
the stream early:
class Demo {
// Keep the first element for each key — "distinct by", which Stream never had
static <T, K> Gatherer<T, ?, T> distinctBy(Function<T, K> keyFn) {
return Gatherer.ofSequential(
HashSet<K>::new,
(seen, element, downstream) -> {
if (seen.add(keyFn.apply(element))) {
downstream.push(element);
}
return true; // keep going
});
}
void run() {
record Person(String name, String city) { }
var people = List.of(
new Person("Ana", "Seattle"),
new Person("Bo", "Seattle"),
new Person("Cy", "Nuku'alofa"));
System.out.println(people.stream()
.gather(distinctBy(Person::city))
.map(Person::name)
.toList()); // [Ana, Cy]
}
}
Gatherer.ofSequential is the right factory whenever the operation depends on element
order, which stateful operations usually do. The plain Gatherer.of takes an extra
combiner and can run in parallel.
Returning false from the integrator short-circuits — that is how you would build a
takeUntil that stops at the first match:
class Demo {
static <T> Gatherer<T, ?, T> takeUntil(Predicate<T> stop) {
return Gatherer.ofSequential(
() -> null,
(state, element, downstream) -> {
downstream.push(element);
return !stop.test(element); // false ends the stream
});
}
void run() {
System.out.println(List.of(1, 2, 3, 4, 5).stream()
.gather(takeUntil(n -> n >= 3))
.toList()); // [1, 2, 3]
}
}
Composing them
Gatherers chain, both with each other and with ordinary stream operations, which is what makes them feel like part of the API rather than an escape hatch:
class Demo {
void run() {
var numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8);
System.out.println(numbers.stream()
.filter(n -> n % 2 == 0) // ordinary operation
.gather(Gatherers.windowFixed(2)) // gatherer
.map(window -> window.stream().mapToInt(Integer::intValue).sum())
.toList()); // [6, 14]
// Two gatherers back to back
System.out.println(numbers.stream()
.gather(Gatherers.scan(() -> 0, Integer::sum))
.gather(Gatherers.windowFixed(3))
.toList());
// [[1, 3, 6], [10, 15, 21], [28, 36]]
}
}
There is also andThen, which fuses two gatherers into one so a reusable pipeline
fragment can be passed around as a single value:
class Demo {
void run() {
var pipeline = Gatherers.<Integer>windowFixed(2)
.andThen(Gatherers.scan(() -> 0, (acc, window) ->
acc + window.stream().mapToInt(Integer::intValue).sum()));
System.out.println(List.of(1, 2, 3, 4).stream().gather(pipeline).toList());
// [3, 10]
}
}
Gatherer or Collector?
| Gatherer | Collector | |
|---|---|---|
| Position | intermediate — stream in, stream out | terminal — stream in, value out |
| Called with | .gather(...) | .collect(...) |
| Can short-circuit | yes | no |
| Chainable after | yes — more stream operations follow | no — the stream is finished |
The practical test: if you want to keep operating on the results, you want a gatherer. If you are
producing the final answer, you want a collector. The
distinctBy example above works precisely because the stream continues afterwards.
Do you need one?
Mostly not. The built-in operations cover ordinary work, and a custom gatherer is more machinery than a loop for a one-off job. They earn their place when you have a genuinely reusable stateful transformation — batching, deduplicating by key, rate-limiting a stream — that several call sites want, and where dropping out of the pipeline would break a chain that otherwise reads well.
The wider point is that streams are no longer a closed set of operations. That was the main structural complaint about the API for a decade, and it is now answered.
Next
Flexible Constructor Bodies is next — validating
an argument before calling super().