import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 第 5 课的可运行演示与内置验证，适用于 JDK 17/21。
 * 运行：java reference/ThreadPoolRoutingDemo.java
 * 用有超时的信号控制任务结束，避免依赖 sleep 猜测调度顺序。
 * 四项验证：有界队列扩容与拒绝、无界队列、调用者执行、Future 异常。
 */
public final class ThreadPoolRoutingDemo {
    private static final long WAIT_SECONDS = 10;

    private ThreadPoolRoutingDemo() { }

    public static void main(String[] args) throws Exception {
        verifyBoundedQueue();
        verifyUnboundedQueue();
        verifyCallerRuns();
        verifyFailedFuture();
        System.out.println("全部 4 组验证通过，所有线程池已终止。");
    }

    private static ThreadPoolExecutor createPool(int core, int maximum,
            BlockingQueue<Runnable> queue, RejectedExecutionHandler handler) {
        ThreadFactory defaultFactory = Executors.defaultThreadFactory();
        AtomicInteger sequence = new AtomicInteger();
        ThreadFactory namedFactory = task -> {
            Thread thread = defaultFactory.newThread(task);
            thread.setName("lesson5-worker-" + sequence.incrementAndGet());
            return thread;
        };
        return new ThreadPoolExecutor(core, maximum, 30, TimeUnit.SECONDS,
                queue, namedFactory, handler);
    }

    private static void verifyBoundedQueue() throws Exception {
        ThreadPoolExecutor pool = createPool(2, 4, new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.AbortPolicy());
        CountDownLatch started = new CountDownLatch(4);
        CountDownLatch release = new CountDownLatch(1);
        List<Future<Integer>> tasks = new ArrayList<>();
        try {
            for (int id = 1; id <= 6; id++) {
                tasks.add(submitHeldTask(pool, id, started, release));
                int expectedWorkers = id <= 2 ? id : id <= 4 ? 2 : id - 2;
                int expectedQueued = Math.min(Math.max(id - 2, 0), 2);
                verifySize(pool, expectedWorkers, expectedQueued);
                System.out.printf("任务 %d 提交后：线程数=%d，排队数=%d%n",
                        id, pool.getPoolSize(), pool.getQueue().size());
            }
            awaitSignal(started, "4 个工作线程启动");
            expectRejected(pool, "第 7 个任务因饱和被拒绝");
            release.countDown();
            verifyResults(tasks);
        } finally {
            release.countDown();
            stopPool(pool);
        }
        expectRejected(pool, "关闭后的任务也被拒绝");
    }

    private static void verifyUnboundedQueue() throws Exception {
        ThreadPoolExecutor pool = createPool(2, 4, new LinkedBlockingQueue<>(),
                new ThreadPoolExecutor.AbortPolicy());
        CountDownLatch started = new CountDownLatch(2);
        CountDownLatch release = new CountDownLatch(1);
        List<Future<Integer>> tasks = new ArrayList<>();
        try {
            for (int id = 1; id <= 7; id++) {
                tasks.add(submitHeldTask(pool, id, started, release));
            }
            awaitSignal(started, "2 个工作线程启动");
            verifySize(pool, 2, 5);
            System.out.println("无界队列：提交 7 个任务后仍是 2 个线程、5 个排队。");
            release.countDown();
            verifyResults(tasks);
        } finally {
            release.countDown();
            stopPool(pool);
        }
    }

    private static void verifyCallerRuns() throws Exception {
        ThreadPoolExecutor pool = createPool(1, 1, new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.CallerRunsPolicy());
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        List<Future<Integer>> tasks = new ArrayList<>();
        AtomicReference<Thread> executingThread = new AtomicReference<>();
        try {
            tasks.add(submitHeldTask(pool, 1, started, release));
            awaitSignal(started, "唯一工作线程启动");
            tasks.add(submitHeldTask(pool, 2, started, release));
            Thread submittingThread = Thread.currentThread();
            pool.execute(() -> executingThread.set(Thread.currentThread()));
            check(executingThread.get() == submittingThread, "未由提交线程执行");
            System.out.println("CallerRunsPolicy：饱和时，第 3 个短任务由提交线程执行。");
            release.countDown();
            verifyResults(tasks);
        } finally {
            release.countDown();
            stopPool(pool);
        }
        AtomicInteger executions = new AtomicInteger();
        pool.execute(executions::incrementAndGet);
        check(executions.get() == 0, "关闭后不应执行 CallerRunsPolicy 任务");
        System.out.println("CallerRunsPolicy：关闭后的新任务被丢弃。");
    }

    private static void verifyFailedFuture() throws Exception {
        ThreadPoolExecutor pool = createPool(1, 1, new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.AbortPolicy());
        try {
            Future<Integer> failed = pool.submit(() -> {
                throw new IllegalStateException("演示用任务异常");
            });
            try {
                failed.get(WAIT_SECONDS, TimeUnit.SECONDS);
                throw new AssertionError("异常任务不应返回正常结果");
            } catch (ExecutionException failure) {
                check(failure.getCause() instanceof IllegalStateException,
                        "Future 未保留预期的任务异常");
                System.out.println("submit：通过 Future.get 观察到任务原始异常。");
            }
        } finally {
            stopPool(pool);
        }
    }

    private static Future<Integer> submitHeldTask(ThreadPoolExecutor pool, int id,
            CountDownLatch started, CountDownLatch release) {
        return pool.submit(() -> {
            started.countDown();
            awaitSignal(release, "释放任务 " + id);
            return id;
        });
    }

    private static void awaitSignal(CountDownLatch signal, String description)
            throws InterruptedException {
        if (!signal.await(WAIT_SECONDS, TimeUnit.SECONDS)) {
            throw new AssertionError("等待超时：" + description);
        }
    }

    private static void verifySize(ThreadPoolExecutor pool, int workers, int queued) {
        check(pool.getPoolSize() == workers, "工作线程数不符合预期");
        check(pool.getQueue().size() == queued, "排队任务数不符合预期");
    }

    private static void verifyResults(List<Future<Integer>> tasks) throws Exception {
        for (int index = 0; index < tasks.size(); index++) {
            int result = tasks.get(index).get(WAIT_SECONDS, TimeUnit.SECONDS);
            check(result == index + 1, "任务结果与提交编号不一致");
        }
    }

    private static void expectRejected(ThreadPoolExecutor pool, String description) {
        try {
            pool.execute(() -> { throw new AssertionError("被拒绝的任务不应执行"); });
            throw new AssertionError("未触发预期拒绝：" + description);
        } catch (RejectedExecutionException expected) {
            System.out.println("AbortPolicy：" + description + "。");
        }
    }

    private static void stopPool(ThreadPoolExecutor pool) throws InterruptedException {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(WAIT_SECONDS, TimeUnit.SECONDS)) {
                cancelPending(pool.shutdownNow());
                if (!pool.awaitTermination(WAIT_SECONDS, TimeUnit.SECONDS)) {
                    throw new IllegalStateException("线程池未按时终止");
                }
            }
        } catch (InterruptedException interrupted) {
            cancelPending(pool.shutdownNow());
            Thread.currentThread().interrupt();
            throw interrupted;
        }
    }

    private static void cancelPending(List<Runnable> pending) {
        for (Runnable task : pending) {
            if (task instanceof Future<?> future) {
                future.cancel(false);
            }
        }
    }

    private static void check(boolean condition, String message) {
        if (!condition) {
            throw new AssertionError(message);
        }
    }
}