@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;
            }
        }
    };
    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(flow.isQueueEmpty());
    } finally {
        client.close();
        server.close();
    }
}