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

Question

I'm using WebClient and custom BodyExtractorclass for my spring-boot application

WebClient webLCient = WebClient.create(); webClient.get()    .uri(url, params)    .accept(MediaType.APPLICATION.XML)    .exchange()    .flatMap(response -> {      return response.body(new BodyExtractor());    }) 

BodyExtractor.java

@Override public Mono<T> extract(ClientHttpResponse response, BodyExtractor.Context context) {   Flux<DataBuffer> body = response.getBody();   body.map(dataBuffer -> {     try {       JaxBContext jc = JaxBContext.newInstance(SomeClass.class);       Unmarshaller unmarshaller = jc.createUnmarshaller();        return (T) unmarshaller.unmarshal(dataBuffer.asInputStream())     } catch(Exception e){        return null;     }   }).next(); } 

Above code works with small payload but not on a large payload, I think it's because I'm only reading a single flux value with next and I'm not sure how to combine and read all dataBuffer.

I'm new to reactor, so I don't know a lot of tricks with flux/mono.

Answer 1

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.

Answer 2

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.