@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); }
public ChannelFuture removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (isEmpty()) { return null; } ChannelPromise p = ctx.newPromise(); PromiseCombiner combiner = new PromiseCombiner(); try { for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); if (!(promise instanceof VoidChannelPromise)) { combiner.add(promise); } ctx.write(msg, promise); write = next; } } combiner.finish(p); } catch (Throwable cause) { p.setFailure(cause); } assertEmpty(); return p; }
PendingWriteQueue { public ChannelFuture removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (isEmpty()) { return null; } ChannelPromise p = ctx.newPromise(); PromiseCombiner combiner = new PromiseCombiner(); try { for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); if (!(promise instanceof VoidChannelPromise)) { combiner.add(promise); } ctx.write(msg, promise); write = next; } } combiner.finish(p); } catch (Throwable cause) { p.setFailure(cause); } assertEmpty(); return p; } }
PendingWriteQueue { public ChannelFuture removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (isEmpty()) { return null; } ChannelPromise p = ctx.newPromise(); PromiseCombiner combiner = new PromiseCombiner(); try { for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); if (!(promise instanceof VoidChannelPromise)) { combiner.add(promise); } ctx.write(msg, promise); write = next; } } combiner.finish(p); } catch (Throwable cause) { p.setFailure(cause); } assertEmpty(); return p; } PendingWriteQueue(ChannelHandlerContext ctx); }
PendingWriteQueue { public ChannelFuture removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (isEmpty()) { return null; } ChannelPromise p = ctx.newPromise(); PromiseCombiner combiner = new PromiseCombiner(); try { for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); if (!(promise instanceof VoidChannelPromise)) { combiner.add(promise); } ctx.write(msg, promise); write = next; } } combiner.finish(p); } catch (Throwable cause) { p.setFailure(cause); } assertEmpty(); return p; } 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 removeAndWriteAll() { assert ctx.executor().inEventLoop(); if (isEmpty()) { return null; } ChannelPromise p = ctx.newPromise(); PromiseCombiner combiner = new PromiseCombiner(); try { for (PendingWrite write = head; write != null; write = head) { head = tail = null; size = 0; bytes = 0; while (write != null) { PendingWrite next = write.next; Object msg = write.msg; ChannelPromise promise = write.promise; recycle(write, false); if (!(promise instanceof VoidChannelPromise)) { combiner.add(promise); } ctx.write(msg, promise); write = next; } } combiner.finish(p); } catch (Throwable cause) { p.setFailure(cause); } assertEmpty(); return p; } 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();  }