Java 类io.netty.buffer.CompositeByteBuf 实例源码

protected void writeInContext() {
  CompositeByteBuf cbb = ByteBufAllocator.DEFAULT.compositeBuffer();
  for (;;) {
    ByteBuf buf = writeQueue.poll();
    if (buf == null) {
      break;
    }

    writeQueueSize.decrementAndGet();
    cbb.addComponent(true, buf);

    if (cbb.numComponents() == cbb.maxNumComponents()) {
      netSocket.write(Buffer.buffer(cbb));
      cbb = ByteBufAllocator.DEFAULT.compositeBuffer();
    }
  }
  if (cbb.isReadable()) {
    netSocket.write(Buffer.buffer(cbb));
  }
}

@Override
public List<Object> encode(ByteBufAllocator alloc) {
    ByteBuf meta = metadata(alloc);

    ByteBuf head = alloc.buffer(FastdfsConstants.FDFS_HEAD_LEN);
    head.writeLong(meta.readableBytes() + size);
    head.writeByte(cmd());
    head.writeByte(FastdfsConstants.ERRNO_OK);

    CompositeByteBuf cbb = alloc.compositeBuffer();
    cbb.addComponents(head, meta);
    cbb.writerIndex(head.readableBytes() + meta.readableBytes());

    List<Object> requests = new LinkedList<>();
    requests.add(cbb);
    requests.add(content);
    return requests;
}

/**
 * Turns the given BytesReference into a ByteBuf. Note: the returned ByteBuf will reference the internal
 * pages of the BytesReference. Don't free the bytes of reference before the ByteBuf goes out of scope.
 */
public static ByteBuf toByteBuf(final BytesReference reference) {
    if (reference.length() == 0) {
        return Unpooled.EMPTY_BUFFER;
    }
    if (reference instanceof ByteBufBytesReference) {
        return ((ByteBufBytesReference) reference).toByteBuf();
    } else {
        final BytesRefIterator iterator = reference.iterator();
        // usually we have one, two, or three components from the header, the message, and a buffer
        final List<ByteBuf> buffers = new ArrayList<>(3);
        try {
            BytesRef slice;
            while ((slice = iterator.next()) != null) {
                buffers.add(Unpooled.wrappedBuffer(slice.bytes, slice.offset, slice.length));
            }
            final CompositeByteBuf composite = Unpooled.compositeBuffer(buffers.size());
            composite.addComponents(true, buffers);
            return composite;
        } catch (IOException ex) {
            throw new AssertionError("no IO happens here", ex);
        }
    }
}

@Override
public List<Object> encode(ByteBufAllocator alloc) {
    ByteBuf meta = metadata(alloc);

    ByteBuf head = alloc.buffer(FDFS_HEAD_LEN);
    head.writeLong(meta.readableBytes() + size);
    head.writeByte(cmd());
    head.writeByte(ERRNO_OK);

    CompositeByteBuf cbb = alloc.compositeBuffer();
    cbb.addComponents(head, meta);
    cbb.writerIndex(head.readableBytes() + meta.readableBytes());

    List<Object> requests = new LinkedList<>();
    requests.add(cbb);
    requests.add(content);
    return requests;
}

private String dotStuff(String testString, int chunkSize) throws Exception {
  ByteArrayInputStream stream = new ByteArrayInputStream(testString.getBytes(StandardCharsets.UTF_8));
  DotStuffingChunkedStream chunkedStream = new DotStuffingChunkedStream(stream, chunkSize);

  CompositeByteBuf destBuffer = ALLOCATOR.compositeBuffer();

  while (!chunkedStream.isEndOfInput()) {
    destBuffer.addComponent(true, chunkedStream.readChunk(ALLOCATOR));
  }

  byte[] bytes = new byte[destBuffer.readableBytes()];
  destBuffer.getBytes(0, bytes);

  ReferenceCountUtil.release(destBuffer);
  return new String(bytes, CharsetUtil.UTF_8);
}

