How to correctly read Flux<DataBuffer> and convert it to a single inputStream

This is really not as complicated as other answers imply.

The only way to stream the data without buffering it all in memory is to use a pipe, as @jin-kwon suggested. However, it can be done very simply by using Spring's BodyExtractors and DataBufferUtils utility classes.

Example:

private InputStream readAsInputStream(String url) throws IOException {    PipedOutputStream osPipe = new PipedOutputStream();    PipedInputStream isPipe = new PipedInputStream(osPipe);    ClientResponse response = webClient.get().uri(url)        .accept(MediaType.APPLICATION.XML)        .exchange()        .block();    final int statusCode = response.rawStatusCode();    // check HTTP status code, can throw exception if needed    // ....    Flux<DataBuffer> body = response.body(BodyExtractors.toDataBuffers())        .doOnError(t -> {            log.error("Error reading body.", t);            // close pipe to force InputStream to error,            // otherwise the returned InputStream will hang forever if an error occurs            try(isPipe) {              //no-op            } catch (IOException ioe) {                log.error("Error closing streams", ioe);            }        })        .doFinally(s -> {            try(osPipe) {              //no-op            } catch (IOException ioe) {                log.error("Error closing streams", ioe);            }        });    DataBufferUtils.write(body, osPipe)        .subscribe(DataBufferUtils.releaseConsumer());    return isPipe;}

If you don't care about checking the response code or throwing an exception for a failure status code, you can skip the block() call and intermediate ClientResponse variable by using

flatMap(r -> r.body(BodyExtractors.toDataBuffers()))

instead.

A slightly modified version of Bk Santiago's answer makes use of reduce() instead of collect(). Very similar, but doesn't require an extra class:

Java:

body.reduce(new InputStream() {    public int read() { return -1; }  }, (s: InputStream, d: DataBuffer) -> new SequenceInputStream(s, d.asInputStream())).flatMap(inputStream -> /* do something with single InputStream */

Or Kotlin:

body.reduce(object : InputStream() {  override fun read() = -1}) { s: InputStream, d -> SequenceInputStream(s, d.asInputStream()) }  .flatMap { inputStream -> /* do something with single InputStream */ }

Benefit of this approach over using collect() is simply you don't need to have a different class to gather things up.

I created a new empty InputStream(), but if that syntax is confusing, you can also replace it with ByteArrayInputStream("".toByteArray()) instead to create an empty ByteArrayInputStream as your initial value instead.

Here comes another variant from other answers. And it's still not memory-friendly.

static Mono<InputStream> asStream(WebClient.ResponseSpec response) {    return response.bodyToFlux(DataBuffer.class)        .map(b -> b.asInputStream(true))        .reduce(SequenceInputStream::new);}static void doSome(WebClient.ResponseSpec response) {    asStream(response)        .doOnNext(stream -> {            // do some with stream            // close the stream!!!        })        .block();}