filter
map
flatMap
distinct
sorted
limit
skip
takeWhile
dropWhile

custom intermediate operation for Stream pipelines

Collector customizes how a stream ends.
Gatherer customizes what happens in the middle.

source -> intermediate operations -> terminal operation

orders.stream()
    .filter(Order::isValid)
    .collect(customCollector());

List<List<Order>> batches = new ArrayList<>();
List<Order> current = new ArrayList<>(100);

for (Order order : orders) {
    current.add(order);
    if (current.size() == 100) {
        batches.add(List.copyOf(current));
        current.clear();
    }
}
if (!current.isEmpty()) {
    batches.add(List.copyOf(current));
}

batches.stream()
    .map(batchProcessor::process)
    .toList();

List<Result> results = orders.stream()
    .gather(Gatherers.windowFixed(100))
    .map(batchProcessor::process)
    .toList();

Map<CustomerId, List<Order>> result = orders.stream()
    .collect(Collectors.groupingBy(Order::customerId));

List<List<Order>> batches = orders.stream()
    .gather(Gatherers.windowFixed(100))
    .toList();

Collector consumes the stream and returns a result.
Gatherer consumes upstream elements and pushes downstream elements.

Gatherer<T, A, R>

T = input element type
A = internal state type
R = output element type

Gatherer<Order, ?, List<Order>> fixedWindows = Gatherers.windowFixed(100);

T = Order
A = hidden internal state
R = List<Order>

Order, Order, Order, Order, ...

List<Order>, List<Order>, List<Order>, ...

initializer -> creates state
integrator  -> integrates input element and may push output downstream
combiner    -> combines states for parallel execution
finisher    -> final action at end of upstream

A state = gatherer.initializer().get();
for (T element : source) {
    gatherer.integrator().integrate(state, element, downstream);
}
gatherer.finisher().accept(state, downstream);

Gatherer.Downstream<? super R> downstream

downstream.push(output);

Gatherer generalizes many stateful intermediate transformations.

Gatherers.windowFixed(int windowSize)
Gatherers.windowSliding(int windowSize)
Gatherers.fold(Supplier<R> initial, BiFunction<? super R, ? super T, ? extends R> folder)
Gatherers.scan(Supplier<R> initial, BiFunction<? super R, ? super T, ? extends R> scanner)
Gatherers.mapConcurrent(int maxConcurrency, Function<? super T, ? extends R> mapper)

1, 2, 3, 4, 5, 6, 7, 8

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

[ [1,2,3], [4,5,6], [7,8] ]

List<BatchResult> results = orders.stream()
    .gather(Gatherers.windowFixed(100))
    .map(orderBatch -> orderClient.submitBatch(orderBatch))
    .toList();

Last window may be smaller than windowSize.
Empty stream produces no window.

Should downstream receive mutable or unmodifiable window lists?

1, 2, 3, 4, 5

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

[ [1,2,3], [2,3,4], [3,4,5] ]

List<BigDecimal> movingAverage = balances.stream()
    .gather(Gatherers.windowSliding(3))
    .map(window -> window.stream()
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .divide(BigDecimal.valueOf(window.size()), MathContext.DECIMAL64))
    .toList();

Sliding windows duplicate references across windows.
Do not mutate elements through window views.

List<String> digests = events.stream()
    .gather(Gatherers.fold(
        () -> new StringBuilder(),
        (builder, event) -> builder.append(event.code()).append('|')
    ))
    .map(StringBuilder::toString)
    .toList();

List<String> digest = events.stream()
    .gather(Gatherers.fold(
        () -> "",
        (current, event) -> current + event.code() + "|"
    ))
    .toList();

List<BigDecimal> finalBalance = transactions.stream()
    .gather(Gatherers.fold(
        () -> BigDecimal.ZERO,
        (balance, tx) -> balance.add(tx.amount())
    ))
    .toList();

