@Test public void testFlowToggleAutoRead() throws Exception { final Exchanger<Channel> peerRef = new Exchanger<Channel>(); final AtomicReference<CountDownLatch> latchRef = new AtomicReference<CountDownLatch>(new CountDownLatch(1)); final AtomicInteger counter = new AtomicInteger(); ChannelInboundHandlerAdapter handler = new ChannelInboundHandlerAdapter() { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { peerRef.exchange(ctx.channel(), 1L, TimeUnit.SECONDS); ctx.fireChannelActive(); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ReferenceCountUtil.release(msg); counter.incrementAndGet(); ctx.channel().config().setAutoRead(false); CountDownLatch latch = latchRef.get(); latch.countDown(); } }; FlowControlHandler flow = new FlowControlHandler(); Channel server = newServer(true, flow, handler); Channel client = newClient(server.localAddress()); try { Channel peer = peerRef.exchange(null, 1L, TimeUnit.SECONDS); client.writeAndFlush(newOneMessage()) .syncUninterruptibly(); assertTrue(latchRef.get().await(1L, TimeUnit.SECONDS)); assertFalse(peer.config().isAutoRead()); assertEquals(1, counter.get()); latchRef.set(new CountDownLatch(1)); peer.config().setAutoRead(true); assertTrue(latchRef.get().await(1L, TimeUnit.SECONDS)); assertFalse(peer.config().isAutoRead()); assertEquals(2, counter.get()); latchRef.set(new CountDownLatch(1)); peer.config().setAutoRead(true); assertTrue(latchRef.get().await(1L, TimeUnit.SECONDS)); assertFalse(peer.config().isAutoRead()); assertEquals(3, counter.get()); 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);  }