import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

/**
 * Gatherers.mapConcurrent(maxConcurrency, mapper): the one built-in gatherer whose whole point is
 * concurrency, not stream shape. It runs the mapper for each element on its own virtual thread (a
 * new one per element, not a fixed pool of virtual threads), bounded to at most `maxConcurrency`
 * in flight, and preserves encounter order in the output regardless of which finishes first.
 * Chapter: docs/16-gatherers.md
 */
public class MapConcurrentDemo {

    public static void main(String[] args) throws Exception {
        orderIsPreservedEvenWhenCompletionOrderIsNot();
        concurrencyIsBounded();
        runsOnVirtualThreadsNotThePlatformPool();
        aFailureWaitsForTheOthersRatherThanCancellingThem();
    }

    static void orderIsPreservedEvenWhenCompletionOrderIsNot() {
        // Task N sleeps for (5-N)*40ms, so task 1 finishes LAST and task 4 finishes FIRST -- yet
        // the output list comes back in input order, 1..4, because mapConcurrent buffers results
        // and only pushes downstream once every earlier slot is filled.
        List<Integer> results = Stream.of(1, 2, 3, 4)
                .gather(Gatherers.mapConcurrent(4, n -> {
                    sleep(Duration.ofMillis((5 - n) * 40L));
                    return n * 10;
                }))
                .toList();

        System.out.println("mapConcurrent(4, ...) results, completion order reversed: " + results);
        Check.that(results.equals(List.of(10, 20, 30, 40)),
                "mapConcurrent output preserves encounter order [10,20,30,40] even though task 1 finishes last");
    }

    static void concurrencyIsBounded() {
        // 8 tasks, each holding a permit for 60ms, capped at maxConcurrency=2. If the cap were
        // real only in name, all 8 would overlap and the peak-in-flight counter would read 8.
        AtomicInteger inFlight = new AtomicInteger(0);
        AtomicInteger peak = new AtomicInteger(0);

        List<Integer> results = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
                .gather(Gatherers.mapConcurrent(2, n -> {
                    int now = inFlight.incrementAndGet();
                    peak.updateAndGet(p -> Math.max(p, now));
                    sleep(Duration.ofMillis(60));
                    inFlight.decrementAndGet();
                    return n;
                }))
                .toList();

        System.out.println("mapConcurrent(2, ...) over 8 tasks, peak concurrent in-flight = " + peak.get());
        Check.that(results.size() == 8, "all 8 tasks completed");
        Check.that(peak.get() <= 2, "peak in-flight never exceeded maxConcurrency=2, measured peak=" + peak.get());
        Check.that(peak.get() == 2, "the cap is actually reached, not just respected -- measured peak=" + peak.get());
    }

    static void runsOnVirtualThreadsNotThePlatformPool() {
        // Each element's mapper runs on its own Thread.ofVirtual() thread, not a shared
        // ForkJoinPool.commonPool() worker the way Stream.parallel() would use.
        ConcurrentLinkedQueue<String> threadNames = new ConcurrentLinkedQueue<>();
        AtomicInteger distinctThreads = new AtomicInteger(0);

        Stream.of(1, 2, 3, 4, 5, 6)
                .gather(Gatherers.mapConcurrent(6, n -> {
                    Thread t = Thread.currentThread();
                    threadNames.add(t + " virtual=" + t.isVirtual());
                    return n;
                }))
                .toList();

        threadNames.forEach(System.out::println);
        boolean allVirtual = threadNames.stream().allMatch(s -> s.endsWith("virtual=true"));
        long distinct = threadNames.stream().distinct().count();
        Check.that(allVirtual, "every mapConcurrent worker thread reports isVirtual() == true");
        Check.that(distinct == 6, "6 elements produced 6 distinct virtual threads -- one per element, not a shared pool");
    }

    static void aFailureWaitsForTheOthersRatherThanCancellingThem() {
        // Easy to assume a fail-fast gatherer interrupts its siblings the moment one mapper
        // throws -- structured concurrency's StructuredTaskScope does exactly that. mapConcurrent
        // does NOT: it lets every already-started mapper run to completion (or timeout) before
        // the exception surfaces. Element 3 throws immediately here; elements 1, 2, 4, 5 each
        // sleep 900ms. If the failure were fail-fast we'd see it in well under 900ms -- instead
        // the whole call blocks for as long as the slowest sibling.
        Instant start = Instant.now();
        try {
            Stream.of(1, 2, 3, 4, 5)
                    .gather(Gatherers.mapConcurrent(5, n -> {
                        if (n == 3) {
                            throw new IllegalStateException("boom at element 3");
                        }
                        sleep(Duration.ofMillis(900));
                        return n;
                    }))
                    .toList();
            Check.that(false, "expected the failure to propagate");
        } catch (IllegalStateException e) {
            Duration elapsed = Duration.between(start, Instant.now());
            System.out.println("mapConcurrent propagated: " + e.getMessage() + " after " + elapsed.toMillis() + "ms");
            Check.that(elapsed.toMillis() >= 850,
                    "the failure only surfaced after " + elapsed.toMillis()
                            + "ms -- mapConcurrent waited for the other 4 in-flight tasks, it did not cancel them");
        }
    }

    static void sleep(Duration d) {
        try {
            Thread.sleep(d);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
}