CoalescingBufferQueue { public void add(ByteBuf buf) { add(buf, (ChannelFutureListener) null); } CoalescingBufferQueue(Channel channel); void add(ByteBuf buf); void add(ByteBuf buf, ChannelPromise promise); void add(ByteBuf buf, ChannelFutureListener listener); ByteBuf remove(int bytes, ChannelPromise aggregatePromise); int readableBytes(); boolean isEmpty(); void releaseAndFailAll(Throwable cause); void copyTo(CoalescingBufferQueue dest);  }
@Test public void testAggregateWithFullRead() { writeQueue.add(cat, catPromise); assertQueueSize(3, false); writeQueue.add(mouse, mouseListener); assertQueueSize(8, false); DefaultChannelPromise aggregatePromise = newPromise(); assertEquals("catmouse", dequeue(8, aggregatePromise)); assertQueueSize(0, true); assertFalse(catPromise.isSuccess()); assertFalse(mouseDone); aggregatePromise.setSuccess(); assertTrue(catPromise.isSuccess()); assertTrue(mouseSuccess); assertEquals(0, cat.refCnt()); assertEquals(0, mouse.refCnt()); }
@Test public void testWithVoidPromise() { writeQueue.add(cat, voidPromise); writeQueue.add(mouse, voidPromise); assertQueueSize(8, false); assertEquals("catm", dequeue(4, newPromise())); assertQueueSize(4, false); assertEquals("ouse", dequeue(4, newPromise())); assertQueueSize(0, true); assertEquals(0, cat.refCnt()); assertEquals(0, mouse.refCnt()); }
@Test public void testAggregateWithPartialRead() { writeQueue.add(cat, catPromise); writeQueue.add(mouse, mouseListener); DefaultChannelPromise aggregatePromise = newPromise(); assertEquals("catm", dequeue(4, aggregatePromise)); assertQueueSize(4, false); assertFalse(catPromise.isSuccess()); assertFalse(mouseDone); aggregatePromise.setSuccess(); assertTrue(catPromise.isSuccess()); assertFalse(mouseDone); aggregatePromise = newPromise(); assertEquals("ouse", dequeue(Integer.MAX_VALUE, aggregatePromise)); assertQueueSize(0, true); assertFalse(mouseDone); aggregatePromise.setSuccess(); assertTrue(mouseSuccess); assertEquals(0, cat.refCnt()); assertEquals(0, mouse.refCnt()); }
@Test public void testEmptyBuffersAreCoalesced() { ByteBuf empty = Unpooled.buffer(0, 1); assertQueueSize(0, true); writeQueue.add(cat, catPromise); writeQueue.add(empty, emptyPromise); assertQueueSize(3, false); DefaultChannelPromise aggregatePromise = newPromise(); assertEquals("cat", dequeue(3, aggregatePromise)); assertQueueSize(0, true); assertFalse(catPromise.isSuccess()); assertFalse(emptyPromise.isSuccess()); aggregatePromise.setSuccess(); assertTrue(catPromise.isSuccess()); assertTrue(emptyPromise.isSuccess()); assertEquals(0, cat.refCnt()); assertEquals(0, empty.refCnt()); }