### Question:
FlushConsolidationHandler extends ChannelDuplexHandler { @Override public void flush(ChannelHandlerContext ctx) throws Exception { if (readInProgress) { if (++flushPendingCount == explicitFlushAfterFlushes) { flushNow(ctx); } } else if (consolidateWhenNoReadInProgress) { if (++flushPendingCount == explicitFlushAfterFlushes) { flushNow(ctx); } else { scheduleFlush(ctx); } } else { flushNow(ctx); } } FlushConsolidationHandler();  FlushConsolidationHandler(int explicitFlushAfterFlushes);  FlushConsolidationHandler(int explicitFlushAfterFlushes, boolean consolidateWhenNoReadInProgress); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void flush(ChannelHandlerContext ctx); @Override void channelReadComplete(ChannelHandlerContext ctx); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); @Override void disconnect(ChannelHandlerContext ctx, ChannelPromise promise); @Override void close(ChannelHandlerContext ctx, ChannelPromise promise); @Override void channelWritabilityChanged(ChannelHandlerContext ctx); @Override void handlerRemoved(ChannelHandlerContext ctx);  }### Answer:
@Test public void testFlushViaReadComplete() { final AtomicInteger flushCount = new AtomicInteger(); EmbeddedChannel channel = newChannel(flushCount, false); channel.flush(); channel.runPendingTasks(); assertEquals(1, flushCount.get()); channel.pipeline().fireChannelRead(1L); assertEquals(1, flushCount.get()); channel.pipeline().fireChannelRead(2L); assertEquals(1, flushCount.get()); assertNull(channel.readOutbound()); channel.pipeline().fireChannelReadComplete(); assertEquals(2, flushCount.get()); channel.flush(); channel.runPendingTasks(); assertEquals(3, flushCount.get()); assertNull(channel.readOutbound()); assertFalse(channel.finish()); }