Stream gatherers: custom intermediate operations

import java.util.List;
import java.util.stream.Gatherers;
import java.util.stream.IntStream;
import java.util.stream.Stream;

class Windows {
    static double average(List<Integer> window) {
        return window.stream().mapToInt(Integer::intValue).average().orElseThrow();
    }

    static void main() {
        // Fixed windows: groups of 3, the last one shorter.
        IO.println("windowFixed(3):   " + IntStream.rangeClosed(1, 7).boxed().gather(Gatherers.windowFixed(3)).toList());
        // Sliding windows: each window drops its oldest element and adds the next one.
        IO.println("windowSliding(3): " + IntStream.rangeClosed(1, 7).boxed().gather(Gatherers.windowSliding(3)).toList());

        // Daily maximum temperatures in °C over ten days (made-up readings), and their 3-day moving average.
        List<Integer> maxima = List.of(38, 40, 41, 39, 43, 44, 42, 40, 37, 36);
        List<String> moving = maxima.stream()
                .gather(Gatherers.windowSliding(3))
                .map(w -> "%.1f".formatted(average(w)))
                .toList();
        IO.println("3-day averages:   " + moving);

        // The first 3 days whose average reaches 43 °C: findFirst stops the windows there.
        IO.println("first hot spell:  " + maxima.stream()
                .gather(Gatherers.windowSliding(3))
                .filter(w -> average(w) >= 43)
                .findFirst()
                .orElseThrow());

        // Fewer elements than the window size: one window holding all of them, not none.
        IO.println("2 days, window 3: " + Stream.of(38, 40).gather(Gatherers.windowSliding(3)).toList());

        // Windows are unmodifiable lists.
        List<Integer> first = maxima.stream().gather(Gatherers.windowFixed(3)).findFirst().orElseThrow();
        try {
            first.add(45);
        } catch (UnsupportedOperationException e) {
            IO.println("adding to a window: " + e);
        }

        // A gatherer can end an infinite stream: windows of 3 from 0, 1, 2, ... and only the first two.
        IO.println("infinite source:  " + Stream.iterate(0, i -> i + 1).gather(Gatherers.windowFixed(3)).limit(2).toList());
    }
}

from session import run, version

version()
run("""
java Windows.java
""")

# openjdk version "25.0.4.1" LTS
$ java Windows.java
windowFixed(3):   [[1, 2, 3], [4, 5, 6], [7]]
windowSliding(3): [[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6], [5, 6, 7]]
3-day averages:   [39.7, 40.0, 41.0, 42.0, 43.0, 42.0, 39.7, 37.7]
first hot spell:  [43, 44, 42]
2 days, window 3: [[38, 40]]
adding to a window: java.lang.UnsupportedOperationException
infinite source:  [[0, 1, 2], [3, 4, 5]]

import java.util.List;
import java.util.stream.Gatherers;

class Scans {
    static void main() {
        // A wallet: money added (+) and spent (-). scan emits the balance after each change.
        List<Integer> changes = List.of(500, -120, -80, 1000, -650);
        IO.println("balances: " + changes.stream().gather(Gatherers.scan(() -> 0, Integer::sum)).toList());

        // fold emits one result at the end. Its function runs in order, so it may depend on order:
        // the digits 1 to 9 become the number 123456789.
        List<Integer> digits = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9);
        IO.println("fold:                 " + digits.stream()
                .gather(Gatherers.fold(() -> 0, (number, d) -> number * 10 + d))
                .findFirst().orElseThrow());

        // The same function in reduce is wrong on a parallel stream: reduce needs an associative function,
        // and number * 10 + d is not, so joining the partial results of the parts gives another number.
        IO.println("reduce, sequential:   " + digits.stream().reduce(0, (number, d) -> number * 10 + d));
        IO.println("reduce, parallel:     " + digits.parallelStream().reduce(0, (number, d) -> number * 10 + d));