If you only need one final result, Collector may be clearer.
If you need to stay inside pipeline as an intermediate stage, Gatherer can be useful.

10, -3, 5

List<Integer> running = Stream.of(10, -3, 5)
    .gather(Gatherers.scan(
        () -> 0,
        Integer::sum
    ))
    .toList();

[10, 7, 12]

record Transaction(String id, BigDecimal amount) {}
record BalancePoint(String transactionId, BigDecimal balance) {}

List<BigDecimal> balances = transactions.stream()
    .map(Transaction::amount)
    .gather(Gatherers.scan(
        () -> BigDecimal.ZERO,
        BigDecimal::add
    ))
    .toList();

record RunningState(BigDecimal balance) {}

List<BalancePoint> points = transactions.stream()
    .gather(Gatherer.ofSequential(
        () -> new RunningState(BigDecimal.ZERO),
        (state, tx, downstream) -> {
            BigDecimal next = state.balance().add(tx.amount());
            state = new RunningState(next); // does not update external holder; not useful
            return downstream.push(new BalancePoint(tx.id(), next));
        }
    ))
    .toList();

final class BalanceState {
    BigDecimal balance = BigDecimal.ZERO;
}

Gatherer<Transaction, BalanceState, BalancePoint> runningBalance = Gatherer.ofSequential(
    BalanceState::new,
    (state, tx, downstream) -> {
        state.balance = state.balance.add(tx.amount());
        return downstream.push(new BalancePoint(tx.id(), state.balance));
    }
);

List<BalancePoint> points = transactions.stream()
    .gather(runningBalance)
    .toList();

List<CustomerProfile> profiles = customerIds.stream()
    .gather(Gatherers.mapConcurrent(
        20,
        customerClient::fetchProfile
    ))
    .toList();

The mapper must be safe for concurrent execution.
The downstream semantics must tolerate completion scheduling as documented.
The max concurrency must be chosen based on downstream capacity, not wishful thinking.

static <T, R> Gatherer<T, ?, R> mapping(Function<? super T, ? extends R> mapper) {
    Objects.requireNonNull(mapper, "mapper");

    return Gatherer.of(
        (unused, element, downstream) -> downstream.push(mapper.apply(element))
    );
}

List<String> names = users.stream()
    .gather(mapping(User::name))
    .toList();

Understand that integrator receives state, input element, and downstream.

Input:  A, A, B, B, A
Output: A, B, A

A, B

final class LastSeen<T> {
    boolean hasValue;
    T value;
}

static <T> Gatherer<T, LastSeen<T>, T> deduplicateConsecutive() {
    return Gatherer.ofSequential(
        LastSeen::new,
        (state, element, downstream) -> {
            if (!state.hasValue || !Objects.equals(state.value, element)) {
                state.hasValue = true;
                state.value = element;
                return downstream.push(element);
            }
            return true;
        }
    );
}

List<String> result = Stream.of("A", "A", "B", "B", "A")
    .gather(deduplicateConsecutive())
    .toList();

// [A, B, A]

Consecutive deduplication depends on boundary between partitions.
A naive combiner cannot correctly know whether last element of left equals first element of right unless state carries boundary information and deferred output strategy is carefully designed.

Create a batch until a boundary predicate is true.

final class BatchState<T> {
    final List<T> buffer = new ArrayList<>();
}

static <T> Gatherer<T, BatchState<T>, List<T>> batchUntil(
    Predicate<? super T> isBoundary
) {
    Objects.requireNonNull(isBoundary, "isBoundary");

    return Gatherer.ofSequential(
        BatchState::new,
        (state, element, downstream) -> {
            state.buffer.add(element);

            if (isBoundary.test(element)) {
                List<T> batch = List.copyOf(state.buffer);
                state.buffer.clear();
                return downstream.push(batch);
            }

            return true;
        },
        (state, downstream) -> {
            if (!state.buffer.isEmpty()) {
                downstream.push(List.copyOf(state.buffer));
            }
        }
    );
}