@Override
public List<Object> encode(ByteBufAllocator alloc) {
    ByteBuf meta = metadata(alloc);

    ByteBuf head = alloc.buffer(FDFS_HEAD_LEN);
    head.writeLong(meta.readableBytes() + size);
    head.writeByte(cmd());
    head.writeByte(ERRNO_OK);

    CompositeByteBuf cbb = alloc.compositeBuffer();
    cbb.addComponents(head, meta);
    cbb.writerIndex(head.readableBytes() + meta.readableBytes());

    List<Object> requests = new LinkedList<>();
    requests.add(cbb);
    requests.add(content);
    return requests;
}

@Override
public List<Object> encode(ByteBufAllocator alloc) {
    ByteBuf meta = metadata(alloc);

    ByteBuf head = alloc.buffer(FastdfsConstants.FDFS_HEAD_LEN);
    head.writeLong(meta.readableBytes() + size);
    head.writeByte(cmd());
    head.writeByte(FastdfsConstants.ERRNO_OK);

    CompositeByteBuf cbb = alloc.compositeBuffer();
    cbb.addComponents(head, meta);
    cbb.writerIndex(head.readableBytes() + meta.readableBytes());

    List<Object> requests = new LinkedList<>();
    requests.add(cbb);
    requests.add(content);
    return requests;
}

@Override
public List<Object> encode(ByteBufAllocator alloc) {
    ByteBuf meta = metadata(alloc);

    ByteBuf head = alloc.buffer(FDFS_HEAD_LEN);
    head.writeLong(meta.readableBytes() + size);
    head.writeByte(cmd());
    head.writeByte(ERRNO_OK);

    CompositeByteBuf cbb = alloc.compositeBuffer();
    cbb.addComponents(head, meta);
    cbb.writerIndex(head.readableBytes() + meta.readableBytes());

    List<Object> requests = new LinkedList<>();
    requests.add(cbb);
    requests.add(content);
    return requests;
}

public static byte[] readToBytes(ByteBuf byteBuf){


        if(byteBuf instanceof CompositeByteBuf){
            CompositeByteBuf compositeByteBuf = (CompositeByteBuf) byteBuf;
            ByteArrayOutputStream baous = new ByteArrayOutputStream();
            for(ByteBuf single : compositeByteBuf){
                try {
                    baous.write(readToBytes(single));
                } catch (IOException e) {
                    throw new IllegalStateException("write to ByteArrayOutputStream error", e);
                }
            }
            return baous.toByteArray();
        }else{
            byte []result = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(result);
            return result;
        }
    }

@Test
public void testNetty(){

    CompositeByteBuf byteBuf = ByteBufAllocator.DEFAULT.compositeBuffer();
    byteBuf.addComponent(Unpooled.wrappedBuffer("12345".getBytes()));
    byteBuf.addComponent(Unpooled.wrappedBuffer("abcde".getBytes()));

    System.out.println(ByteBufUtils.readToString(byteBuf));

    ByteBuf buf = Unpooled.wrappedBuffer(Unpooled.wrappedBuffer("134".getBytes()), Unpooled.wrappedBuffer("abc".getBytes()));
    System.out.println(buf.readableBytes());
    byte []result = new byte[buf.readableBytes()];
    buf.readBytes(result);
    System.out.println(new String(result));

}

/**
 * Try to add the given {@link CompositeByteBuf}. Returns {@code true} on success,
 * {@code false} otherwise.
 */
boolean add(CompositeByteBuf buf) {
    ByteBuffer[] buffers = buf.nioBuffers();
    if (count + buffers.length >= Native.IOV_MAX) {
        // No more room!
        return false;
    }
    for (int i = 0; i < buffers.length; i++) {
        ByteBuffer nioBuffer = buffers[i];
        int offset = nioBuffer.position();
        int len = nioBuffer.limit() - nioBuffer.position();
        if (len == 0) {
            // No need to add an empty buffer so just continue
            continue;
        }
        long addr = PlatformDependent.directBufferAddress(nioBuffer);

        add(addr, offset, len);
    }
    return true;
}