        // fold has no combiner, so a parallel stream runs it on one thread, in order, and the answer stays right.
        IO.println("fold, parallel:       " + digits.parallelStream()
                .gather(Gatherers.fold(() -> 0, (number, d) -> number * 10 + d))
                .findFirst().orElseThrow());
    }
}

from session import run, version

version()
run("""
java Scans.java
""")

# openjdk version "25.0.4.1" LTS
$ java Scans.java
balances: [500, 380, 300, 1300, 650]
fold:                 123456789
reduce, sequential:   123456789
reduce, parallel:     2259
fold, parallel:       123456789

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Gatherers;
import java.util.stream.IntStream;

class Concurrent {
    static final AtomicInteger running = new AtomicInteger();
    static final AtomicInteger peak = new AtomicInteger();

    /** Stands in for a slow pricing service: order n takes (13 - n) × 20 ms, so the later orders finish first. */
    static String price(int order) {
        peak.accumulateAndGet(running.incrementAndGet(), Math::max);
        try {
            if (order == 99) throw new IllegalStateException("pricing service down for order " + order);
            Thread.sleep((13 - order) * 20L);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        } finally {
            running.decrementAndGet();
        }
        return "order " + order + (Thread.currentThread().isVirtual() ? " on a virtual thread" : " on a platform thread");
    }

    static long millisSince(long start) {
        return (System.nanoTime() - start) / 1_000_000;
    }

    static void main() {
        long start = System.nanoTime();
        List<String> oneByOne = IntStream.rangeClosed(1, 12).boxed().map(Concurrent::price).toList();
        long mapMillis = millisSince(start);
        int mapPeak = peak.getAndSet(0);

        start = System.nanoTime();
        List<String> fourAtATime = IntStream.rangeClosed(1, 12).boxed()
                .gather(Gatherers.mapConcurrent(4, Concurrent::price))
                .toList();
        long concurrentMillis = millisSince(start);

        IO.println("map:              at most " + mapPeak + " call at a time, " + oneByOne.getFirst());
        IO.println("mapConcurrent(4): at most " + peak.get() + " calls at a time, " + fourAtATime.getFirst());
        IO.println("every call on a virtual thread: " + fourAtATime.stream().allMatch(s -> s.endsWith("virtual thread")));
        IO.println("results in the order of the input: " + oneByOne.equals(fourAtATime.stream()
                .map(s -> s.replace("virtual", "platform")).toList()));
        IO.println("map took about %,d ms, mapConcurrent(4) about %,d ms".formatted(
                Math.round(mapMillis / 100.0) * 100, Math.round(concurrentMillis / 100.0) * 100));

        // A call that fails: its exception comes out of the terminal operation, and the calls still running are cancelled.
        try {
            IntStream.of(1, 2, 99, 4).boxed().gather(Gatherers.mapConcurrent(4, Concurrent::price)).toList();
        } catch (RuntimeException e) {
            IO.println("a failing call: " + e);
        }
    }
}

from session import machine, run, version

version()
machine()
run("""
java Concurrent.java
""")

# openjdk version "25.0.4.1" LTS
# measured on: Apple M5 Pro, 18 cores
$ java Concurrent.java
map:              at most 1 call at a time, order 1 on a platform thread
mapConcurrent(4): at most 4 calls at a time, order 1 on a virtual thread
every call on a virtual thread: true
results in the order of the input: true
map took about 1,600 ms, mapConcurrent(4) about 500 ms
a failing call: java.lang.IllegalStateException: pricing service down for order 99

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.function.Function;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

class Custom {
    /** Keeps the first element for each key and drops later ones with the same key. */
    static <T, K> Gatherer<T, ?, T> distinctBy(Function<? super T, ? extends K> key) {
        return Gatherer.ofSequential(
                HashSet<K>::new,                                         // initializer: the keys seen so far
                Gatherer.Integrator.ofGreedy((seen, element, downstream) ->
                        // A new key: pass the element on. A key seen before: drop it and ask for more.
                        seen.add(key.apply(element)) ? downstream.push(element) : true));
    }

