testChannelInitializerEventExecutor() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/ChannelInitializerTest.java lines 263–398

    @SuppressWarnings("deprecation")
    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void testChannelInitializerEventExecutor() throws Throwable {
        final AtomicInteger invokeCount = new AtomicInteger();
        final AtomicInteger completeCount = new AtomicInteger();
        final AtomicReference<Throwable> errorRef = new AtomicReference<Throwable>();
        LocalAddress addr = new LocalAddress("test");

        final EventExecutor executor = new DefaultEventLoop() {
            private final ScheduledExecutorService execService = Executors.newSingleThreadScheduledExecutor();
            private Thread thread;

            @Override
            public void shutdown() {
                execService.shutdown();
            }

            @Override
            public synchronized boolean inEventLoop(Thread thread) {
                // Return false every other time which will ensure we hit the execute(...) path
                if (thread == null) {
                    thread = Thread.currentThread();
                    return false;
                }
                boolean result = thread == Thread.currentThread();
                thread = null;
                return result;
            }

            @Override
            public boolean isShuttingDown() {
                return false;
            }

            @Override
            public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
                throw new IllegalStateException();
            }

            @Override
            public Future<?> terminationFuture() {
                throw new IllegalStateException();
            }

            @Override
            public boolean isShutdown() {
                return execService.isShutdown();
            }

            @Override
            public boolean isTerminated() {
                return execService.isTerminated();
            }

            @Override
            public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
                return execService.awaitTermination(timeout, unit);
            }

            @Override
            public void execute(Runnable command) {
                execService.execute(command);
            }
        };

        final CountDownLatch latch = new CountDownLatch(1);
        ServerBootstrap serverBootstrap = new ServerBootstrap()
                .channel(LocalServerChannel.class)
                .group(group)
                .localAddress(addr)
                .childHandler(new ChannelInitializer<LocalChannel>() {
                    @Override
                    protected void initChannel(LocalChannel ch) {
                        ch.pipeline().addLast(executor, new ChannelInitializer<Channel>() {
                            @Override
                            protected void initChannel(Channel ch) {
                                invokeCount.incrementAndGet();
                                ChannelHandlerContext ctx = ch.pipeline().context(this);
                                assertNotNull(ctx);
                                ch.pipeline().addAfter(ctx.executor(),