private ByteBuf writeCompressed(ByteBuf message) throws IOException {
    CompositeByteBuf compressed = alloc.compositeBuffer();
    try (OutputStream compressingStream = compressor.compress(new ByteBufOutputStream(compressed))) {
        compressingStream.write(ByteBufUtil.getBytes(message));
    } finally {
        message.release();
    }

    int numCompressedBytes = compressed.readableBytes();
    if (maxOutboundMessageSize >= 0 && numCompressedBytes > maxOutboundMessageSize) {
        compressed.release();
        throw Status.RESOURCE_EXHAUSTED
                .withDescription(
                        String.format(
                                "message too large %d > %d", numCompressedBytes, maxOutboundMessageSize))
                .asRuntimeException();
    }

    ByteBuf header = alloc.buffer(HEADER_LENGTH);
    header.writeByte(COMPRESSED);
    header.writeInt(numCompressedBytes);
    compressed.addComponent(true, 0, header);

    return compressed;
}

private ByteBuf writeUncompressed(ByteBuf message) {
    int messageLength = message.readableBytes();
    if (maxOutboundMessageSize >= 0 && messageLength > maxOutboundMessageSize) {
        throw Status.RESOURCE_EXHAUSTED
                .withDescription(
                        String.format("message too large %d > %d", messageLength, maxOutboundMessageSize))
                .asRuntimeException();
    }
    CompositeByteBuf buf = alloc.compositeBuffer();
    ByteBuf header = alloc.buffer(HEADER_LENGTH);
    header.writeByte(UNCOMPRESSED);
    header.writeInt(messageLength);
    buf.addComponent(true, header);
    buf.addComponent(true, message);

    return buf;
}

private byte[] fetchDecoderOutput() {
    CompositeByteBuf decoded = Unpooled.compositeBuffer();
    for (;;) {
        ByteBuf buf = decoder.readInbound();
        if (buf == null) {
            break;
        }
        if (!buf.isReadable()) {
            buf.release();
            continue;
        }
        decoded.addComponent(true, buf);
    }
    byte[] ret = ByteBufUtil.getBytes(decoded);
    decoded.release();
    return ret;
}

protected ByteBuf createContent(ByteBuf pathByteBuf, ByteBuf... restOfContent) {
    if (restOfContent == null || restOfContent.length == 0) {
        return pathByteBuf;
    } else {
        CompositeByteBuf composite = Unpooled.compositeBuffer(1 + restOfContent.length);
        composite.addComponent(pathByteBuf);
        composite.writerIndex(composite.writerIndex() + pathByteBuf.readableBytes());

        for (ByteBuf component : restOfContent) {
            composite.addComponent(component);
            composite.writerIndex(composite.writerIndex() + component.readableBytes());
        }

        return composite;
    }
}

private static ByteBuf encode(List<LookupCommand> commands) {
    CompositeByteBuf compositeBuf = Unpooled.compositeBuffer(commands.size()); //FIXME pooled allocator?
    for (LookupCommand command : commands) {
        byte[] pathBytes = command.path().getBytes(CharsetUtil.UTF_8);
        short pathLength = (short) pathBytes.length;

        ByteBuf commandBuf = Unpooled.buffer(4 + pathLength); //FIXME a way of using the pooled allocator?
        commandBuf.writeByte(command.opCode());
        //flags
        if (command.xattr()) {
            commandBuf.writeByte(SUBDOC_FLAG_XATTR_PATH);
        } else {
            commandBuf.writeByte(0);
        }
        commandBuf.writeShort(pathLength);
        //no value length
        commandBuf.writeBytes(pathBytes);

        compositeBuf.addComponent(commandBuf);
        compositeBuf.writerIndex(compositeBuf.writerIndex() + commandBuf.readableBytes());
    }
    return compositeBuf;
}

private static ByteBuf encode(EmbeddedChannel ch, Envelope... envelopes) {
    for (Envelope env : envelopes) {
        ch.writeOutbound(env);

        if (env.getBuffer() != null) {
            verify(env.getBuffer(), times(1)).recycleBuffer();
        }
    }

    CompositeByteBuf encodedEnvelopes = new CompositeByteBuf(ByteBufAllocator.DEFAULT, false, envelopes.length);

    ByteBuf buf;
    while ((buf = (ByteBuf) ch.readOutbound()) != null) {
        encodedEnvelopes.addComponent(buf);
    }

    return encodedEnvelopes.writerIndex(encodedEnvelopes.capacity());
}

