testWriteInWritePromiseCompletePreservesOrder() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/local/LocalChannelTest.java lines 461–531

    @Test
    public void testWriteInWritePromiseCompletePreservesOrder() throws InterruptedException {
        Bootstrap cb = new Bootstrap();
        ServerBootstrap sb = new ServerBootstrap();
        final CountDownLatch messageLatch = new CountDownLatch(2);
        final ByteBuf data = Unpooled.buffer();
        final ByteBuf data2 = Unpooled.buffer();
        data.writeInt(Integer.BYTES).writeInt(2);
        data2.writeInt(Integer.BYTES).writeInt(1);

        try {
            cb.group(group1)
            .channel(LocalChannel.class)
            .handler(new TestHandler());

            sb.group(group2)
            .channel(LocalServerChannel.class)
            .childHandler(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                    if (msg instanceof ByteBuf) {
                        ByteBuf buf = (ByteBuf) msg;
                        while (buf.isReadable()) {
                            int size = buf.readInt();
                            ByteBuf slice = buf.readRetainedSlice(size);
                            try {
                                if (slice.readInt() == messageLatch.getCount()) {
                                    messageLatch.countDown();
                                }
                            } finally {
                                slice.release();
                            }
                        }
                        buf.release();
                    } else {
                        super.channelRead(ctx, msg);
                    }
                }
            });

            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();

                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 ->
                                ccCpy.writeAndFlush(data2.retainedDuplicate(), ccCpy.newPromise()));
                        ccCpy.writeAndFlush(data.retainedDuplicate(), promise);
                    }
                });

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