testExecutorPreserveOrdering() — netty Function Reference

Source Code

codec-http/src/test/java/io/netty/handler/codec/http/HttpContentCompressorTest.java lines 423–519

    @Test
    public void testExecutorPreserveOrdering() throws Exception {
        final EventLoopGroup compressorGroup = new MultiThreadIoEventLoopGroup(1, LocalIoHandler.newFactory());
        EventLoopGroup localGroup = new MultiThreadIoEventLoopGroup(1, LocalIoHandler.newFactory());
        Channel server = null;
        Channel client = null;
        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
                .channel(LocalServerChannel.class)
                .group(localGroup)
                .childHandler(new ChannelInitializer<LocalChannel>() {
                @Override
                protected void initChannel(LocalChannel ch) throws Exception {
                    ch.pipeline()
                        .addLast(new HttpServerCodec())
                        .addLast(new HttpObjectAggregator(1024))
                        .addLast(compressorGroup, new HttpContentCompressor())
                        .addLast(new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
                                throws Exception {
                                super.write(ctx, msg, promise);
                            }
                        })
                        .addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                if (msg instanceof FullHttpRequest) {
                                    FullHttpResponse res =
                                        new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                                            Unpooled.copiedBuffer("Hello, World", CharsetUtil.US_ASCII));
                                    ctx.writeAndFlush(res);
                                    ReferenceCountUtil.release(msg);
                                    return;
                                }
                                super.channelRead(ctx, msg);
                            }
                        });
                }
            });

            LocalAddress address = new LocalAddress(UUID.randomUUID().toString());
            server = bootstrap.bind(address).sync().channel();

            final BlockingQueue<HttpObject> responses = new LinkedBlockingQueue<HttpObject>();

            client = new Bootstrap()
                .channel(LocalChannel.class)
                .remoteAddress(address)
                .group(localGroup)
                .handler(new ChannelInitializer<LocalChannel>() {
                @Override
                protected void initChannel(LocalChannel ch) throws Exception {
                    ch.pipeline().addLast(new HttpClientCodec()).addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            if (msg instanceof HttpObject) {
                                responses.put((HttpObject) msg);
                                return;
                            }
                            super.channelRead(ctx, msg);
                        }
                    });
                }
            }).connect().sync().channel();

            client.writeAndFlush(newRequest()).sync();

            assertEncodedResponse((HttpResponse) responses.poll(1, TimeUnit.SECONDS));
            HttpContent c = (HttpContent) responses.poll(1, TimeUnit.SECONDS);
            assertNotNull(c);
            assertEquals("1f8b0800000000000000f248cdc9c9d75108cf2fca4901000000ffff",
                    ByteBufUtil.hexDump(c.content()));
            c.release();

            c = (HttpContent) responses.poll(1, TimeUnit.SECONDS);
            assertNotNull(c);
            assertEquals("0300c6865b260c000000", ByteBufUtil.hexDump(c.content()));
            c.release();

            LastHttpContent last = (LastHttpContent) responses.poll(1, TimeUnit.SECONDS);