public static ByteBuf encodePacket(final Packet packet) throws IOException {
  ByteBuf dataBytes = packet.getData();
  boolean hasData = dataBytes != null;

  CompositeByteBuf compositeByteBuf = PooledByteBufAllocator.DEFAULT.compositeBuffer(hasData ? 1 : 2);

  byte[] typeBytes = packet.getType().getValueAsBytes();
  int headerCapacity = typeBytes.length + DELIMITER_LENGTH + DELIMITER_LENGTH + (hasData ? DELIMITER_LENGTH : 0);
  ByteBuf headerByteBuf = PooledByteBufAllocator.DEFAULT.buffer(headerCapacity, headerCapacity);
  headerByteBuf.writeBytes(typeBytes);
  headerByteBuf.writeBytes(DELIMITER_BYTES);
  headerByteBuf.writeBytes(DELIMITER_BYTES);
  if (hasData) {
    headerByteBuf.writeBytes(DELIMITER_BYTES);
  }
  compositeByteBuf.addComponent(headerByteBuf);
  int compositeReadableBytes = headerByteBuf.readableBytes();

  if (hasData) {
    compositeByteBuf.addComponent(dataBytes);
    compositeReadableBytes += dataBytes.readableBytes();
  }

  compositeByteBuf.writerIndex(compositeReadableBytes);
  return compositeByteBuf;
}

/** Trims memory overhead if the openWritesLock is immediately available. */
protected boolean maybeTrimOutputBuffers() {
    if (openWritesLock.tryLock()) {
        try {
            for (StreamOut out : openStreamWrites.values()) {
                synchronized (out) {
                    if (out.outputBuffer.readableBytes() == 0) {
                        out.outputBuffer.capacity(0);
                    } else {
                        out.outputBuffer.discardReadBytes();
                        if (out.outputBuffer instanceof CompositeByteBuf) {
                            ((CompositeByteBuf) out.outputBuffer).consolidate();
                        }
                    }
                }
            }
        } finally {
            openWritesLock.unlock();
        }
        return true;
    }
    return false;
}

public void testToChannelBuffer() throws IOException {
    BytesReference ref = getRandomizedBytesReference(randomIntBetween(1, 3 * PAGE_SIZE));
    ByteBuf buffer = Netty4Utils.toByteBuf(ref);
    BytesReference bytesReference = Netty4Utils.toBytesReference(buffer);
    if (ref instanceof ByteBufBytesReference) {
        assertEquals(buffer, ((ByteBufBytesReference) ref).toByteBuf());
    } else if (AbstractBytesReferenceTestCase.getNumPages(ref) > 1) { // we gather the buffers into a channel buffer
        assertTrue(buffer instanceof CompositeByteBuf);
    }
    assertArrayEquals(BytesReference.toBytes(ref), BytesReference.toBytes(bytesReference));
}

@Override
public ByteBuf encrypt(ByteBuf data) {
    if (encCipher == null) {
        byte[] iv = EncryptUtils.randomBytes(getIvLength());
        encCipher = createCipher(iv, true);
        CompositeByteBuf bufs = new CompositeByteBuf(data.alloc(), data.isDirect(), 2);
        bufs.addComponents(true, Unpooled.wrappedBuffer(iv), process(encCipher, data));
        return bufs;
    }

    return process(encCipher, data);
}

/**
 * Listing 5.4
 */
public static void byteBufComposite(ByteBuf headerBuf, ByteBuf bodyBuf) {
    CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
    messageBuf.addComponents(headerBuf, bodyBuf);
    // ....
    messageBuf.removeComponent(0); // remove the header             //2

    for (int i = 0; i < messageBuf.numComponents(); i++) {                      //3
        System.out.println(messageBuf.component(i).toString());
    }
}

/**
 * Listing 5.5
 */
public static void byteBufCompositeArray(CompositeByteBuf compBuf) {
    int length = compBuf.readableBytes();                       //1
    byte[] array = new byte[length];                        //2
    compBuf.getBytes(compBuf.readerIndex(), array);                         //3
    handleArray(array, 0, length);
}

