如何解压缩Flux（以及如何编写一个）？

Question

我需要在没有中间存储的情况下读写压缩（GZIP）流。目前，我使用Springrestemplate进行编写，使用Apache-HTTP-client进行读取（有关restemplate不能用于读取大数据流的原因的解释，请参阅我的回答）。实现相当简单，我在响应上加了一个gzip输入流，然后继续。

现在，我想转而使用Spring5WebClient（只是因为我不喜欢现状）。然而，WebClient本质上是被动的，处理流量

• 另见ReactorNetty第251期

我承认对（去）压缩知之甚少，但是，我做了我的研究，但是网上可用的材料似乎都没有特别有用。

java nio直接缓冲区上的压缩

用nio编写GZIP文件

从FileChannel（JavaNIO）读取GZIP文件

（de）使用NIO压缩文件

可迭代gzip在Java中放气/充气




Answer 1

注意这一点，因为它让我有点困惑-API从5.1开始有点变化。

我有一个类似于ChannelInboundHandler的公认答案的设置：

public class GzipJsonHeadersHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpResponse
                && !HttpStatus.resolve(((HttpResponse) msg).status().code()).is1xxInformational()) {
            HttpHeaders headers = ((HttpResponse) msg).headers();
            headers.clear();
            headers.set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
            headers.set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
        }
        ctx.fireChannelRead(msg);
    }
}

（为了简单起见，我需要的标题值只是在那里硬编码的，否则它是相同的。）

然而，注册它是不同的：

WebClient.builder()
    .clientConnector(
            new ReactorClientHttpConnector(
                    HttpClient.from(
                            TcpClient.create()
                                    .doOnConnected(c -> {
                                        c.addHandlerFirst(new HttpContentDecompressor());
                                        c.addHandlerFirst(new HttpResponseHeadersHandler());
                                    })
                    ).compress(true)
            )
    )
    .build();

Netty现在似乎在系统列表之外（和之后）维护一个处理程序的用户列表，并且只将处理程序放在用户列表的前面。因此，它需要显式调用HttpContentDecompressor，以确保它在插入正确头的处理程序之后明确执行。

Answer 2

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);
    }
}

然后我创建一个ClientHttpConnector与WebClient一起使用，并在NettyContextInit之后添加处理程序：

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());

当然，对于没有经过GZIP压缩的响应，这将失败，因此我仅将这个WebClient实例用于特定用例，我确信响应是压缩的。

编写很容易：Spring有一个资源编码器，因此可以简单地将InputStream转换为InputStreamResource，瞧！