将对 OutputStream 的写入转换为可供 ServerResponse 使用的 Flux<DataBuffer>答案

【问题标题】：Convert writes to OutputStream into a Flux<DataBuffer> usable by ServerResponse将对 OutputStream 的写入转换为可供 ServerResponse 使用的 Flux<DataBuffer>
【发布时间】：2018-03-22 10:30:04
【问题描述】：

我有一个旧库，我必须使用它来检索文件。这个遗留库不会在 InputStream 中返回，就像您通常期望的那样阅读内容，但它希望它被传递一个开放的 OutputStream，它可以写入。

我必须编写一个 Webflux REST 服务，将这个 OutputStream 写入 org.springframework.web.reactive.function.server.ServerResponse 主体。

legacyLib.BlobRead(outputStream); // writes the stream to an outputstream, that has to be provided by me, and somehow has to end up in the ServerResponse

由于我想将 Stream 直接传递给 ServerResponse，我可能必须做这样的事情，对吧？

ServerResponse.ok().body(magicOutpuStreamToFluxConverter(), DataBuffer.class);

【问题讨论】：

您能否修改您的问题以使这段代码 sn-p 更清晰一些？没有类型的伪代码使您更难理解哪一部分是您从库中获得的实际类型，哪一部分是您希望看到的解决方案的伪代码。

标签： spring-webflux

【解决方案1】：

这是 RequestHandler 中重要的部分。我遗漏了一些通常可能不需要的错误处理/异常捕获。请注意，我 publishedOn 一个不同的 Scheduler 用于读取（或者至少，这是我想要做的），所以这个阻塞读取不会干扰我的主事件线程：

    private Mono<ServerResponse> writeToServerResponse(@NotNull FPTag tag) {
        final long blobSize = tag.getBlobSize();
        return ServerResponse.ok()
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .body(Flux.<DataBuffer>create((FluxSink<DataBuffer> emitter) -> {
          // for a really big blob I want to read it in chunks, so that my server doesn't use too much memory
          for(int i = 0; i < blobSize; i+= tagChunkSize) {
            // new DataBuffer that is written to, then emitted later
            DefaultDataBuffer dataBuffer = new DefaultDataBufferFactory().allocateBuffer();
            try (OutputStream outputStream = dataBuffer.asOutputStream()) {
              // write to the outputstream of DataBuffer
              tag.BlobReadPartial(outputStream, i, tagChunkSize, FPLibraryConstants.FP_OPTION_DEFAULT_OPTIONS);
              // don't know if flushing is strictly neccessary
              outputStream.flush();
            } catch (IOException | FPLibraryException e) {
              log.error("Error reading + writing from tag to http outputstream", e);
              emitter.error(e);
            }
            emitter.next(dataBuffer);
          }
          // if blob is finished, send "complete" to my flux of DataBuffers
          emitter.complete();
        }, FluxSink.OverflowStrategy.BUFFER).publishOn(Schedulers.newElastic("centera")).doOnComplete(() -> closeQuietly(tag)), DataBuffer.class);
  
    }

【讨论】：