@Test public void basicsTestWithNettyByteBuf() throws Exception { for (boolean useCompositeByteBuf : Arrays.asList(false, true)) { List<byte[]> inputBuffers = getBuffers(1000, 20, 201, 0, 79, 1005); RetainingAsyncWritableChannel channel = new RetainingAsyncWritableChannel(); for (int i = 0; i < inputBuffers.size(); i++) { byte[] data = inputBuffers.get(i); ByteBuf chunk; if (data.length == 0) { chunk = Unpooled.wrappedBuffer(data); } else if (!useCompositeByteBuf) { chunk = ByteBufAllocator.DEFAULT.heapBuffer(data.length); chunk.writeBytes(data); } else { CompositeByteBuf composite = ByteBufAllocator.DEFAULT.compositeHeapBuffer(100); ByteBuf c = ByteBufAllocator.DEFAULT.heapBuffer(data.length / 2); c.writeBytes(data, 0, data.length / 2); composite.addComponent(true, c); c = ByteBufAllocator.DEFAULT.heapBuffer(data.length - data.length / 2); c.writeBytes(data, data.length / 2, data.length - data.length / 2); composite.addComponent(true, c); chunk = composite; } writeAndCheckCallback(chunk, channel, chunk.readableBytes(), null, null); } checkStream(inputBuffers, channel); channel.close(); writeAndCheckCallback(ByteBuffer.allocate(0), channel, 0, ClosedChannelException.class, null); } }
@Override public void close() { open = false; synchronized (bufferLock) { if (compositeBuffer != null) { compositeBuffer.release(); compositeBuffer = null; } } }
RetainingAsyncWritableChannel implements AsyncWritableChannel { @Override public void close() { open = false; synchronized (bufferLock) { if (compositeBuffer != null) { compositeBuffer.release(); compositeBuffer = null; } } } }
RetainingAsyncWritableChannel implements AsyncWritableChannel { @Override public void close() { open = false; synchronized (bufferLock) { if (compositeBuffer != null) { compositeBuffer.release(); compositeBuffer = null; } } } RetainingAsyncWritableChannel();  RetainingAsyncWritableChannel(int sizeLimitInBytes); }
RetainingAsyncWritableChannel implements AsyncWritableChannel { @Override public void close() { open = false; synchronized (bufferLock) { if (compositeBuffer != null) { compositeBuffer.release(); compositeBuffer = null; } } } RetainingAsyncWritableChannel();  RetainingAsyncWritableChannel(int sizeLimitInBytes); @Override Future<Long> write(ByteBuffer src, Callback<Long> callback); @Override Future<Long> write(ByteBuf src, Callback<Long> callback); @Override boolean isOpen(); @Override void close(); ByteBuf consumeContentAsByteBuf(); ByteBufInputStream consumeContentAsInputStream(); long getBytesWritten(); }
RetainingAsyncWritableChannel implements AsyncWritableChannel { @Override public void close() { open = false; synchronized (bufferLock) { if (compositeBuffer != null) { compositeBuffer.release(); compositeBuffer = null; } } } RetainingAsyncWritableChannel();  RetainingAsyncWritableChannel(int sizeLimitInBytes); @Override Future<Long> write(ByteBuffer src, Callback<Long> callback); @Override Future<Long> write(ByteBuf src, Callback<Long> callback); @Override boolean isOpen(); @Override void close(); ByteBuf consumeContentAsByteBuf(); ByteBufInputStream consumeContentAsInputStream(); long getBytesWritten();  }