FlowControlHandler extends ChannelDuplexHandler { boolean isQueueEmpty() { return queue.isEmpty(); } 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 testFlowAutoReadOn() throws Exception { final CountDownLatch latch = new CountDownLatch(3); ChannelInboundHandlerAdapter handler = new ChannelDuplexHandler() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { latch.countDown(); } }; FlowControlHandler flow = new FlowControlHandler(); Channel server = newServer(true, flow, handler); Channel client = newClient(server.localAddress()); try { client.writeAndFlush(newOneMessage()) .syncUninterruptibly(); assertTrue(latch.await(1L, SECONDS)); assertTrue(flow.isQueueEmpty()); } finally { client.close(); server.close(); } }
@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); 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 { Channel peer = peerRef.exchange(null, 1L, SECONDS); client.writeAndFlush(newOneMessage()) .syncUninterruptibly(); assertTrue(msgRcvLatch1.await(1L, SECONDS)); peer.config().setAutoRead(true); setAutoReadLatch1.countDown(); assertTrue(msgRcvLatch1.await(1L, SECONDS)); peer.config().setAutoRead(true); setAutoReadLatch2.countDown(); assertTrue(msgRcvLatch3.await(1L, SECONDS)); assertTrue(flow.isQueueEmpty()); } finally { client.close(); server.close(); } }