import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 第8课教师示例：逐项验证课件里关于 CompletableFuture（可编排的异步结果）的行为说法。
 * 运行：java reference/CompletableFutureDemo.java（Java 17 或 21，不需要额外依赖）。
 * 全部通过时最后一行输出“全部通过”；任何一项不符会抛出 AssertionError。
 * 用 sleep 模拟远程查询，只访问内存；耗时断言留有余量，不是性能测试。
 */
public class CompletableFutureDemo {

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

    static void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("被中断", e);
        }
    }

    /** 课件第三节的写法：有界队列、线程有名字、满了让调用者自己执行。 */
    static ThreadPoolExecutor newQueryPool() {
        AtomicInteger seq = new AtomicInteger();
        return new ThreadPoolExecutor(3, 3, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100),
                r -> new Thread(r, "query-" + seq.incrementAndGet()),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = newQueryPool();
        try {
            whichThread(pool);
            parallelVersusSequential(pool);
            chaining(pool);
            exceptions(pool);
            allOfWithFailure(pool);
            cancelDoesNotInterrupt(pool);
            timeouts(pool);
        } finally {
            pool.shutdown();
            check(pool.awaitTermination(5, TimeUnit.SECONDS), "线程池应能按时关闭");
        }
        System.out.println("全部通过");
    }

    /** 第一、三节：不传线程池用公共池；传了就用自己的；公共池线程是守护线程。 */
    static void whichThread(ThreadPoolExecutor pool) {
        int cores = Runtime.getRuntime().availableProcessors();
        AtomicReference<Thread> commonThread = new AtomicReference<>();
        String commonName = CompletableFuture.supplyAsync(() -> {
            commonThread.set(Thread.currentThread());
            return Thread.currentThread().getName();
        }).join();
        String customName = CompletableFuture.supplyAsync(() -> Thread.currentThread().getName(), pool).join();
        System.out.println("处理器数=" + cores + "，公共池并行度=" + java.util.concurrent.ForkJoinPool.getCommonPoolParallelism()
                + "，默认线程=" + commonName + "，自定义线程=" + customName);
        if (java.util.concurrent.ForkJoinPool.getCommonPoolParallelism() >= 2) {
            check(commonName.startsWith("ForkJoinPool.commonPool-worker"), "不传线程池时应在公共池执行，实际 " + commonName);
            check(commonThread.get().isDaemon(), "公共池工作线程应是守护线程");
        }
        check(customName.startsWith("query-"), "传入线程池时应在该线程池执行，实际 " + customName);
        check(java.util.concurrent.ForkJoinPool.getCommonPoolParallelism() == Math.max(1, cores - 1),
                "未设系统属性时公共池并行度应为 max(1, 处理器数-1)");

        // 前一阶段已经完成时，非 async 的 thenApply 在调用方线程里执行（实现行为，演示观察）
        AtomicReference<String> applyThread = new AtomicReference<>();
        CompletableFuture.completedFuture("x").thenApply(v -> {
            applyThread.set(Thread.currentThread().getName());
            return v;
        }).join();
        check(applyThread.get().equals(Thread.currentThread().getName()),
                "已完成阶段上的 thenApply 应在调用方线程执行，实际 " + applyThread.get());
        System.out.println("A 由谁执行：通过（已完成阶段上的 thenApply 在 " + applyThread.get() + " 执行）");
    }

    /** 第一节：三个互不依赖的查询，串行要加起来，并行约等于最慢的那个。 */
    static void parallelVersusSequential(ThreadPoolExecutor pool) {
        long start = System.nanoTime();
        String a = queryApplicant();
        String b = queryMaterials();
        String c = queryRecords();
        long sequential = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);

        start = System.nanoTime();
        CompletableFuture<String> fa = CompletableFuture.supplyAsync(CompletableFutureDemo::queryApplicant, pool);
        CompletableFuture<String> fb = CompletableFuture.supplyAsync(CompletableFutureDemo::queryMaterials, pool);
        CompletableFuture<String> fc = CompletableFuture.supplyAsync(CompletableFutureDemo::queryRecords, pool);
        CompletableFuture.allOf(fa, fb, fc).join();
        String detail = fa.join() + "|" + fb.join() + "|" + fc.join();
        long parallel = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);

        check((a + "|" + b + "|" + c).equals(detail), "两种方式结果应相同");
        check(sequential >= 750, "串行应不少于 300+200+250=750ms，实际 " + sequential);
        check(parallel >= 300 && parallel < 600, "并行应约等于最慢的 300ms，实际 " + parallel);
        System.out.println("B 串行=" + sequential + "ms，并行=" + parallel + "ms：通过");
    }

    static String queryApplicant() { sleep(300); return "申请人"; }
    static String queryMaterials() { sleep(200); return "材料3份"; }
    static String queryRecords()   { sleep(250); return "审批2条"; }

    /** 第二节：thenApply 变换结果；thenCompose 接一个返回 CompletableFuture 的异步步骤并“摊平”；thenCombine 合并两个独立结果。 */
    static void chaining(ThreadPoolExecutor pool) {
        CompletableFuture<Integer> length = CompletableFuture.supplyAsync(() -> "A-001", pool)
                .thenApply(String::length);
        check(length.join() == 5, "thenApply 应把 \"A-001\" 变成 5");

        // 用 thenApply 接一个异步方法，得到“结果里套着结果”
        CompletableFuture<CompletableFuture<String>> nested = CompletableFuture.supplyAsync(() -> "A-001", pool)
                .thenApply(id -> queryOwnerAsync(id, pool));
        // 用 thenCompose 接，直接得到结果
        CompletableFuture<String> flat = CompletableFuture.supplyAsync(() -> "A-001", pool)
                .thenCompose(id -> queryOwnerAsync(id, pool));
        check(nested.join().join().equals("A-001的申请人") && flat.join().equals("A-001的申请人"),
                "thenApply 需要 join 两次，thenCompose 一次");

        CompletableFuture<String> combined = CompletableFuture.supplyAsync(() -> "材料3份", pool)
                .thenCombine(CompletableFuture.supplyAsync(() -> "审批2条", pool), (m, r) -> m + "，" + r);
        check(combined.join().equals("材料3份，审批2条"), "thenCombine 应合并两个结果");
        System.out.println("C 组合：thenApply／thenCompose／thenCombine 通过");
    }

    static CompletableFuture<String> queryOwnerAsync(String id, ThreadPoolExecutor pool) {
        return CompletableFuture.supplyAsync(() -> id + "的申请人", pool);
    }

    /** 第四节：异常沿链往后传，后面的 thenApply 不执行；join 抛 CompletionException、get 抛 ExecutionException；exceptionally／handle 可以兜底。 */
    static void exceptions(ThreadPoolExecutor pool) throws InterruptedException {
        AtomicBoolean laterStepRan = new AtomicBoolean(false);
        CompletableFuture<String> failed = CompletableFuture.<String>supplyAsync(() -> {
            throw new IllegalStateException("材料服务不可用");
        }, pool).thenApply(v -> {
            laterStepRan.set(true);
            return v + "!";
        });

        try {
            failed.join();
            check(false, "join 应抛异常");
        } catch (CompletionException e) {
            check(e.getCause() instanceof IllegalStateException, "join 的 CompletionException 里包着原始异常");
        }
        try {
            failed.get();
            check(false, "get 应抛异常");
        } catch (ExecutionException e) {
            check(e.getCause() instanceof IllegalStateException, "get 的 ExecutionException 里包着原始异常");
        }
        check(!laterStepRan.get(), "前一步失败时，后面的 thenApply 不应执行");

        AtomicReference<Throwable> seen = new AtomicReference<>();
        String recovered = failed.exceptionally(ex -> {
            seen.set(ex);
            return "材料暂不可用";
        }).join();
        check(recovered.equals("材料暂不可用"), "exceptionally 应给出替代结果");
        check(seen.get() instanceof CompletionException && seen.get().getCause() instanceof IllegalStateException,
                "回调里拿到的异常被包了一层 CompletionException，实际 " + seen.get());

        String handled = failed.handle((v, ex) -> ex == null ? v : "降级:" + ex.getCause().getMessage()).join();
        check(handled.equals("降级:材料服务不可用"), "handle 同时拿到结果和异常");

        String ok = CompletableFuture.supplyAsync(() -> "正常", pool)
                .handle((v, ex) -> ex == null ? v : "降级").join();
        check(ok.equals("正常"), "成功时 handle 拿到的 ex 是 null");

        AtomicBoolean logged = new AtomicBoolean(false);
        try {
            failed.whenComplete((v, ex) -> logged.set(ex != null)).join();
            check(false, "whenComplete 不改变异常结果");
        } catch (CompletionException e) {
            check(logged.get(), "whenComplete 能看到异常，但结果仍是异常");
        }
        System.out.println("D 异常：传播、join/get 异常类型、exceptionally/handle/whenComplete 通过");
    }

    /** 第四节：allOf 中有一个失败，allOf 也失败；其他任务的结果仍可单独取。 */
    static void allOfWithFailure(ThreadPoolExecutor pool) {
        CompletableFuture<String> good = CompletableFuture.supplyAsync(() -> "申请人", pool);
        CompletableFuture<String> bad = CompletableFuture.supplyAsync(() -> {
            throw new IllegalStateException("审批服务超时");
        }, pool);
        try {
            CompletableFuture.allOf(good, bad).join();
            check(false, "有一个失败时 allOf().join() 应抛异常");
        } catch (CompletionException e) {
            check(e.getCause() instanceof IllegalStateException, "allOf 的异常原因是失败任务的异常");
        }
        check(good.join().equals("申请人"), "成功的任务结果仍可单独取");
        System.out.println("E allOf 部分失败：通过");
    }

    /** 第五节：cancel(true) 不会中断正在执行的任务。 */
    static void cancelDoesNotInterrupt(ThreadPoolExecutor pool) {
        AtomicBoolean finished = new AtomicBoolean(false);
        AtomicBoolean wasInterrupted = new AtomicBoolean(false);
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                wasInterrupted.set(true);
                Thread.currentThread().interrupt();
                return "被中断";
            }
            finished.set(true);
            return "执行完了";
        }, pool);
        sleep(50);
        check(f.cancel(true), "cancel 应返回 true");
        check(f.isCancelled() && f.isCompletedExceptionally(), "取消后应处于已取消、异常完成状态");
        try {
            f.join();
            check(false, "取消后 join 应抛 CancellationException");
        } catch (CancellationException expected) {
            // ok
        }
        sleep(400);
        check(finished.get() && !wasInterrupted.get(), "任务应照常执行完，没有被中断");
        System.out.println("F cancel(true)：结果已取消，但任务照常执行完、未被中断：通过");
    }

    /** 第五节：orTimeout 到时异常完成；completeOnTimeout 到时给默认值；两者都不停止原任务。 */
    static void timeouts(ThreadPoolExecutor pool) {
        AtomicBoolean taskFinished = new AtomicBoolean(false);
        CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
            sleep(400);
            taskFinished.set(true);
            return "慢结果";
        }, pool);
        long start = System.nanoTime();
        try {
            slow.orTimeout(100, TimeUnit.MILLISECONDS).join();
            check(false, "orTimeout 到时应异常完成");
        } catch (CompletionException e) {
            check(e.getCause() instanceof TimeoutException, "原因应是 TimeoutException，实际 " + e.getCause());
        }
        long waited = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
        check(waited >= 90 && waited < 350, "应在约 100ms 时结束等待，实际 " + waited);

        String fallback = CompletableFuture.supplyAsync(() -> { sleep(400); return "慢结果"; }, pool)
                .completeOnTimeout("默认值", 100, TimeUnit.MILLISECONDS).join();
        check(fallback.equals("默认值"), "completeOnTimeout 到时应给默认值");

        check(!taskFinished.get(), "超时那一刻，原任务还没执行完");
        sleep(450);
        check(taskFinished.get(), "原任务没有被停止，之后照常执行完");
        System.out.println("G 超时：orTimeout／completeOnTimeout 通过，原任务未被停止");
    }
}