How to decompress a Flux<DataBuffer> (and how to write one)?

Question

I have a requirement to read and write compressed (GZIP) streams without intermediate storage. Currently, I'm using Spring RestTemplate to do the writing, and Apache HTTP client to do the reading (see my answer here for an explanation of why RestTemplate can't be used for reading large streams). The implementation is fairly straightforward, where I slap a GZIPInputStream on the response InputStream and move on.

Now, I'd like to switch to using Spring 5 WebClient (just because I'm not a fan of status quo). However, WebClient is reactive in nature and deals with Flux<Stuff>; I believe it's possible to get a Flux<DataBuffer>, where DataBuffer is an abstraction over ByteBuffer. Question is, how do I decompress it on the fly without having to store the full stream in memory (OutOfMemoryError, I'm looking at you), or writing to local disk? It's worth mentioning that WebClient uses Netty under the hood.

• Also see Reactor Netty issue-251.
• Also related to Spring integration issue-2300.

I'll admit to not knowing much about (de)compression, however, I did my research, but none of the material available online seemed particularly helpful.

compression on java nio direct buffers

Writing GZIP file with nio

Reading a GZIP file from a FileChannel (Java NIO)

(de)compressing files using NIO

Iterable gzip deflate/inflate in Java

Answer 1

public class HttpResponseHeadersHandler extends ChannelInboundHandlerAdapter {
    private final HttpHeaders httpHeaders;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpResponse &&
                !HttpStatus.resolve(((HttpResponse) msg).status().code()).is1xxInformational()) {
            HttpHeaders headers = ((HttpResponse) msg).headers();

            httpHeaders.forEach(e -> {
                log.warn("Modifying {} from: {} to: {}.", e.getKey(), headers.get(e.getKey()), e.getValue());
                headers.set(e.getKey(), e.getValue());
            });
        }
        ctx.fireChannelRead(msg);
    }
}

Then I create a ClientHttpConnector to use with WebClient and in afterNettyContextInit add the handler:

ctx.addHandlerLast(new ReadTimeoutHandler(readTimeoutMillis, TimeUnit.MILLISECONDS));
ctx.addHandlerLast(new Slf4JLoggingHandler());
if (forceDecompression) {
    io.netty.handler.codec.http.HttpHeaders httpHeaders = new ReadOnlyHttpHeaders(
            true,
            CONTENT_ENCODING, GZIP,
            CONTENT_TYPE, APPLICATION_JSON
    );
    HttpResponseHeadersHandler headersModifier = new HttpResponseHeadersHandler(httpHeaders);
    ctx.addHandlerFirst(headersModifier);
}
ctx.addHandlerLast(new HttpContentDecompressor());

This, of course, would fail for responses that are not GZIP compressed, so I use this instance of WebClient for a particular use case only, where I know for sure that the response is compressed.

Writing is easy: Spring has a ResourceEncoder, so InputStream can simply be converted to InputStreamResource, and voila!