Reactor - How to compress Flux<ByteBuffer> on the fly?

You want to avoid buffering of full data, but you can archive each ByteBuffer chunk or, if your chunks are sufficiently small, to consolidate chunks in groups and then to archive.

This will not require too long memory, but will compress your data. The actual level of compression depends on content of your source data and on number of chunks consolidated before archiving. I think, you cat adjust it manually to have the best ratio.

The example of probable code is below:

public class Test_GzipFlux {

/**
 * Returns Flux of gzip-ed buffers after (optional) buffer consolidation
 * @param inFlux input stream of buffers
 * @param consolidatedBufCount number of buffers to consolidate before gzip-ing
 */
public static Flux<ByteBuffer> gzipFlux(Flux<ByteBuffer> inFlux, 
                                        int consolidatedBufCount, int outChunkMaxLength) {
    return inFlux.buffer(consolidatedBufCount)
                 .map(inList->zipBuffers(inList, outChunkMaxLength));
}

/**
 * Consolidates buffers from input list, applies gzip, returns result as single buffer
 * @param inList portion of chunks to be consolidated
 * @param outChunkMaxLength estimated length of output chunk. 
 *        !!! to avoid pipe deadlock, this length to be sufficient
 *        !!! for consolidated data after gzip 
 */
private static ByteBuffer zipBuffers(List<ByteBuffer> inList, int outChunkMaxLength) {
    try {
        PipedInputStream pis = new PipedInputStream(outChunkMaxLength);
        GZIPOutputStream gos = new GZIPOutputStream(new PipedOutputStream(pis));

        for (var buf: inList) {
            gos.write(buf.array());
        }
        gos.close();
        byte[] outBytes = new byte[pis.available()];
        pis.read(outBytes);
        pis.close();
        return ByteBuffer.wrap(outBytes);
    } catch (IOException e) {
        throw new RuntimeException(e.getMessage(), e);
    }       
}

private static void test() {
    int inLength  = ... // actual full length of source data
    Flux<ByteBuffer> source = ... // your source Flux
    
    // these are parameters for your adjustment
    int consolidationCount = 5;
    int outChunkMaxLength= 30 * 1024;

    Flux<ByteBuffer> result = gzipFlux(source,consolidationCount, outChunkMaxLength);           

    int outLen = result.reduce(0, (res, bb) -> res + bb.array().length).block();
    System.out.println("ratio=" + (double)inLength/outLen);
}
}