code: public static Mono<byte[]> collectBytesInByteBufStream(Flux<ByteBuf> stream, boolean autoReleaseEnabled) {
          if (autoReleaseEnabled) {
              // A stream is auto-release enabled means - the ByteBuf chunks in the stream get
              // released as consumer consumes each chunk.
              return Mono.using(Unpooled::compositeBuffer,
                  cbb -> stream.collect(() -> cbb,
                      (cbb1, buffer) -> cbb1.addComponent(true, Unpooled.wrappedBuffer(buffer).retain())),
                      ReferenceCountUtil::release)
                      .filter((CompositeByteBuf cbb) -> cbb.isReadable())
                      .map(FluxUtil::byteBufToArray);
          } else {
              return stream.collect(Unpooled::compositeBuffer,
                  (cbb1, buffer) -> cbb1.addComponent(true, Unpooled.wrappedBuffer(buffer)))
                      .filter((CompositeByteBuf cbb) -> cbb.isReadable())
                      .map(FluxUtil::byteBufToArray);
          }
      }
docs: Collects ByteBuf emitted by a Flux into a byte array.
  @param stream A stream which emits ByteBuf instances.
  @param autoReleaseEnabled if ByteBuf instances in stream gets automatically released as they consumed
  @return A Mono which emits the concatenation of all the ByteBuf instances given by the source Flux.