PendingWriteQueue { public ChannelFuture removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (size == 1) { return removeAndWrite(); } PendingWrite write = head; if (write == null) { return null; } head = tail = null; size = 0; ChannelPromise p = ctx.newPromise(); ChannelPromiseAggregator aggregator = new ChannelPromiseAggregator(p); while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); ctx.write(msg, promise); aggregator.add(promise); write = next; } assertEmpty(); return p; } PendingWriteQueue(ChannelHandlerContext ctx); boolean isEmpty(); int size(); void add(Object msg, ChannelPromise promise); void removeAndFailAll(Throwable cause); void removeAndFail(Throwable cause); ChannelFuture removeAndWriteAll(); ChannelFuture removeAndWrite(); ChannelPromise remove(); Object current();  }
@Test public void testRemoveAndWriteAll() { assertWrite(new TestHandler() { @Override public void flush(ChannelHandlerContext ctx) throws Exception { assertFalse("Should not be writable anymore", ctx.channel().isWritable()); ChannelFuture future = queue.removeAndWriteAll(); future.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { assertQueueEmpty(queue); } }); super.flush(ctx); } }, 3); }