@Override
    public void flush(ChannelHandlerContext ctx) throws Exception {
        if (readInProgress) {
            // If there is still a read in progress we are sure we will see a channelReadComplete(...) call. Thus
            // we only need to flush if we reach the explicitFlushAfterFlushes limit.
            if (++flushPendingCount == explicitFlushAfterFlushes) {
                flushNow(ctx);
            }
        } else if (consolidateWhenNoReadInProgress) {
            // Flush immediately if we reach the threshold, otherwise schedule
            if (++flushPendingCount == explicitFlushAfterFlushes) {
                flushNow(ctx);
            } else {
                scheduleFlush(ctx);
            }
        } else {
            // Always flush directly
            flushNow(ctx);
        }
    }
@Test
    public void testFlushViaReadComplete() {
        final AtomicInteger flushCount = new AtomicInteger();
        EmbeddedChannel channel = newChannel(flushCount, false);
        // Flush should go through as there is no read loop in progress.
        channel.flush();
        channel.runPendingTasks();
        assertEquals(1, flushCount.get());

        // Simulate read loop;
        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());
        // Now flush again as the read loop is complete.
        channel.flush();
        channel.runPendingTasks();
        assertEquals(3, flushCount.get());
        assertEquals(1L, (Long) channel.readOutbound());
        assertEquals(2L, (Long) channel.readOutbound());
        assertNull(channel.readOutbound());
        assertFalse(channel.finish());
    }