@Test public void testRemoveAndWrite() { assertWrite(new TestHandler() { @Override public void flush(ChannelHandlerContext ctx) throws Exception { assertFalse("Should not be writable anymore", ctx.channel().isWritable()); ChannelFuture future = queue.removeAndWrite(); future.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { assertQueueEmpty(queue); } }); super.flush(ctx); } }, 1); }
public ChannelFuture removeAndWrite() { assert ctx.executor().inEventLoop(); PendingWrite write = head; if (write == null) { return null; } Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, true); return ctx.write(msg, promise); }
PendingWriteQueue { public ChannelFuture removeAndWrite() { assert ctx.executor().inEventLoop(); PendingWrite write = head; if (write == null) { return null; } Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, true); return ctx.write(msg, promise); } }
PendingWriteQueue { public ChannelFuture removeAndWrite() { assert ctx.executor().inEventLoop(); PendingWrite write = head; if (write == null) { return null; } Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, true); return ctx.write(msg, promise); } PendingWriteQueue(ChannelHandlerContext ctx); }
PendingWriteQueue { public ChannelFuture removeAndWrite() { assert ctx.executor().inEventLoop(); PendingWrite write = head; if (write == null) { return null; } Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, true); return ctx.write(msg, promise); } PendingWriteQueue(ChannelHandlerContext ctx); boolean isEmpty(); int size(); long bytes(); void add(Object msg, ChannelPromise promise); ChannelFuture removeAndWriteAll(); void removeAndFailAll(Throwable cause); void removeAndFail(Throwable cause); ChannelFuture removeAndWrite(); ChannelPromise remove(); Object current(); }
PendingWriteQueue { public ChannelFuture removeAndWrite() { assert ctx.executor().inEventLoop(); PendingWrite write = head; if (write == null) { return null; } Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, true); return ctx.write(msg, promise); } PendingWriteQueue(ChannelHandlerContext ctx); boolean isEmpty(); int size(); long bytes(); void add(Object msg, ChannelPromise promise); ChannelFuture removeAndWriteAll(); void removeAndFailAll(Throwable cause); void removeAndFail(Throwable cause); ChannelFuture removeAndWrite(); ChannelPromise remove(); Object current();  }