/**
 * Returns a {@link CompositeByteBuf} that contains the same data as {@code sourceBuffer}, but with
 * SMTP dot-stuffing applied, and (if {@code} appendCRLF is true) a CRLF appended.
 *
 * <p>If dot-stuffing is not required, and {@code appendCRLF} is false, {@code sourceBuffer} is
 * returned. In all other cases, {@code allocator} will be used to create a new {@code ByteBuf}
 * with a {@code refCnt} of one.
 *
 * <p>The {@code previousBytes} parameter is used to maintain dot-stuffing across a series
 * of buffers. Pass the last two bytes of a previous buffer here to ensure an initial dot
 * will be escaped if necessary. Passing null indicates this is the first or only buffer
 * for this message.
 *
 * @param allocator the {@code ByteBufAllocator} to use for new {@code ByteBuf}s
 * @param sourceBuffer the source message data
 * @param previousBytes the previous two bytes of the message, or null
 * @param termination whether to append CRLF to the end of the returned buffer
 */
public static ByteBuf createDotStuffedBuffer(ByteBufAllocator allocator, ByteBuf sourceBuffer, byte[] previousBytes, MessageTermination termination) {
  int dotIndex = findDotAtBeginningOfLine(sourceBuffer, 0, normalisePreviousBytes(previousBytes));

  try {
    if (dotIndex == -1) {
      if (termination == MessageTermination.ADD_CRLF) {
        return allocator.compositeBuffer(2).addComponents(true, sourceBuffer.retainedSlice(), CR_LF_BUFFER.slice());
      } else {
        return sourceBuffer.retainedSlice();
      }
    }

    // Build a CompositeByteBuf to avoid copying
    CompositeByteBuf compositeByteBuf = allocator.compositeBuffer();
    compositeByteBuf.addComponents(true, sourceBuffer.retainedSlice(0, dotIndex), DOT_DOT_BUFFER.slice());

    int nextDotIndex;
    while ((nextDotIndex = findDotAtBeginningOfLine(sourceBuffer, dotIndex + 1, NOT_CR_LF)) != -1) {
      compositeByteBuf.addComponents(true, sourceBuffer.retainedSlice(dotIndex + 1, nextDotIndex - dotIndex - 1), DOT_DOT_BUFFER.slice());
      dotIndex = nextDotIndex;
    }

    compositeByteBuf.addComponent(true, sourceBuffer.retainedSlice(dotIndex + 1, sourceBuffer.readableBytes() - dotIndex - 1));

    if (termination == MessageTermination.ADD_CRLF) {
      compositeByteBuf.addComponent(true, CR_LF_BUFFER.slice());
    }

    return compositeByteBuf;
  } finally {
    sourceBuffer.release();
  }
}

private String terminate(String testString, int chunkSize) throws Exception {
  ByteArrayInputStream stream = new ByteArrayInputStream(testString.getBytes(StandardCharsets.UTF_8));
  CrlfTerminatingChunkedStream chunkedStream = new CrlfTerminatingChunkedStream(stream, chunkSize);

  CompositeByteBuf destBuffer = ALLOCATOR.compositeBuffer();
  while (!chunkedStream.isEndOfInput()) {
    destBuffer.addComponent(true, chunkedStream.readChunk(ALLOCATOR));
  }

  byte[] bytes = new byte[destBuffer.readableBytes()];
  destBuffer.getBytes(0, bytes);
  destBuffer.release();
  return new String(bytes, CharsetUtil.UTF_8);
}

private static void releaseBuffers(CompositeByteBuf composite, List<ByteBuf> chunkBuffers) {
    while (composite.numComponents() > 0) {
        composite.removeComponent(0);
    }
    ReferenceCountUtil.safeRelease(composite);
    chunkBuffers.forEach(ReferenceCountUtil::safeRelease);
}

CompletableFuture<ByteString> readDataTypeDictionaryBytes(NodeId nodeId, int fragmentSize) {
    CompositeByteBuf fragmentBuffer = Unpooled.compositeBuffer();

    CompletableFuture<ByteBuf> future = readFragments(
        nodeId,
        fragmentBuffer,
        fragmentSize,
        0
    );

    return future.thenApply(buffer -> {
        // trim any junk at the end. some servers have a bug
        // that cause a null byte to be appended to the end,
        // which makes it invalid XML.
        int length = buffer.readableBytes();

        for (int i = buffer.writerIndex() - 1; i >= 0; i--) {
            byte lastByte = buffer.getByte(i);

            boolean empty =
                (lastByte == 0 ||
                    Character.isWhitespace(lastByte) ||
                    Character.isSpaceChar(lastByte));

            if (!empty) break;
            else length -= 1;
        }

        byte[] bs = new byte[length];
        buffer.readBytes(bs, 0, length);

        if (logger.isDebugEnabled()) {
            String xmlString = new String(bs);
            logger.debug("Dictionary XML: {}", xmlString);
        }

        return ByteString.of(bs);
    });
}

