import java.util.ArrayList;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 第7课教师示例：逐项验证课件里关于 BlockingQueue（阻塞队列）的行为说法。
 * 运行：java reference/BlockingQueueDemo.java（Java 17 或 21，不需要额外依赖）。
 * 全部通过时最后一行输出“全部通过”；任何一项不符会抛出 AssertionError。
 * 只使用内存队列，所有等待都有超时，线程都会结束。
 */
public class BlockingQueueDemo {

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

    static long millisSince(long startNanos) {
        return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
    }

    public static void main(String[] args) throws Exception {
        fourFormsOfMethods();
        nullIsRejected();
        capacityFacts();
        blockingInThreads();
        producerConsumerWithPoisonObject();
        System.out.println("全部通过");
    }

    /** 第二节：四组方法在“放不进去／取不出来”时各自怎么处理。 */
    static void fourFormsOfMethods() throws Exception {
        BlockingQueue<String> full = new ArrayBlockingQueue<>(2);
        check(full.offer("a") && full.offer("b"), "前两次 offer 应成功");
        check(!full.offer("c"), "队满时 offer 返回 false");
        try {
            full.add("c");
            check(false, "队满时 add 应抛异常");
        } catch (IllegalStateException expected) {
            System.out.println("队满 add -> IllegalStateException: " + expected.getMessage());
        }
        long start = System.nanoTime();
        check(!full.offer("c", 100, TimeUnit.MILLISECONDS), "带超时的 offer 到时返回 false");
        check(millisSince(start) >= 90, "带超时的 offer 应等够时间");

        BlockingQueue<String> empty = new ArrayBlockingQueue<>(2);
        check(empty.poll() == null, "队空时 poll 返回 null");
        check(empty.peek() == null, "队空时 peek 返回 null");
        try {
            empty.remove();
            check(false, "队空时 remove 应抛异常");
        } catch (NoSuchElementException expected) {
            System.out.println("队空 remove -> NoSuchElementException");
        }
        try {
            empty.element();
            check(false, "队空时 element 应抛异常");
        } catch (NoSuchElementException expected) {
            // 与 remove 同类
        }
        start = System.nanoTime();
        check(empty.poll(100, TimeUnit.MILLISECONDS) == null, "带超时的 poll 到时返回 null");
        check(millisSince(start) >= 90, "带超时的 poll 应等够时间");
        System.out.println("A 四组方法：通过");
    }

    /** 第二节：阻塞队列不接受 null。 */
    static void nullIsRejected() throws Exception {
        List<BlockingQueue<String>> queues = List.of(
                new ArrayBlockingQueue<>(1), new LinkedBlockingQueue<>(), new SynchronousQueue<>());
        for (BlockingQueue<String> q : queues) {
            try {
                q.offer(null);
                check(false, q.getClass().getSimpleName() + " 的 offer(null) 应抛 NPE");
            } catch (NullPointerException expected) {
                // ok
            }
            try {
                q.put(null);
                check(false, q.getClass().getSimpleName() + " 的 put(null) 应抛 NPE");
            } catch (NullPointerException expected) {
                // ok
            }
        }
        System.out.println("B null 被拒绝（三种实现）：通过");
    }

    /** 第三、五节：容量与实现差别。 */
    static void capacityFacts() throws Exception {
        check(new LinkedBlockingQueue<String>().remainingCapacity() == Integer.MAX_VALUE,
                "LinkedBlockingQueue 不指定容量时剩余容量为 Integer.MAX_VALUE");
        check(new LinkedBlockingQueue<String>(3).remainingCapacity() == 3, "指定容量 3");
        ArrayBlockingQueue<String> array = new ArrayBlockingQueue<>(2);
        array.offer("x");
        check(array.remainingCapacity() == 1, "放入 1 个后剩余 1");

        SynchronousQueue<String> handoff = new SynchronousQueue<>();
        check(handoff.remainingCapacity() == 0 && handoff.size() == 0, "SynchronousQueue 没有容量");
        check(!handoff.offer("x"), "没有接收者时 offer 立即失败");
        check(handoff.peek() == null && handoff.poll() == null, "没有发送者时 peek／poll 为空");
        System.out.println("C 容量：LinkedBlockingQueue 默认剩余=" + Integer.MAX_VALUE
                + "，SynchronousQueue 容量=0：通过");
    }

