@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(); } }
boolean isQueueEmpty() { return queue.isEmpty(); }
FlowControlHandler extends ChannelDuplexHandler { boolean isQueueEmpty() { return queue.isEmpty(); } }
FlowControlHandler extends ChannelDuplexHandler { boolean isQueueEmpty() { return queue.isEmpty(); } FlowControlHandler();  FlowControlHandler(boolean releaseMessages); }
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); }
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);  }