如何结合Vert.x WebClient与RecordParser处理大响应？

我尝试使用RecordParser处理WebClient返回的大响应。

Vert.x官方文档指出：

当预期会收到大响应时，请使用BodyCodec.pipe。该Body Codec会将响应体缓冲泵入WriteStream，并在异步结果响应中告知操作成功或失败

但我未找到将WriteStream直接传入RecordParser的便捷方法。我编写了以下简化代码并可正常运行，但自行实现此类桥接逻辑易因异步协议出错，存在潜在bug。想了解Vert.x是否提供此类原生集成？

RecordParser parser = RecordParser.newDelimited("\n", b -> log.info("r={}", b.toString()));
RecordParserWriteStream bridge = new RecordParserWriteStream(parser);
client
    .get(sut.actualPort(), "localhost", "/stream?file=stream2.txt")
    .as(BodyCodec.pipe(bridge))
    .send(
        ar -> {
          if (ar.succeeded()) {
            ctx.completeNow();
          } else {
            ctx.failNow(ar.cause());
          }
        });


@Slf4j
@RequiredArgsConstructor
public class RecordParserWriteStream implements WriteStream<Buffer> {

    private final RecordParser recordParser;

    @Override
    public WriteStream<Buffer> exceptionHandler(@Nullable Handler<Throwable> handler) {
        recordParser.exceptionHandler(handler);
        return this;
    }

    @Override
    public void write(Buffer data, Handler<AsyncResult<Void>> handler) {
        log.info("write {}", data.length());
        recordParser.handle(data);
        Promise<Void> promise = Promise.promise();
        promise.complete();
        handler.handle(promise.future());
    }

    @Override
    public void end(Handler<AsyncResult<Void>> handler) {
        Promise<Void> promise = Promise.promise();
        promise.complete();
        handler.handle(promise.future());
    }

    @Override
    public boolean writeQueueFull() {
        return false;
    }

    @Override
    public WriteStream<Buffer> drainHandler(@Nullable Handler<Void> handler) {
        return this;
    }

    @Override
    public Future<Void> write(Buffer data) {
        throw new UnsupportedOperationException();
    }

    @Override
    public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
        throw new UnsupportedOperationException();
    }
}

我在旧Stack Overflow回答中看到有人推荐使用HttpClient，但官方文档仍推荐WebClient搭配BodyCodec.pipe/WriteStream。

使用HttpClient的实现代码如下：

RecordParser parser = RecordParser.newDelimited("\n", h -> log.info("r={}", h.toString()));
client
    .request(HttpMethod.GET, sut.actualPort(), "localhost", "/stream?file=stream1.txt")
    .compose(HttpClientRequest::send)
    .onComplete(
        ar -> {
          if (ar.succeeded()) {
            HttpClientResponse response = ar.result();
            response.handler(parser);
            response.endHandler(e -> ctx.completeNow());
          } else {
            ctx.failNow(ar.cause());
          }
        });

请问调用HTTP服务并将响应传入RecordParser的最佳方式是什么？

内容的提问来源于stack exchange，提问作者kodstark