NioIoHandlerTest Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/nio/NioIoHandlerTest.java lines 37–135

public class NioIoHandlerTest {

    @Test
    void testNioHandlerCanBeDrivenByMainThread() throws Exception {
        IoHandlerFactory factory = NioIoHandler.newFactory();
        final Thread current = Thread.currentThread();
        final Queue<Runnable> tasks = new ConcurrentLinkedQueue<>();
        // Create our own IoExecutor that is driven by the main thread
        final ThreadAwareExecutor executor = new ThreadAwareExecutor() {
            @Override
            public boolean isExecutorThread(Thread thread) {
                return current == thread;
            }

            @Override
            public void execute(Runnable command) {
                tasks.add(command);
            }
        };
        IoHandlerContext context = new IoHandlerContext() {
            @Override
            public boolean canBlock() {
                // Just busy spin.
                return false;
            }

            @Override
            public long delayNanos(long currentTimeNanos) {
                return 0;
            }

            @Override
            public long deadlineNanos() {
                return -1;
            }
        };
        IoHandler handler = factory.newHandler(executor);

        // Open a ServerSocketChannel that we can connect to.
        ServerSocketChannel channel = SelectorProvider.provider().openServerSocketChannel();
        channel.configureBlocking(false);
        channel.bind(new InetSocketAddress(0));
        // Do the connect in another thread so we don't block our io execution loop.
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                Socket socket = new Socket();
                try {
                    socket.connect(channel.getLocalAddress());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        AtomicBoolean acceptedConnection = new AtomicBoolean();
        IoRegistration registration = handler.register(new NioSelectableChannelIoHandle<ServerSocketChannel>(channel) {
            @Override
            protected void handle(ServerSocketChannel channel, SelectionKey key) {
                if (key.isAcceptable()) {
                    try {
                        // Just accept the connection and close it.
                        java.nio.channels.SocketChannel accepted = channel.accept();
                        accepted.close();
                    } catch (Exception e) {
                        // ignore
                    } finally {
                        acceptedConnection.set(true);
                    }
                }
            }
        });
        registration.submit(NioIoOps.ACCEPT);
        t.start();

        // Let's loop until our registration was cancelled.
        while (registration.isValid()) {
            handler.run(context);
            for (;;) {
                // Execute all tasks that were dispatched.
                Runnable r = tasks.poll();
                if (r == null) {