@Test public void testFlushViaThresholdOutsideOfReadLoop() { final AtomicInteger flushCount = new AtomicInteger(); EmbeddedChannel channel = newChannel(flushCount, true); for (int i = 0; i < EXPLICIT_FLUSH_AFTER_FLUSHES; i++) { channel.flush(); } assertEquals(1, flushCount.get()); assertFalse(channel.finish()); }
@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 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 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); }
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); }
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);  }