testPeerWriteInWritePromiseCompleteDifferentEventLoopPreservesOrder() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/local/LocalChannelTest.java lines 533–609

    @Test
    public void testPeerWriteInWritePromiseCompleteDifferentEventLoopPreservesOrder() throws InterruptedException {
        Bootstrap cb = new Bootstrap();
        ServerBootstrap sb = new ServerBootstrap();
        final CountDownLatch messageLatch = new CountDownLatch(2);
        final ByteBuf data = Unpooled.wrappedBuffer(new byte[1024]);
        final ByteBuf data2 = Unpooled.wrappedBuffer(new byte[512]);
        final CountDownLatch serverChannelLatch = new CountDownLatch(1);
        final AtomicReference<Channel> serverChannelRef = new AtomicReference<Channel>();

        cb.group(group1)
                .channel(LocalChannel.class)
                .handler(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        if (data2.equals(msg)) {
                            ReferenceCountUtil.safeRelease(msg);
                            messageLatch.countDown();
                        } else {
                            super.channelRead(ctx, msg);
                        }
                    }
                });

        sb.group(group2)
                .channel(LocalServerChannel.class)
                .childHandler(new ChannelInitializer<LocalChannel>() {
                    @Override
                    public void initChannel(LocalChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                if (data.equals(msg)) {
                                    ReferenceCountUtil.safeRelease(msg);
                                    messageLatch.countDown();
                                } else {
                                    super.channelRead(ctx, msg);
                                }
                            }
                        });
                        serverChannelRef.set(ch);
                        serverChannelLatch.countDown();
                    }
                });

        Channel sc = null;
        Channel cc = null;
        try {
            // Start server
            sc = sb.bind(TEST_ADDRESS).syncUninterruptibly().channel();

            // Connect to the server
            cc = cb.connect(sc.localAddress()).syncUninterruptibly().channel();
            assertTrue(serverChannelLatch.await(5, SECONDS));

            final Channel ccCpy = cc;
            // Make sure a write operation is executed in the eventloop
            cc.pipeline().lastContext().executor().execute(new Runnable() {
                @Override
                public void run() {
                    ChannelPromise promise = ccCpy.newPromise();
                    promise.addListener(future -> {
                        Channel serverChannelCpy = serverChannelRef.get();
                        serverChannelCpy.writeAndFlush(data2.retainedDuplicate(), serverChannelCpy.newPromise());
                    });
                    ccCpy.writeAndFlush(data.retainedDuplicate(), promise);
                }
            });

            assertTrue(messageLatch.await(5, SECONDS));
        } finally {
            closeChannel(cc);
            closeChannel(sc);
            data.release();
            data2.release();
        }
    }