FlowControlHandler extends ChannelDuplexHandler { @Override public void read(ChannelHandlerContext ctx) throws Exception { if (dequeue(ctx, 1) == 0) { shouldConsume = true; ctx.read(); } } FlowControlHandler();  FlowControlHandler(boolean releaseMessages); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void read(ChannelHandlerContext ctx); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void channelReadComplete(ChannelHandlerContext ctx);  }
@Test public void testAutoReadingOff() throws Exception { final Exchanger<Channel> peerRef = new Exchanger<Channel>(); final CountDownLatch latch = new CountDownLatch(3); ChannelInboundHandlerAdapter handler = new ChannelInboundHandlerAdapter() { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { peerRef.exchange(ctx.channel(), 1L, SECONDS); ctx.fireChannelActive(); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ReferenceCountUtil.release(msg); latch.countDown(); } }; Channel server = newServer(false, handler); Channel client = newClient(server.localAddress()); try { Channel peer = peerRef.exchange(null, 1L, SECONDS); client.writeAndFlush(newOneMessage()) .syncUninterruptibly(); peer.read(); assertTrue(latch.await(1L, SECONDS)); } finally { client.close(); server.close(); } }