testChannelReRegisterRead() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/socket/nio/NioSocketChannelTest.java lines 177–249

    private static void testChannelReRegisterRead(final boolean sameEventLoop) throws Exception {
        EventLoopGroup group = new MultiThreadIoEventLoopGroup(2, NioIoHandler.newFactory());
        final CountDownLatch latch = new CountDownLatch(1);

        // Just some random bytes
        byte[] bytes = new byte[1024];
        ThreadLocalRandom.current().nextBytes(bytes);

        Channel sc = null;
        Channel cc = null;
        ServerBootstrap b = new ServerBootstrap();
        try {
            b.group(group)
             .channel(NioServerSocketChannel.class)
             .childOption(ChannelOption.SO_KEEPALIVE, true)
             .childHandler(new ChannelInitializer<Channel>() {
                 @Override
                 protected void initChannel(Channel ch) throws Exception {
                     ChannelPipeline pipeline = ch.pipeline();
                     pipeline.addLast(new SimpleChannelInboundHandler<ByteBuf>() {
                         @Override
                         protected void channelRead0(ChannelHandlerContext ctx, ByteBuf byteBuf) {
                             // We was able to read something from the Channel after reregister.
                             latch.countDown();
                         }

                         @Override
                         public void channelActive(final ChannelHandlerContext ctx) throws Exception {
                             final EventLoop loop = group.next();
                             if (sameEventLoop) {
                                 deregister(ctx, loop);
                             } else {
                                 loop.execute(new Runnable() {
                                     @Override
                                     public void run() {
                                         deregister(ctx, loop);
                                     }
                                 });
                             }
                         }

                         private void deregister(ChannelHandlerContext ctx, final EventLoop loop) {
                             // As soon as the channel becomes active re-register it to another
                             // EventLoop. After this is done we should still receive the data that
                             // was written to the channel.
                             ctx.deregister().addListener((ChannelFutureListener) cf -> {
                                 Channel channel = cf.channel();
                                 assertNotSame(loop, channel.eventLoop());
                                 group.next().register(channel);
                             });
                         }
                     });
                 }
             });

            sc = b.bind(0).syncUninterruptibly().channel();

            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInboundHandlerAdapter());
            cc = bootstrap.connect(sc.localAddress()).syncUninterruptibly().channel();
            cc.writeAndFlush(Unpooled.wrappedBuffer(bytes)).syncUninterruptibly();
            latch.await();
        } finally {
            if (cc != null) {
                cc.close();
            }
            if (sc != null) {
                sc.close();
            }
            group.shutdownGracefully();
        }
    }