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 ) ; } }
Collects ByteBuf emitted by a Flux into a byte array .