import java.util.ArrayList;
import java.util.List;
import java.util.function.Predicate;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

/**
 * Writing a Gatherer from scratch. Gatherer<T, A, R> is four composable pieces:
 * initializer (private mutable state A), integrator (per-element: read/write state A, push 0+ R
 * downstream, return whether to keep going), combiner (merge two states for parallel streams --
 * omit it and the gatherer just runs sequentially), finisher (flush anything left when the
 * upstream is exhausted). Two custom gatherers here: dedupeConsecutive (stateful, one-in/0-or-1
 * out) and takeUntil (short-circuiting -- something no Collector can do, because a Collector
 * never gets a say in whether the source stops producing).
 * Chapter: docs/16-gatherers.md
 */
public class CustomGatherer {

    public static void main(String[] args) {
        dedupeConsecutiveDemo();
        takeUntilStopsTheUpstream();
    }

    /** Collapses consecutive equal elements, like Unix `uniq` -- keeps non-adjacent duplicates. */
    static <T> Gatherer<T, ?, T> dedupeConsecutive() {
        return Gatherer.ofSequential(
                // initializer: a single-slot box holding "last element pushed downstream", or
                // null meaning "nothing pushed yet". A List<Object> of size 1 doubles as a
                // nullable box without a wrapper class.
                () -> new ArrayList<Object>(List.of()),
                // integrator: push only when this element differs from the last one we pushed.
                (state, element, downstream) -> {
                    boolean isFirst = state.isEmpty();
                    boolean changed = isFirst || !state.get(0).equals(element);
                    if (changed) {
                        if (isFirst) {
                            state.add(element);
                        } else {
                            state.set(0, element);
                        }
                        return downstream.push(element);
                    }
                    return true; // swallowed a duplicate; keep pulling from upstream
                });
    }

    static void dedupeConsecutiveDemo() {
        List<String> deduped = Stream.of("a", "a", "b", "b", "b", "a", "c", "c")
                .gather(CustomGatherer.<String>dedupeConsecutive())
                .toList();
        System.out.println("dedupeConsecutive of a,a,b,b,b,a,c,c = " + deduped);
        Check.that(deduped.equals(List.of("a", "b", "a", "c")),
                "consecutive runs collapse to [a, b, a, c]; the second 'a' survives because it isn't adjacent to the first");
    }

    /**
     * takeWhile(p) stops BEFORE the first failing element. takeUntil(p) stops AFTER it --
     * includes the element that fails the predicate, then no more. Not expressible with
     * takeWhile(p.negate()) plus one extra element without re-running the predicate; building it
     * as a Gatherer means integrator itself decides when to stop pulling from upstream.
     */
    static <T> Gatherer<T, ?, T> takeUntil(Predicate<? super T> stopAfter) {
        return Gatherer.ofSequential((state, element, downstream) -> {
            downstream.push(element);
            return !stopAfter.test(element); // false = stop; upstream is not pulled again
        });
    }

    static void takeUntilStopsTheUpstream() {
        List<Integer> upstreamSeen = new ArrayList<>();
        List<Integer> result = Stream.of(1, 2, 3, 4, 5, 6)
                .peek(upstreamSeen::add)
                .gather(takeUntil((Integer n) -> n == 3))
                .toList();

        System.out.println("takeUntil(n == 3) result = " + result + ", upstream elements actually pulled = " + upstreamSeen);
        Check.that(result.equals(List.of(1, 2, 3)), "takeUntil includes the element that satisfies the stop condition: [1, 2, 3]");
        Check.that(upstreamSeen.equals(List.of(1, 2, 3)),
                "the upstream peek() only saw 1, 2, 3 -- elements 4, 5, 6 were never pulled, this genuinely short-circuits");
    }

}