io.netty.buffer.Unpooled#compositeBuffer ( )源码实例Demo

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;
    }
}
 

/**
 * 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);
        }
    }
}
 

public static void main(String[] args) {
  ByteBuf bb = null;
  try {
    bb = Unpooled.buffer();
    test(bb);
  } finally {
    ReferenceCountUtil.release(bb);
  }

  System.out.println("----");

  CompositeByteBuf composite = null;
  try {
    composite = Unpooled.compositeBuffer(2);
    testComposite(composite);
  } finally {
    ReferenceCountUtil.release(composite);
  }
}
 

protected final ByteBuf captureWrite(ChannelHandlerContext ctx) {
  ArgumentCaptor<ByteBuf> captor = ArgumentCaptor.forClass(ByteBuf.class);
  verify(ctx, atLeastOnce()).write(captor.capture(), any(ChannelPromise.class));
  CompositeByteBuf composite = Unpooled.compositeBuffer();
  for (ByteBuf buf : captor.getAllValues()) {
    composite.addComponent(buf);
    composite.writerIndex(composite.writerIndex() + buf.readableBytes());
  }
  return composite;
}
 

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

        ByteBuf commandBuf = Unpooled.buffer(4 + pathLength + command.fragment().readableBytes());
        commandBuf.writeByte(command.opCode());
        byte subdocFlags = 0;
        if (command.createIntermediaryPath()) {
            subdocFlags |= KeyValueHandler.SUBDOC_BITMASK_MKDIR_P;
        }
        if (command.xattr()) {
            subdocFlags |= KeyValueHandler.SUBDOC_FLAG_XATTR_PATH;
        }
        if (command.expandMacros()) {
            subdocFlags |= KeyValueHandler.SUBDOC_FLAG_EXPAND_MACROS;
        }
        commandBuf.writeByte(subdocFlags);
        commandBuf.writeShort(pathLength);
        commandBuf.writeInt(command.fragment().readableBytes());
        commandBuf.writeBytes(pathBytes);

        //copy the fragment but don't move indexes (in case it is retained and reused)
        commandBuf.writeBytes(command.fragment(), command.fragment().readerIndex(), command.fragment().readableBytes());
        //eagerly release the fragment once it's been copied
        command.fragment().release();

        //add the command to the composite buffer
        compositeBuf.addComponent(commandBuf);
        compositeBuf.writerIndex(compositeBuf.writerIndex() + commandBuf.readableBytes());
    }
    return compositeBuf;
}
 

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

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

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

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

protected ByteBuf readDecompressed(final int dataLength) throws Exception {
    CompositeByteBuf compressed = Unpooled.compositeBuffer();
    ByteBuf msg;
    while ((msg = channel.readOutbound()) != null) {
        compressed.addComponent(true, msg);
    }
    return decompress(compressed, dataLength);
}
 

protected static ByteBuf readDecompressed(final EmbeddedChannel channel) {
    CompositeByteBuf decompressed = Unpooled.compositeBuffer();
    ByteBuf msg;
    while ((msg = channel.readInbound()) != null) {
        decompressed.addComponent(true, msg);
    }
    return decompressed;
}
 

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

    channel.writeOutbound(in.retain());
    in.resetReaderIndex(); // rewind the buffer to write the same data
    channel.writeOutbound(in);
    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 = channel.readOutbound();
        if (m == null) {
            break;
        }
        actual.addComponent(true, m);
    }
    assertEquals(expected, actual);

    expected.release();
    actual.release();
}
 

@Override
public Optional<TMessage> decodeOneMessage(ByteBuf in) throws TException {
  if (!in.isReadable()) {
    return Optional.empty();
  }
  compositeByteBuf.addComponent(true, in.retainedSlice());
  try {
    Optional<TMessage> outBuf;
    while (true) {
      int readerIndex = compositeByteBuf.readerIndex();
      outBuf = decodeOneImpl(compositeByteBuf);
      if (outBuf.isPresent()
          || readerIndex == compositeByteBuf.readerIndex()
          || compositeByteBuf.readableBytes() == 0) {
        break;
      }
    }
    if (outBuf.isPresent()) {
      in.skipBytes(in.readableBytes() - compositeByteBuf.readableBytes());
      compositeByteBuf.release();
      compositeByteBuf = Unpooled.compositeBuffer();
    } else {
      in.skipBytes(in.readableBytes());
    }
    return outBuf;
  } catch (Throwable t) {
    compositeByteBuf.release();
    compositeByteBuf = Unpooled.compositeBuffer();
    throw t;
  }
}
 

protected final ByteBuf captureWrite(ChannelHandlerContext ctx) {
  ArgumentCaptor<ByteBuf> captor = ArgumentCaptor.forClass(ByteBuf.class);
  verify(ctx, atLeastOnce()).write(captor.capture(), any(ChannelPromise.class));
  CompositeByteBuf composite = Unpooled.compositeBuffer();
  for (ByteBuf buf : captor.getAllValues()) {
    composite.addComponent(buf);
    composite.writerIndex(composite.writerIndex() + buf.readableBytes());
  }
  return composite;
}
 

@Override
public CompletableFuture<Void> beforeSendResponseAsync(Invocation invocation, HttpServletResponseEx responseEx) {
  Response response = (Response) responseEx.getAttribute(RestConst.INVOCATION_HANDLER_RESPONSE);
  ProduceProcessor produceProcessor =
      (ProduceProcessor) responseEx.getAttribute(RestConst.INVOCATION_HANDLER_PROCESSOR);
  Object body = response.getResult();
  if (response.isFailed()) {
    body = ((InvocationException) body).getErrorData();
  }

  if (null != invocation && isDownloadFileResponseType(invocation, response)) {
    return responseEx.sendPart(PartUtils.getSinglePart(null, body));
  }

  responseEx.setContentType(produceProcessor.getName() + "; charset=utf-8");

  CompletableFuture<Void> future = new CompletableFuture<>();
  try (BufferOutputStream output = new BufferOutputStream(Unpooled.compositeBuffer())) {
    produceProcessor.encodeResponse(output, body);

    responseEx.setBodyBuffer(output.getBuffer());
    future.complete(null);
  } catch (Throwable e) {
    future.completeExceptionally(ExceptionFactory.convertProducerException(e));
  }
  return future;
}
 

/**
 * Concatenates result of outbound Netty pipeline into an ByteBuf object.
 * <p/>
 * The optional byte buffer is returned only if a pipeline processing
 * resulted in outgoing bytes. Otherwise optional return value is absent.
 */
