Java Stream Gatherers: Custom Intermediate Operations (Java 24+)

import java.util.stream.*;
import static java.util.stream.Gatherers.*;

// Gatherers class — built-in implementations

// windowFixed(n) — partition into non-overlapping windows of size n
List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5, 6, 7)
    .gather(windowFixed(3))
    .toList();
System.out.println(windows);
// [[1, 2, 3], [4, 5, 6], [7]]   — last window may be smaller

// windowSliding(n) — overlapping windows of size n
List<List<Integer>> sliding = Stream.of(1, 2, 3, 4, 5)
    .gather(windowSliding(3))
    .toList();
System.out.println(sliding);
// [[1, 2, 3], [2, 3, 4], [3, 4, 5]]

// fold(initial, folder) — running aggregation (stateful reduce, emits one result)
// Equivalent to a stateful reduce, outputs the final accumulated value
int sum = Stream.of(1, 2, 3, 4, 5)
    .gather(fold(() -> 0, (acc, x) -> acc + x))
    .findFirst().orElse(0);
System.out.println(sum); // 15

// scan(initial, folder) — like fold but emits EACH intermediate state
List<Integer> runningSum = Stream.of(1, 2, 3, 4, 5)
    .gather(scan(() -> 0, (acc, x) -> acc + x))
    .toList();
System.out.println(runningSum); // [1, 3, 6, 10, 15]

// mapConcurrent(n, mapper) — concurrent map with at most n tasks in flight
// Preserves encounter order; uses virtual threads for concurrency
List<String> results = Stream.of("url1", "url2", "url3", "url4")
    .gather(mapConcurrent(4, url -> fetch(url))) // 4 concurrent fetches
    .toList();

// windowFixed — batch processing
// Perfect for: chunk API requests, process CSV in batches, paging

List<String> emails = getEmailList(); // 1000 emails
emails.stream()
    .gather(windowFixed(50))  // groups of 50
    .forEach(batch -> sendBulkEmail(batch)); // 20 API calls instead of 1000

// windowSliding — rolling window calculations
// Moving average of temperature readings
List<Double> temperatures = List.of(20.0, 22.0, 19.5, 23.0, 21.5, 24.0, 22.5);
List<Double> movingAvg = temperatures.stream()
    .gather(windowSliding(3))
    .map(window -> window.stream()
        .mapToDouble(Double::doubleValue)
        .average()
        .orElse(0.0))
    .toList();
System.out.println(movingAvg);
// [20.5, 21.5, 21.33, 22.83, 22.67]

// Consecutive duplicate groups (run-length encoding)
// windowSliding(2) + filter to find where consecutive elements differ
List<Integer> data = List.of(1, 1, 2, 2, 2, 3, 1, 1);
// Group consecutive identical elements using a custom gatherer (see below)

// windowFixed in parallel stream — note: parallel may produce different groupings
List<List<Integer>> parallelWindows = IntStream.rangeClosed(1, 100)
    .boxed()
    .parallel()
    .gather(windowFixed(10))
    .toList(); // may have windows from different threads — order not guaranteed

// scan — prefix sums, running min/max, cumulative products

// Running maximum
List<Integer> nums = List.of(3, 1, 4, 1, 5, 9, 2, 6);
List<Integer> runningMax = nums.stream()
    .gather(scan(() -> Integer.MIN_VALUE, Math::max))
    .toList();
System.out.println(runningMax); // [3, 3, 4, 4, 5, 9, 9, 9]

// Running concatenation
List<String> words = List.of("the", "quick", "brown", "fox");
List<String> sentences = words.stream()
    .gather(scan(() -> "", (acc, w) -> acc.isEmpty() ? w : acc + " " + w))
    .toList();
System.out.println(sentences);
// [the, the quick, the quick brown, the quick brown fox]

// fold — like reduce but always produces exactly one element in the output stream
// Useful when you want a single accumulated value at the end of a pipeline stage
Optional<String> combined = Stream.of("a", "b", "c")
    .gather(fold(() -> new StringBuilder(),
               (sb, s) -> { sb.append(s); return sb; }))
    .map(StringBuilder::toString)
    .findFirst();
