testFlowToggleAutoRead() — netty Function Reference

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

Function java Buffer Allocators

Entity Profile

Buffer→ Allocators→ testFlowToggleAutoRead() — netty Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java lines 267–348

    @Test
    public void testFlowToggleAutoRead() throws Exception {
        final Exchanger<Channel> peerRef = new Exchanger<Channel>();
        final CountDownLatch msgRcvLatch1 = new CountDownLatch(1);
        final CountDownLatch msgRcvLatch2 = new CountDownLatch(1);
        final CountDownLatch msgRcvLatch3 = new CountDownLatch(1);
        final CountDownLatch setAutoReadLatch1 = new CountDownLatch(1);
        final CountDownLatch setAutoReadLatch2 = new CountDownLatch(1);

        ChannelInboundHandlerAdapter handler = new ChannelInboundHandlerAdapter() {
            private int msgRcvCount;
            private int expectedMsgCount;
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                peerRef.exchange(ctx.channel(), 1L, SECONDS);
                ctx.fireChannelActive();
            }

            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws InterruptedException {
                ReferenceCountUtil.release(msg);

                // Disable auto reading after each message
                ctx.channel().config().setAutoRead(false);

                if (msgRcvCount++ != expectedMsgCount) {
                    return;
                }
                switch (msgRcvCount) {
                    case 1:
                        msgRcvLatch1.countDown();
                        if (setAutoReadLatch1.await(1L, SECONDS)) {
                            ++expectedMsgCount;
                        }
                        break;
                    case 2:
                        msgRcvLatch2.countDown();
                        if (setAutoReadLatch2.await(1L, SECONDS)) {
                            ++expectedMsgCount;
                        }
                        break;
                    default:
                        msgRcvLatch3.countDown();
                        break;
                }
            }
        };

        final FlowControlHandler flow = new FlowControlHandler();
        Channel server = newServer(true, flow, handler);
        Channel client = newClient(server.localAddress());
        try {
            // The client connection on the server side
            Channel peer = peerRef.exchange(null, 1L, SECONDS);

            client.writeAndFlush(newOneMessage())
                .syncUninterruptibly();

            // channelRead(1)
            assertTrue(msgRcvLatch1.await(1L, SECONDS));

            // channelRead(2)
            peer.config().setAutoRead(true);
            setAutoReadLatch1.countDown();
            assertTrue(msgRcvLatch1.await(1L, SECONDS));

            // channelRead(3)
            peer.config().setAutoRead(true);
            setAutoReadLatch2.countDown();
            assertTrue(msgRcvLatch3.await(1L, SECONDS));

            assertTrue(peer.eventLoop().submit(new Callable<Boolean>() {
                @Override
                public Boolean call() {
                    return flow.isQueueEmpty();
                }
            }).get());
        } finally {
            client.close();
            server.close();
        }