@Test public void basicsTest() throws Exception { List<byte[]> inputBuffers = getBuffers(1000, 20, 201, 0, 79, 1005); RetainingAsyncWritableChannel channel = new RetainingAsyncWritableChannel(); for (int i = 0; i < inputBuffers.size(); i++) { ByteBuffer buf = ByteBuffer.wrap(inputBuffers.get(i)); writeAndCheckCallback(buf, channel, buf.remaining(), 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();  }