    /** 第一、二节：put／take 真的会等；等待可以被中断。 */
    static void blockingInThreads() throws Exception {
        // take 在队空时等待，直到有人 put
        BlockingQueue<String> q1 = new ArrayBlockingQueue<>(1);
        AtomicReference<String> got = new AtomicReference<>();
        long[] takeElapsed = new long[1];
        Thread taker = new Thread(() -> {
            long s = System.nanoTime();
            try {
                got.set(q1.take());
                takeElapsed[0] = millisSince(s);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        taker.start();
        Thread.sleep(200);
        check(taker.getState() == Thread.State.WAITING, "队空时 take 的线程应在等待，实际 " + taker.getState());
        q1.put("x");
        taker.join(2000);
        check("x".equals(got.get()) && takeElapsed[0] >= 150, "take 应等到 put 之后才返回");

        // put 在队满时等待，直到有人 take
        BlockingQueue<String> q2 = new ArrayBlockingQueue<>(1);
        q2.put("a");
        Thread putter = new Thread(() -> {
            try {
                q2.put("b");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        putter.start();
        Thread.sleep(200);
        check(putter.getState() == Thread.State.WAITING, "队满时 put 的线程应在等待，实际 " + putter.getState());
        check(q2.size() == 1 && "a".equals(q2.peek()), "put 等待期间队列内容不变");
        check("a".equals(q2.take()), "取走 a");
        putter.join(2000);
        check(!putter.isAlive() && "b".equals(q2.peek()), "空出位置后 put 完成，队列里是 b");

        // 阻塞中的 take 可以被中断
        BlockingQueue<String> q3 = new LinkedBlockingQueue<>();
        AtomicReference<Boolean> interrupted = new AtomicReference<>(false);
        AtomicReference<Boolean> flagAfterCatch = new AtomicReference<>(true);
        AtomicReference<Boolean> flagAfterRestore = new AtomicReference<>(false);
        Thread waiter = new Thread(() -> {
            try {
                q3.take();
            } catch (InterruptedException e) {
                interrupted.set(true);
                flagAfterCatch.set(Thread.currentThread().isInterrupted());   // 异常抛出时中断标记已被清除
                Thread.currentThread().interrupt();                            // 恢复中断标记
                flagAfterRestore.set(Thread.currentThread().isInterrupted());
            }
        });
        waiter.start();
        Thread.sleep(100);
        waiter.interrupt();
        waiter.join(2000);
        check(interrupted.get() && !waiter.isAlive(), "阻塞中的 take 应被中断并抛 InterruptedException");
        check(!flagAfterCatch.get(), "抛出 InterruptedException 后中断标记应已被清除");
        check(flagAfterRestore.get(), "调用 interrupt() 后中断标记应恢复为 true");
        System.out.println("D 阻塞与中断：通过");
    }

    /** 第六节：有界队列 + 结束标记（poison object）的生产者—消费者；多个消费者各需要一个结束标记。 */
    static void producerConsumerWithPoisonObject() throws Exception {
        final int END = -1;
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(2);
        List<Integer> consumed = new ArrayList<>();
        int[] maxSize = new int[1];

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 6; i++) {
                    queue.put(i);
                    maxSize[0] = Math.max(maxSize[0], queue.size());
                }
                queue.put(END);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    int x = queue.take();
                    if (x == END) break;
                    consumed.add(x);
                    Thread.sleep(10);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        producer.start();
        consumer.start();
        producer.join(5000);
        consumer.join(5000);
        check(!producer.isAlive() && !consumer.isAlive(), "生产者和消费者都应结束");
        check(consumed.equals(List.of(0, 1, 2, 3, 4, 5)), "按放入顺序取出，实际 " + consumed);
        check(maxSize[0] <= 2, "有界队列长度不会超过容量，实际 " + maxSize[0]);
        System.out.println("E1 单消费者：consumed=" + consumed + "，生产者看到的最大队长=" + maxSize[0]);

        // 两个消费者：只放一个结束标记，另一个消费者会一直等下去
        BlockingQueue<Integer> shared = new LinkedBlockingQueue<>();
        List<Thread> consumers = new ArrayList<>();
        for (int c = 0; c < 2; c++) {
            Thread t = new Thread(() -> {
                try {
                    while (shared.take() != END) { /* 处理 */ }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            consumers.add(t);
            t.start();
        }
        shared.put(END);
        Thread.sleep(200);
        long alive = consumers.stream().filter(Thread::isAlive).count();
        check(alive == 1, "只放 1 个结束标记时应还剩 1 个消费者在等，实际 " + alive);
        shared.put(END);
        for (Thread t : consumers) t.join(2000);
        check(consumers.stream().noneMatch(Thread::isAlive), "放够 2 个结束标记后两个消费者都结束");
        System.out.println("E2 两个消费者：1 个结束标记只停 1 个，放够 2 个才全停");
        System.out.println("E 生产者—消费者：通过");
    }
}