System.out.println(combined.orElse("")); // "abc"

// Practical: running balance for bank transactions
record Transaction(String type, double amount) {}
List<Transaction> txns = List.of(
    new Transaction("credit", 1000),
    new Transaction("debit",  200),
    new Transaction("credit", 500),
    new Transaction("debit",  150)
);
List<Double> balance = txns.stream()
    .map(t -> t.type().equals("credit") ? t.amount() : -t.amount())
    .gather(scan(() -> 0.0, Double::sum))
    .toList();
System.out.println(balance); // [1000.0, 800.0, 1300.0, 1150.0]

import java.util.stream.*;
import java.util.function.*;

// Gatherer<T, A, R> — input T, accumulator/state A, output R
// Components:
//   initializer()  — creates the mutable state A
//   integrator()   — processes each element T, updates state A, emits R via Downstream
//   combiner()     — merges two states for parallel operation (optional)
//   finisher()     — called at stream end to emit final state (optional)

// Example 1: distinct by key (like distinctByKey from common libraries)
static <T, K> Gatherer<T, ?, T> distinctByKey(Function<T, K> keyExtractor) {
    return Gatherer.ofSequential(
        HashSet::new,                                  // initializer: empty set of seen keys
        Gatherer.Integrator.ofGreedy((seen, element, downstream) -> {
            K key = keyExtractor.apply(element);
            if (seen.add(key)) {                       // if key not seen before
                downstream.push(element);              // emit the element
            }
            return true;                               // continue processing
        })
    );
}

// Usage
record Person(String name, String city) {}
List<Person> people = List.of(
    new Person("Alice", "NYC"),
    new Person("Bob",   "LA"),
    new Person("Carol", "NYC"),
    new Person("Dave",  "LA")
);
List<Person> firstPerCity = people.stream()
    .gather(distinctByKey(Person::city))
    .toList();
System.out.println(firstPerCity);
// [Person[name=Alice, city=NYC], Person[name=Bob, city=LA]]

// Example 2: take while consecutive (emit elements while condition holds on adjacent pair)
static Gatherer<Integer, ?, Integer> takeWhileIncreasing() {
    return Gatherer.ofSequential(
        () -> new int[]{Integer.MIN_VALUE}, // state: last seen value
        Gatherer.Integrator.of((state, element, downstream) -> {
            if (element > state[0]) {
                state[0] = element;
                return downstream.push(element); // continue if downstream wants more
            }
            return false; // stop — sequence no longer increasing
        })
    );
}

Stream.of(1, 3, 5, 4, 7, 9)
    .gather(takeWhileIncreasing())
    .forEach(System.out::print); // 135 — stops at 4 (not > 5)

// Gatherers can be composed with andThen()
// Result: elements go through first gatherer, then second

// Sliding window of running sums
Gatherer<Integer, ?, List<Integer>> windowedScan =
    Gatherers.scan(() -> 0, Integer::sum)
             .andThen(Gatherers.windowSliding(3));

List<List<Integer>> result = Stream.of(1, 2, 3, 4, 5)
    .gather(windowedScan)
    .toList();
// scan first: [1, 3, 6, 10, 15]
// windowSliding(3): [[1,3,6], [3,6,10], [6,10,15]]
System.out.println(result);

// Chaining in a stream pipeline
List<String> processed = employeeStream
    .gather(distinctByKey(Employee::department)) // first by department
    .gather(Gatherers.windowFixed(3))            // batch into groups of 3
    .filter(batch -> batch.size() == 3)          // only full batches
    .flatMap(Collection::stream)
    .map(Employee::name)
    .toList();

// Parallel-compatible gatherer must implement combiner()
// Sequential-only gatherer: Gatherer.ofSequential(...)
// Parallel-compatible:      Gatherer.of(..., combiner, ...)
// Without combiner, parallel stream falls back to sequential for gather step