testRemoveFlowControl() — netty Function Reference

Architecture documentation for the testRemoveFlowControl() function in FlowControlHandlerTest.java from the netty codebase.

Function java Buffer Allocators

Entity Profile

Buffer→ Allocators→ testRemoveFlowControl() — netty Function Reference

Dependency Diagram

graph TD
  9b233acd_9531_bef5_56cd_53724366232a["testRemoveFlowControl()"]
  403e241f_d76e_484e_d952_7f7a46681916["FlowControlHandlerTest"]
  9b233acd_9531_bef5_56cd_53724366232a -->|defined in| 403e241f_d76e_484e_d952_7f7a46681916
  style 9b233acd_9531_bef5_56cd_53724366232a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java lines 596–664

    @Test
    public void testRemoveFlowControl() throws Exception {
        final Exchanger<Channel> peerRef = new Exchanger<Channel>();

        final CountDownLatch latch = new CountDownLatch(3);

        ChannelInboundHandlerAdapter handler = new ChannelDuplexHandler() {
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                peerRef.exchange(ctx.channel(), 1L, SECONDS);
                //do the first read
                ctx.read();
                super.channelActive(ctx);
            }
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                latch.countDown();
                super.channelRead(ctx, msg);
            }
        };

        final FlowControlHandler flow = new FlowControlHandler() {
            private int num;
            @Override
            public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
                super.channelRead(ctx, msg);
                ++num;
                if (num >= 3) {
                    //We have received 3 messages. Remove myself later
                    final ChannelHandler handler = this;
                    ctx.channel().eventLoop().execute(new Runnable() {
                        @Override
                        public void run() {
                            ctx.pipeline().remove(handler);
                        }
                    });
                }
            }
        };
        ChannelInboundHandlerAdapter tail = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                //consume this msg
                ReferenceCountUtil.release(msg);
            }
        };

        Channel server = newServer(false /* no auto read */, flow, handler, tail);
        Channel client = newClient(server.localAddress());
        try {
            // The client connection on the server side
            Channel peer = peerRef.exchange(null, 1L, SECONDS);

            // Write one message
            client.writeAndFlush(newOneMessage()).sync();

            // We should receive 3 messages
            assertTrue(latch.await(1L, SECONDS));
            assertTrue(peer.eventLoop().submit(new Callable<Boolean>() {
                @Override
                public Boolean call() {
                    return flow.isQueueEmpty();
                }
            }).get());
        } finally {
            client.close();
            server.close();
        }
    }