    /** Passes amounts on while their running total stays within the budget, then stops the stream. */
    static Gatherer<Integer, ?, Integer> withinBudget(int budget) {
        class Spent {
            int total;
        }
        return Gatherer.ofSequential(
                Spent::new,
                (spent, amount, downstream) -> {          // not greedy: it may return false
                    if (spent.total + amount > budget) return false;   // no more elements, please
                    spent.total += amount;
                    return downstream.push(amount);
                });
    }

    /** Packs parcel weights into batches of at most maxKg, in order; a parcel heavier than maxKg travels alone. */
    static Gatherer<Integer, ?, List<Integer>> batches(int maxKg) {
        class Batch {
            final List<Integer> parcels = new ArrayList<>();
            int kg;
        }
        return Gatherer.ofSequential(
                Batch::new,
                Gatherer.Integrator.ofGreedy((batch, parcel, downstream) -> {
                    boolean more = true;
                    if (!batch.parcels.isEmpty() && batch.kg + parcel > maxKg) {   // the parcel does not fit: close the batch
                        more = downstream.push(List.copyOf(batch.parcels));
                        batch.parcels.clear();
                        batch.kg = 0;
                    }
                    batch.parcels.add(parcel);
                    batch.kg += parcel;
                    return more;
                }),
                (batch, downstream) -> {                              // finisher: the last, unfinished batch
                    if (!batch.parcels.isEmpty() && !downstream.isRejecting()) downstream.push(List.copyOf(batch.parcels));
                });
    }

    record SignUp(String name, String email) {}

    static void main() {
        List<SignUp> signUps = List.of(
                new SignUp("Asha", "asha@example.com"), new SignUp("Ravi", "ravi@example.com"),
                new SignUp("Asha R", "ASHA@example.com"), new SignUp("Meena", "meena@example.com"),
                new SignUp("Ravi K", "ravi@example.com"));
        Function<SignUp, String> email = s -> s.email().toLowerCase(Locale.ROOT);
        IO.println("distinctBy, sequential: " + signUps.stream().gather(distinctBy(email)).map(SignUp::name).toList());
        IO.println("distinctBy, parallel:   " + signUps.parallelStream().gather(distinctBy(email)).map(SignUp::name).toList());

        IO.println("withinBudget(1000):         " + Stream.of(400, 250, 300, 150, 500).gather(withinBudget(1000)).toList());
        IO.println("withinBudget, endless list: " + Stream.iterate(100, a -> a + 100).gather(withinBudget(1000)).toList());

        List<Integer> parcels = List.of(4, 3, 2, 5, 1, 6, 9);
        IO.println("batches(8):          " + parcels.stream().gather(batches(8)).toList());
        IO.println("batches(8), limit 1: " + parcels.stream().gather(batches(8)).limit(1).toList());
        IO.println("no parcels:          " + Stream.<Integer>empty().gather(batches(8)).toList());

        // Gatherers compose: drop repeated sign-ups, then pack the rest into pairs.
        IO.println("andThen: " + signUps.stream()
                .gather(distinctBy(email).andThen(java.util.stream.Gatherers.windowFixed(2)))
                .map(pair -> pair.stream().map(SignUp::name).toList())
                .toList());
    }
}

from session import run, version

version()
run("""
java Custom.java
""")

# openjdk version "25.0.4.1" LTS
$ java Custom.java
distinctBy, sequential: [Asha, Ravi, Meena]
distinctBy, parallel:   [Asha, Ravi, Meena]
withinBudget(1000):         [400, 250, 300]
withinBudget, endless list: [100, 200, 300, 400]
batches(8):          [[4, 3], [2, 5, 1], [6], [9]]
batches(8), limit 1: [[4, 3]]
no parcels:          []
andThen: [[Asha, Ravi], [Meena]]

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.stream.Gatherer;
import java.util.stream.IntStream;
import java.util.stream.Stream;

