import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;

public class SelectorDemo {
    static void require(boolean ok, String message) {
        if (!ok) throw new AssertionError(message);
    }
    public static void main(String[] args) throws Exception {
        String mode = args.length == 0 ? "normal" : args[0];
        if (!mode.equals("normal") && !mode.equals("stale") && !mode.equals("busy"))
            throw new IllegalArgumentException("normal|stale|busy");
        // Blocking setup is intentional: only the accepted endpoint uses the selector.
        try (Selector selector = Selector.open();
             ServerSocketChannel listener = ServerSocketChannel.open()) {
            listener.bind(new InetSocketAddress("127.0.0.1", 0));
            try (SocketChannel client = SocketChannel.open(listener.getLocalAddress());
                 SocketChannel peer = listener.accept()) {
                peer.configureBlocking(false);
                SelectionKey key = peer.register(selector, SelectionKey.OP_READ);
                client.shutdownOutput(); // EOF at peer; client can still read the reply.
                ByteBuffer input = ByteBuffer.allocate(8);
                ByteBuffer pending = null;
                long deadline = System.nanoTime() + 5_000_000_000L;
                while (key.isValid() && pending == null) {
                    require(System.nanoTime() < deadline, "EOF timeout");
                    selector.select(100);
                    var it = selector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey selected = it.next();
                        if (!mode.equals("stale")) it.remove();
                        if (selected.isValid() && selected.isReadable()) {
                            int n = peer.read(input);
                            if (n == -1) {
                                key.interestOpsAnd(~SelectionKey.OP_READ);
                                pending = StandardCharsets.UTF_8.encode("bye\n");
                                key.interestOpsOr(SelectionKey.OP_WRITE);
                                System.out.println("EOF -> READ off, WRITE on");
                            }
                        }
                    }
                }
                require(selector.selectedKeys().isEmpty(), "selectedKeys retained: missing iterator.remove");
                while (pending.hasRemaining()) {
                    require(System.nanoTime() < deadline, "write timeout");
                    selector.select(100);
                    var it = selector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey selected = it.next();
                        it.remove();
                        if (selected.isValid() && selected.isWritable()) peer.write(pending);
                    }
                }
                if (mode.equals("busy")) {
                    int writableEvents = 0;
                    for (int i = 0; i < 20; i++) {
                        selector.select(20);
                        for (SelectionKey selected : selector.selectedKeys())
                            if (selected.isWritable()) writableEvents++;
                        selector.selectedKeys().clear();
                    }
                    System.out.println("empty queue writable events=" + writableEvents);
                    require(writableEvents == 0, "OP_WRITE retained with empty queue");
                }
                key.interestOpsAnd(~SelectionKey.OP_WRITE);
                require(key.interestOps() == 0, "interest set not empty");
                System.out.println("drained -> WRITE off");
                key.cancel();
                peer.close();
                selector.selectNow(); // processes cancelled keys
                require(!key.isValid() && selector.keys().isEmpty(), "registration leaked");
                client.configureBlocking(false);
                ByteBuffer reply = ByteBuffer.allocate(16);
                boolean eof = false;
                while (System.nanoTime() < deadline) {
                    if (client.read(reply) == -1) { eof = true; break; }
                    Thread.sleep(1);
                }
                reply.flip();
                String body = StandardCharsets.UTF_8.decode(reply).toString();
                require(eof && body.equals("bye\n"), "reply or EOF missing");
                System.out.println("client=bye; EOF=true; registered=0; PASS");
            }
        }
    }
}