private static Optional<ByteBuf> grabSentBytes(EmbeddedChannel channel) {
    CompositeByteBuf outboundBytes = Unpooled.compositeBuffer();

    Object result = channel.readOutbound();
    while (result != null) {
        outboundBytes.addComponent((ByteBuf) result);
        result = channel.readOutbound();
    }

    if (outboundBytes.numComponents() > 0) {
        return Optional.of((ByteBuf) outboundBytes);
    }
    return Optional.empty();
}
 

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 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
protected void doStartInput() {
    super.doStartInput();
    cumulation = Unpooled.compositeBuffer(INIT_SIZE);
}
 

/**
 * @param args
 */
public static void main(String[] args)
{
    byte[] bs1 = new byte[]
    { 1, 10, 11, 12 };
    byte[] bs2 = new byte[]
    { 2, 2, 2, 2 };
    byte[] bs3 = new byte[]
    { 3, 3, 3, 3 };
    byte[] bs4 = new byte[]
    { 4, 4, 4, 4 };
    byte[] bs5 = new byte[]
    { 5, 5, 5, 5 };
    byte[] bs6 = new byte[]
    { 6, 6, 6, 6 };

    ByteBuffer buffer1 = ByteBuffer.allocate(1024);
    buffer1.put(bs1);
    buffer1.flip();

    ByteBuf buf1 = Unpooled.copiedBuffer(buffer1);// .copiedBuffer(bs1);

    buffer1.put(bs3);

    ByteBuf buf2 = Unpooled.copiedBuffer(bs2);
    ByteBuf buf3 = Unpooled.copiedBuffer(bs3);
    ByteBuf buf4 = Unpooled.copiedBuffer(bs4);
    ByteBuf buf5 = Unpooled.copiedBuffer(bs5);
    ByteBuf buf6 = Unpooled.copiedBuffer(bs6);

    CompositeByteBuf cb = Unpooled.compositeBuffer();
    cb.addComponents(buf1, buf2, buf3);

    byte dd = cb.getByte(0);

    CompositeByteBuf cb2 = Unpooled.compositeBuffer();
    cb.addComponents(buf4, buf5, buf6);

    // cb.c
    // cb2.writerIndex(128 * 1024);

    cb.addComponent(cb2);

    Long number = cb2.readLong(); // causes IllegalBufferAccessException
                                  // here!

}