@Test public void testRemoveAndFailAll() { assertWriteFails(new TestHandler() { @Override public void flush(ChannelHandlerContext ctx) throws Exception { queue.removeAndFailAll(new TestException()); super.flush(ctx); } }, 3); }
public void removeAndFailAll(Throwable cause) { assert ctx.executor().inEventLoop(); if (cause == null) { throw new NullPointerException("cause"); } for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; ReferenceCountUtil.safeRelease(write.msg); ChannelPromise promise = write.promise; recycle(write, false); safeFail(promise, cause); write = next; } } assertEmpty(); }
PendingWriteQueue { public void removeAndFailAll(Throwable cause) { assert ctx.executor().inEventLoop(); if (cause == null) { throw new NullPointerException("cause"); } for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; ReferenceCountUtil.safeRelease(write.msg); ChannelPromise promise = write.promise; recycle(write, false); safeFail(promise, cause); write = next; } } assertEmpty(); } }
PendingWriteQueue { public void removeAndFailAll(Throwable cause) { assert ctx.executor().inEventLoop(); if (cause == null) { throw new NullPointerException("cause"); } for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; ReferenceCountUtil.safeRelease(write.msg); ChannelPromise promise = write.promise; recycle(write, false); safeFail(promise, cause); write = next; } } assertEmpty(); } PendingWriteQueue(ChannelHandlerContext ctx); }
PendingWriteQueue { public void removeAndFailAll(Throwable cause) { assert ctx.executor().inEventLoop(); if (cause == null) { throw new NullPointerException("cause"); } for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; ReferenceCountUtil.safeRelease(write.msg); ChannelPromise promise = write.promise; recycle(write, false); safeFail(promise, cause); write = next; } } assertEmpty(); } 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 void removeAndFailAll(Throwable cause) { assert ctx.executor().inEventLoop(); if (cause == null) { throw new NullPointerException("cause"); } for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; ReferenceCountUtil.safeRelease(write.msg); ChannelPromise promise = write.promise; recycle(write, false); safeFail(promise, cause); write = next; } } assertEmpty(); } 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();  }