class Runs {
    /** Groups consecutive equal elements (by equals, null allowed) into unmodifiable lists, in order. */
    static <T> Gatherer<T, ?, List<T>> runs() {
        class Run {
            final List<T> items = new ArrayList<>();
        }
        return Gatherer.ofSequential(
                Run::new,
                Gatherer.Integrator.ofGreedy((run, element, downstream) -> {
                    boolean more = true;
                    if (!run.items.isEmpty() && !Objects.equals(run.items.getLast(), element)) {
                        more = downstream.push(Collections.unmodifiableList(new ArrayList<>(run.items)));
                        run.items.clear();
                    }
                    run.items.add(element);
                    return more;
                }),
                (run, downstream) -> {
                    if (!run.items.isEmpty() && !downstream.isRejecting()) {
                        downstream.push(Collections.unmodifiableList(new ArrayList<>(run.items)));
                    }
                });
    }

    /** Run-length encoding built on runs(): "AAABBC" becomes "3A2B1C". */
    static String encode(String text) {
        return text.codePoints().mapToObj(Character::toString)
                .gather(runs())
                .map(run -> run.size() + run.getFirst())
                .collect(Collectors.joining());
    }

    static void check(String what, boolean ok) {
        if (!ok) throw new AssertionError("FAILED: " + what);
        IO.println("ok   " + what);
    }

    static void main() {
        IO.println("runs of 1, 1, 2, 3, 3, 3, 1: " + Stream.of(1, 1, 2, 3, 3, 3, 1).gather(runs()).toList());
        IO.println("encode(\"AAABBC\"): " + encode("AAABBC"));

        check("runs in input order", Stream.of(1, 1, 2, 3, 3, 3, 1).gather(runs()).toList()
                .equals(List.of(List.of(1, 1), List.of(2), List.of(3, 3, 3), List.of(1))));
        check("no elements: no runs", Stream.empty().gather(runs()).toList().isEmpty());
        check("one element: one run", Stream.of("x").gather(runs()).toList().equals(List.of(List.of("x"))));
        check("all equal: one run", Stream.of(7, 7, 7, 7).gather(runs()).toList().equals(List.of(List.of(7, 7, 7, 7))));
        check("nulls form a run too", Stream.of(null, null, "x").gather(runs()).toList()
                .equals(List.of(Arrays.asList(null, null), List.of("x"))));
        check("emoji stay whole: \"👍👍a\" encodes as 2👍1a", encode("👍👍a").equals("2👍1a"));
        check("an endless source stops after 2 runs", Stream.iterate(0, i -> i + 1).map(i -> i / 3)
                .gather(runs()).limit(2).toList().equals(List.of(List.of(0, 0, 0), List.of(1, 1, 1))));

        List<Integer> big = IntStream.range(0, 300_000).map(i -> i / 7).boxed().toList();
        check("parallel stream: the same runs as sequential",
                big.parallelStream().gather(runs()).toList().equals(big.stream().gather(runs()).toList()));
        check("each run is unmodifiable", throwsUnsupported(() -> Stream.of(1, 1).gather(runs()).toList().getFirst().add(2)));
        IO.println("all checks passed");
    }

    static boolean throwsUnsupported(Runnable r) {
        try {
            r.run();
            return false;
        } catch (UnsupportedOperationException e) {
            return true;
        }
    }
}

from session import run, version

version()
run("""
java Runs.java
""")

# openjdk version "25.0.4.1" LTS
$ java Runs.java
runs of 1, 1, 2, 3, 3, 3, 1: [[1, 1], [2], [3, 3, 3], [1]]
encode("AAABBC"): 3A2B1C
ok   runs in input order
ok   no elements: no runs
ok   one element: one run
ok   all equal: one run
ok   nulls form a run too
ok   emoji stay whole: "👍👍a" encodes as 2👍1a
ok   an endless source stops after 2 runs
ok   parallel stream: the same runs as sequential
ok   each run is unmodifiable
all checks passed