import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;

/**
 * Controlled admission-order experiment for the capacity topic. It follows the
 * jdk-17.0.12+7 ThreadPoolExecutor execute() branches described in the article:
 * core worker first, then queue offer, then a non-core worker, then rejection.
 * Latches and observable pool state provide the evidence; no sleeps are used
 * for ordering. This is a mechanism experiment, not a performance benchmark.
 *
 * Run from the topic directory with JDK 17 or newer: java labs/PoolAdmission.java
 */
public class PoolAdmission {

    private static final List<String> FAILURES = new ArrayList<>();

    public static void main(String[] args) throws Exception {
        boundedQueueCase();
        unboundedQueueCase();
        shutdownCase();
        if (FAILURES.isEmpty()) {
            System.out.println("ALL CHECKS PASSED");
        } else {
            FAILURES.forEach(System.out::println);
            System.exit(1);
        }
    }

    /** core=1, max=2, queue=1: queue fills before a non-core worker starts. */
    static void boundedQueueCase() throws Exception {
        CountDownLatch running = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        AtomicInteger rejects = new AtomicInteger();
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.AbortPolicy());
        try {
            pool.execute(() -> await(release));          // occupies the core worker
            waitUntil("core worker started", () -> pool.getPoolSize() == 1);
            pool.execute(() -> await(release));          // goes to the queue
            waitUntil("task 2 queued", () -> pool.getQueue().size() == 1);
            pool.execute(() -> await(release));          // queue full -> non-core worker
            waitUntil("non-core worker started", () -> pool.getPoolSize() == 2);
            try {
                pool.execute(() -> await(release));      // queue full, max reached
            } catch (RejectedExecutionException e) {
                rejects.incrementAndGet();
            }
            check("task 4 rejected exactly once", rejects.get() == 1);
            check("pool size is max 2", pool.getPoolSize() == 2);
            check("queue still holds 1 task", pool.getQueue().size() == 1);
        } finally {
            release.countDown();
            running.countDown();
            pool.shutdown();
            pool.awaitTermination(5, TimeUnit.SECONDS);
        }
        System.out.println("bounded: pool=2 queue=1 rejects=" + rejects.get());
    }

    /** Unbounded queue: offer succeeds, so maxPoolSize never engages. */
    static void unboundedQueueCase() throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 0, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(), new ThreadPoolExecutor.AbortPolicy());
        try {
            pool.execute(() -> await(release));
            pool.execute(() -> await(release));
            pool.execute(() -> await(release));
            waitUntil("two tasks queued", () -> pool.getQueue().size() == 2);
            check("no second worker without a bounded queue", pool.getPoolSize() == 1);
        } finally {
            release.countDown();
            pool.shutdown();
            pool.awaitTermination(5, TimeUnit.SECONDS);
        }
        System.out.println("unbounded: pool=1 queued=2 (max never engaged)");
    }

    /** Submitting after shutdown is rejected; shutdown racing offer needs re-check. */
    static void shutdownCase() {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.AbortPolicy());
        pool.shutdown();
        try {
            pool.execute(() -> { });
            FAILURES.add("submit after shutdown must be rejected");
        } catch (RejectedExecutionException expected) {
            System.out.println("shutdown: submit rejected as expected");
        }
    }

    static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    static void waitUntil(String what, BooleanSupplier condition) throws InterruptedException {
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
        while (System.nanoTime() < deadline) {
            if (condition.getAsBoolean()) return;
            Thread.sleep(5);
        }
        FAILURES.add("timeout waiting for: " + what);
    }

    static void check(String what, boolean pass) {
        if (!pass) FAILURES.add("check failed: " + what);
    }
}