@Override
protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {

  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug("ChunkCreationHandler called with msg {} of size {} with chunkSize {}",
        msg, msg.readableBytes(), chunkSize);
  }

  if (!ctx.channel().isOpen()) {
    logger.debug("Channel closed, skipping encode inside {}.", RpcConstants.CHUNK_CREATION_HANDLER);
    msg.release();
    return;
  }

  // Calculate the number of chunks based on configured chunk size and input msg size
  int numChunks = (int) Math.ceil((double) msg.readableBytes() / chunkSize);

  // Initialize a composite buffer to hold numChunks chunk.
  final CompositeByteBuf cbb = ctx.alloc().compositeBuffer(numChunks);

  int cbbWriteIndex = 0;
  int currentChunkLen = min(msg.readableBytes(), chunkSize);

  // Create slices of chunkSize from input msg and add it to the composite buffer.
  while (numChunks > 0) {
    final ByteBuf chunkBuf = msg.slice(msg.readerIndex(), currentChunkLen);
    chunkBuf.retain();
    cbb.addComponent(chunkBuf);
    cbbWriteIndex += currentChunkLen;
    msg.skipBytes(currentChunkLen);
    --numChunks;
    currentChunkLen = min(msg.readableBytes(), chunkSize);
  }

  // Update the writerIndex of composite byte buffer. Netty doesn't do it automatically.
  cbb.writerIndex(cbbWriteIndex);

  // Add the final composite bytebuf into output buffer.
  out.add(cbb);
}

@Override
protected ByteBuf getWriteByteBuf() {

    int length = payload.length;
    CompositeByteBuf result = new CompositeByteBuf(UnpooledByteBufAllocator.DEFAULT, false, payload.length + 1);
    String prefix = String.format("%c%d\r\n", ASTERISK_BYTE, length);
    result.addComponent(Unpooled.wrappedBuffer(prefix.getBytes()));
    for(Object o : payload){
        ByteBuf buff = ParserManager.parse(o);
        result.addComponent(buff);
    }
    result.setIndex(0, result.capacity());
    return result;
}

private static void checkContentBuffer(FullHttpRequest aggregatedMessage) {
    CompositeByteBuf buffer = (CompositeByteBuf) aggregatedMessage.content();
    assertEquals(2, buffer.numComponents());
    List<ByteBuf> buffers = buffer.decompose(0, buffer.capacity());
    assertEquals(2, buffers.size());
    for (ByteBuf buf: buffers) {
        // This should be false as we decompose the buffer before to not have deep hierarchy
        assertFalse(buf instanceof CompositeByteBuf);
    }
    aggregatedMessage.release();
}

public void testSimpleSendCompositeDirectByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
    CompositeByteBuf buf = Unpooled.compositeBuffer();
    buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
    buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
    buf.writerIndex(4);
    testSimpleSend0(sb, cb, buf, true, BYTES, 1);

    CompositeByteBuf buf2 = Unpooled.compositeBuffer();
    buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
    buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
    buf2.writerIndex(4);
    testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}

public void testSimpleSendCompositeHeapByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
    CompositeByteBuf buf = Unpooled.compositeBuffer();
    buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 0, 2));
    buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
    buf.writerIndex(4);
    testSimpleSend0(sb, cb, buf, true, BYTES, 1);

    CompositeByteBuf buf2 = Unpooled.compositeBuffer();
    buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 0, 2));
    buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
    buf2.writerIndex(4);
    testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}

public void testSimpleSendCompositeMixedByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
    CompositeByteBuf buf = Unpooled.compositeBuffer();
    buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
    buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
    buf.writerIndex(4);
    testSimpleSend0(sb, cb, buf, true, BYTES, 1);

    CompositeByteBuf buf2 = Unpooled.compositeBuffer();
    buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
    buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
    buf2.writerIndex(4);
    testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}

@Test
public void testNioBuffersExpand2() {
    TestChannel channel = new TestChannel();

    ChannelOutboundBuffer buffer = new ChannelOutboundBuffer(channel);

    CompositeByteBuf comp = compositeBuffer(256);
    ByteBuf buf = directBuffer().writeBytes("buf1".getBytes(CharsetUtil.US_ASCII));
    for (int i = 0; i < 65; i++) {
        comp.addComponent(buf.copy()).writerIndex(comp.writerIndex() + buf.readableBytes());
    }
    buffer.addMessage(comp, comp.readableBytes(), channel.voidPromise());

    assertEquals("Should still be 0 as not flushed yet", 0, buffer.nioBufferCount());
    buffer.addFlush();
    ByteBuffer[] buffers = buffer.nioBuffers();
    assertEquals(65, buffer.nioBufferCount());
    for (int i = 0;  i < buffer.nioBufferCount(); i++) {
        if (i < 65) {
            assertEquals(buffers[i], buf.internalNioBuffer(0, buf.readableBytes()));
        } else {
            assertNull(buffers[i]);
        }
    }
    release(buffer);
    buf.release();
}

@Test
public void testStreamStartIsOnlyWrittenOnce() throws Exception {
    ByteBuf in = Unpooled.wrappedBuffer(new byte[] {
        'n', 'e', 't', 't', 'y'
    });

    channel.writeOutbound(in.copy());
    in.readerIndex(0); // rewind the buffer to write the same data
    channel.writeOutbound(in.copy());
    assertTrue(channel.finish());

    ByteBuf expected = Unpooled.wrappedBuffer(new byte[] {
        (byte) 0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59,
         0x01, 0x09, 0x00, 0x00, 0x6f, -0x68, -0x7e, -0x5e, 'n', 'e', 't', 't', 'y',
         0x01, 0x09, 0x00, 0x00, 0x6f, -0x68, -0x7e, -0x5e, 'n', 'e', 't', 't', 'y',
    });

    CompositeByteBuf actual = Unpooled.compositeBuffer();
    for (;;) {
        ByteBuf m = (ByteBuf) channel.readOutbound();
        if (m == null) {
            break;
        }
        actual.addComponent(m);
        actual.writerIndex(actual.writerIndex() + m.readableBytes());
    }
    assertEquals(releaseLater(expected), releaseLater(actual));
    in.release();
}

@Override
public boolean processMessage(Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        if (msg instanceof CompositeByteBuf) {
            return add((CompositeByteBuf) msg);
        } else {
            return add((ByteBuf) msg);
        }
    }
    return false;
}

/**
 * Returns a {@link IovArray} which is filled with the {@link CompositeByteBuf}.
 */
static IovArray get(CompositeByteBuf buf) throws Exception {
    IovArray array = ARRAY.get();
    array.clear();
    array.add(buf);
    return array;
}

public ByteBuf serializeRequest(I message) throws IOException {
    switch (requestType) {
        case PROTOBUF:
            return serializeProto((Message) message);
        default:
            CompositeByteBuf out = alloc.compositeBuffer();
            try (ByteBufOutputStream os = new ByteBufOutputStream(out)) {
                ByteStreams.copy(method.streamRequest(message), os);
            }
            return out;
    }
}

public ByteBuf serializeResponse(O message) throws IOException {
    switch (responseType) {
        case PROTOBUF:
            return serializeProto((Message) message);
        default:
            CompositeByteBuf out = alloc.compositeBuffer();
            try (ByteBufOutputStream os = new ByteBufOutputStream(out)) {
                ByteStreams.copy(method.streamResponse(message), os);
            }
            return out;
    }
}

private void add(CompositeByteBuf buf, HttpObject httpObject) {
    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;
        ByteBuf content = httpContent.content();
        if (content.isReadable()) {
            buf.addComponent(content.retain());
            buf.writerIndex(buf.writerIndex() + content.readableBytes());
        }
    }
}