问回收累积的字节缓冲区Pubilsher

EN

public static Mono<CompositeByteBuf> collectToComposite(Publisher<DataBuffer> data) {
  return Flux.from(data).reduce(
      EMPTY,
      (CompositeByteBuf acc, DataBuffer buffer) -> {
        ByteBuf byteBuf = toByteBuf(buffer);
        CompositeByteBuf composite = (acc == EMPTY) ? byteBuf.alloc().compositeBuffer(256) : acc;
        composite.addComponent(true, byteBuf);
        return composite;
      }
  ).map(composite -> composite != EMPTY ? composite : createEmptyComposite());
}

public static Mono<CompositeByteBuf> collectToComposite(Publisher<DataBuffer> data) {
  // such code is not suitable for multiple subscribers
  class CompositeHolder {
    CompositeByteBuf composite;

    void addComponent(ByteBuf component) {
      if (composite == null) {
        composite = component.alloc().compositeBuffer(256);
      }
      composite.addComponent(true, component);
    }
  }
  CompositeHolder holder = new CompositeHolder();
  return Flux.from(data)
      .doOnNext(buffer -> holder.addComponent(toByteBuf(buffer)))
      .doOnError(e -> holder.composite.release())
      .then(Mono.fromSupplier(() -> holder.composite));
}