List<List<String>> records = lines.stream()
    .gather(batchUntil(line -> line.equals("END")))
    .toList();

The finisher flushes the final incomplete batch.

record Transaction(String id, BigDecimal amount) {}
record BalancePoint(String transactionId, BigDecimal balance) {}

final class BalanceState {
    BigDecimal balance = BigDecimal.ZERO;
}

static Gatherer<Transaction, BalanceState, BalancePoint> runningBalance() {
    return Gatherer.ofSequential(
        BalanceState::new,
        (state, tx, downstream) -> {
            state.balance = state.balance.add(tx.amount());
            return downstream.push(new BalancePoint(tx.id(), state.balance));
        }
    );
}

List<BalancePoint> points = transactions.stream()
    .gather(runningBalance())
    .toList();

return downstream.push(output);

downstream.push(output);
return true;

return downstream.push(output);

for (R output : outputs) {
    if (!downstream.push(output)) {
        return false;
    }
}
return true;

static final List<Object> leakedStates = new ArrayList<>();

Gatherer<T, State, R> broken = Gatherer.ofSequential(
    State::new,
    (state, element, downstream) -> {
        leakedStates.add(state);
        return true;
    }
);

static Gatherer<T, List<T>, List<T>> leakingBatches() {
    return Gatherer.ofSequential(
        ArrayList::new,
        (buffer, element, downstream) -> {
            buffer.add(element);
            if (buffer.size() == 10) {
                downstream.push(buffer); // leaks mutable internal buffer
                buffer.clear();          // downstream sees cleared list
            }
            return true;
        }
    );
}

downstream.push(List.copyOf(buffer));
buffer.clear();

Never push mutable internal state downstream unless ownership transfer is explicit and safe.

Gatherer.ofSequential(...)

(left, right) -> left

Declare the gatherer sequential if you cannot prove partition merge correctness.

Read records from a binary file lazily -> Spliterator
Batch existing stream elements into fixed windows -> Gatherer
Aggregate all elements into report -> Collector

orders.stream()
    .mapMulti((order, downstream) -> {
        for (LineItem item : order.items()) {
            downstream.accept(item);
        }
    })
    .toList();

mapMulti is per-element expansion.
Gatherer is stream-stateful transformation.

Take stream of IDs, batch them into 100 IDs, call remote API once per batch.

List<CustomerProfile> profiles = customerIds.stream()
    .gather(Gatherers.windowFixed(100))
    .map(customerClient::fetchProfiles)
    .flatMap(List::stream)
    .toList();

List<CustomerProfile> profiles = customerIds.stream()
    .gather(Gatherers.windowFixed(100))
    .map(batch -> customerProfileGateway.fetchBatch(batch, retryPolicy))
    .flatMap(List::stream)
    .toList();

Raise alert if any 5 consecutive transactions exceed threshold.

List<RiskAlert> alerts = transactions.stream()
    .gather(Gatherers.windowSliding(5))
    .map(window -> riskRules.evaluate(window))
    .filter(Optional::isPresent)
    .map(Optional::get)
    .toList();

List<RiskAlert> alerts = transactions.stream()
    .gather(Gatherers.windowSliding(5))
    .map(riskRules::evaluate)
    .flatMap(Optional::stream)
    .toList();

Window order must match transaction encounter order.

transactions.stream()
    .sorted(Comparator.comparing(Transaction::postedAt).thenComparing(Transaction::id))
    .gather(Gatherers.windowSliding(5))

Turn ordered events into state snapshots.

record CaseEvent(String id, String type, Instant at) {}
record CaseSnapshot(String eventId, String status) {}

final class CaseState {
    String status = "NEW";

    void apply(CaseEvent event) {
        status = switch (event.type()) {
            case "SUBMIT" -> "SUBMITTED";
            case "APPROVE" -> "APPROVED";
            case "REJECT" -> "REJECTED";
            default -> status;
        };
    }
}

