rokon12

record TrendSignal(double price, String trend) {}

public static Gatherer<Double, ?, TrendSignal> trendDetection(int lookback) {
    return Gatherer.of(
        () -> new Object() { 
            List<Double> history = new ArrayList<>();  // 1️⃣ Price history buffer
        },
        (state, price, downstream) -> {
            state.history.add(price);                  // 2️⃣ Add to history
            if (state.history.size() >= lookback) {    // 3️⃣ Enough data?

rokon12

public static Gatherer<Double, ?, Double> movingAverage(int windowSize) {
    return Gatherer.of(
        LinkedList::new,  // 1️⃣ Use LinkedList for efficient add/remove
        (window, price, downstream) -> {
            window.add(price);                    // 2️⃣ Add new price to window
            if (window.size() > windowSize) {     // 3️⃣ Window too big?
                window.removeFirst();             // 4️⃣ Remove oldest price
            }
            if (window.size() == windowSize) {    // 5️⃣ Window full?
                double avg = window.stream()

rokon12

public static <T> Gatherer<T, ?, List<T>> 
        batchByCondition(Predicate<List<T>> batchComplete) {
    return Gatherer.of(
        ArrayList::new,  // 1️⃣ Current batch being built
        (batch, element, downstream) -> {
            batch.add(element);                        // 2️⃣ Add to current batch
            if (batchComplete.test(batch)) {           // 3️⃣ Check if batch is "full"
                downstream.push(new ArrayList<>(batch)); // 4️⃣ Send copy downstream
                batch.clear();                         // 5️⃣ Start fresh batch
            }

rokon12

public static <T, R extends Comparable<R>> Gatherer<T, ?, T> 
        takeWhileIncreasing(Function<T, R> valueExtractor) {
    return Gatherer.of(
        () -> new Object() { R lastValue = null; },  // 1️⃣ Anonymous class for state
        (state, element, downstream) -> {
            R value = valueExtractor.apply(element);  // 2️⃣ Extract comparable value
            if (state.lastValue == null ||            // 3️⃣ First element OR
                value.compareTo(state.lastValue) > 0) {  // value is increasing
                state.lastValue = value;              // 4️⃣ Update state
                return downstream.push(element);       // 5️⃣ Pass element along

rokon12

public static <T, K> Gatherer<T, ?, T> distinctByKey(Function<T, K> keyExtractor) {
    return Gatherer.of(
        HashSet::new,  // 1️⃣ State: Set to track seen keys
        (state, element, downstream) -> {
            K key = keyExtractor.apply(element);    // 2️⃣ Extract the key
            if (state.add(key)) {                   // 3️⃣ If key is new (add returns true)
                return downstream.push(element);     // 4️⃣ Pass element downstream
            }
            return true;                            // 5️⃣ Continue processing
        }

rokon12

class Statistics {
    int count = 0;
    int sum = 0;
    int min = Integer.MAX_VALUE;
    int max = Integer.MIN_VALUE;
    
    Statistics add(Integer value) {
        count++;                        // 1️⃣ Track how many elements
        sum += value;                   // 2️⃣ Running total
        min = Math.min(min, value);     // 3️⃣ Track minimum

rokon12

List<Long> runningProduct = values.stream()
    .gather(Gatherers.scan(
        () -> 1L,                      // 1️⃣ Start with identity for multiplication
        (acc, val) -> acc * val        // 2️⃣ Multiply accumulated value by current
    ))
    .toList();

System.out.println("Running product: " + runningProduct);
// Output: [1, 1, 5, 15, 120, 240, 2160, 8640, 60480, 362880]
//         ↑   ↑   ↑    ↑     ↑ 

rokon12

List<Integer> runningMax = values.stream()
    .gather(Gatherers.scan(
        () -> Integer.MIN_VALUE,  // 1️⃣ Start with smallest possible value
        Integer::max              // 2️⃣ Keep the maximum at each step
    ))
    .toList();

System.out.println("Running maximum: " + runningMax);
// Output: [-2147483648, 1, 5, 5, 8, 8, 9, 9, 9, 9]
//          ↑ initial      ↑ 5>1  ↑ 8>5  ↑ 9>8

rokon12

record PeakValley(String type, double value, int index) {}

public static Gatherer<Double, ?, PeakValley> peakValleyDetection() {
    return Gatherer.of(
        () -> new Object() {
            Double prev = null;      // 1️⃣ Previous value
            Double current = null;   // 2️⃣ Current value
            int index = 0;          // 3️⃣ Track position
        },
        (state, next, downstream) -> {

rokon12

record TrendSignal(double price, String trend) {}

public static Gatherer<Double, ?, TrendSignal> trendDetection(int lookback) {
    return Gatherer.of(
        () -> new Object() { 
            List<Double> history = new ArrayList<>();  // 1️⃣ Price history buffer
        },
        (state, price, downstream) -> {
            state.history.add(price);                  // 2️⃣ Add to history
            if (state.history.size() >= lookback) {    // 3️⃣ Enough data?