@Test public void testFlushViaDisconnect() { final AtomicInteger flushCount = new AtomicInteger(); EmbeddedChannel channel = newChannel(flushCount, false); channel.pipeline().fireChannelRead(1L); assertEquals(0, flushCount.get()); assertNull(channel.readOutbound()); channel.disconnect(); assertEquals(1, flushCount.get()); assertEquals(1L, channel.readOutbound()); assertNull(channel.readOutbound()); assertFalse(channel.finish()); }
@Override public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { resetReadAndFlushIfNeeded(ctx); ctx.disconnect(promise); }
FlushConsolidationHandler extends ChannelDuplexHandler { @Override public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { resetReadAndFlushIfNeeded(ctx); ctx.disconnect(promise); } }
FlushConsolidationHandler extends ChannelDuplexHandler { @Override public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { resetReadAndFlushIfNeeded(ctx); ctx.disconnect(promise); } FlushConsolidationHandler();  FlushConsolidationHandler(int explicitFlushAfterFlushes);  FlushConsolidationHandler(int explicitFlushAfterFlushes, boolean consolidateWhenNoReadInProgress); }
FlushConsolidationHandler extends ChannelDuplexHandler { @Override public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { resetReadAndFlushIfNeeded(ctx); ctx.disconnect(promise); } 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 disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { resetReadAndFlushIfNeeded(ctx); ctx.disconnect(promise); } 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);  }