static Gatherer<CaseEvent, CaseState, CaseSnapshot> caseSnapshots() {
    return Gatherer.ofSequential(
        CaseState::new,
        (state, event, downstream) -> {
            state.apply(event);
            return downstream.push(new CaseSnapshot(event.id(), state.status));
        }
    );
}

List<CaseSnapshot> snapshots = events.stream()
    .sorted(Comparator.comparing(CaseEvent::at).thenComparing(CaseEvent::id))
    .gather(caseSnapshots())
    .toList();

Stateful projection must have deterministic input order.

List<List<Order>> batches = orders.stream()
    .collect(customBatchCollector());

return batches.stream()
    .map(this::process)
    .toList();

return orders.stream()
    .gather(Gatherers.windowFixed(100))
    .map(this::process)
    .toList();

if (buffer.size() == 100) {
    downstream.push(List.copyOf(buffer));
    buffer.clear();
}

What about final 1..99 elements?

downstream.push(buffer);
buffer.clear();

downstream.push(List.copyOf(buffer));
buffer.clear();

(left, right) -> left

downstream.push(value);
return true;

return downstream.push(value);

map.values().stream()
    .gather(runningBalance())

map.values().stream()
    .sorted(Comparator.comparing(Transaction::postedAt).thenComparing(Transaction::id))
    .gather(runningBalance())

ids.stream()
    .gather(Gatherers.mapConcurrent(100, client::fetch))
    .toList();

@Test
void deduplicatesConsecutiveValues() {
    List<String> result = Stream.of("A", "A", "B", "B", "A")
        .gather(deduplicateConsecutive())
        .toList();

    assertThat(result).containsExactly("A", "B", "A");
}

@Test
void emptyInputProducesEmptyOutput() {
    List<String> result = Stream.<String>empty()
        .gather(deduplicateConsecutive())
        .toList();

    assertThat(result).isEmpty();
}

@Test
void flushesIncompleteFinalBatch() {
    List<List<Integer>> result = Stream.of(1, 2, 3, 4, 5)
        .gather(batchUntil(i -> i == 3))
        .toList();

    assertThat(result).containsExactly(
        List.of(1, 2, 3),
        List.of(4, 5)
    );
}

@Test
void respectsLimitDownstream() {
    List<List<Integer>> result = Stream.iterate(1, i -> i + 1)
        .gather(Gatherers.windowFixed(3))
        .limit(2)
        .toList();

    assertThat(result).containsExactly(
        List.of(1, 2, 3),
        List.of(4, 5, 6)
    );
}

List<BalancePoint> result = shuffledTransactions.stream()
    .sorted(Comparator.comparing(Transaction::postedAt).thenComparing(Transaction::id))
    .gather(runningBalance())
    .toList();

CaseGatherers.caseSnapshots()
RiskGatherers.slidingExposureWindow(5)
ImportGatherers.batchUntilBoundary(...)

CustomGatherers.process()
StreamUtils.doStuff()

Returns a sequential gatherer that emits a CaseSnapshot after each event.
The input stream must be ordered by event time and deterministic tie-breaker.
Null events are not permitted.
The gatherer does not parallelize because state transition semantics are order-dependent.

List<List<T>> batches = makeBatches(input);
return batches.stream().map(this::process).toList();

return input.stream()
    .gather(Gatherers.windowFixed(100))
    .map(this::process)
    .toList();

Use Gatherer for semantic clarity first.
Optimize only after measuring representative workload.

Collector = custom terminal reduction.
Gatherer = custom intermediate transformation.
Spliterator = custom source traversal.

100, 100, 53

record BalancePoint(String transactionId, BigDecimal balance) {}

A, A, B, B, A -> A, B, A

1, 2, 3, 4, 5 with boundary 3 -> [1,2,3], [4,5]

A gatherer is correct when it transforms upstream elements into downstream elements without leaking state, lying about parallelizability, or hiding ordering assumptions.