How to correctly read Flux and convert it to a single inputStream

by

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.

More Related Contents:

  1. Enable CORS in Spring 5 Webflux?
  2. block()/blockFirst()/blockLast() are blocking error when calling bodyToMono AFTER exchange()
  3. How to customize SpringWebFlux WebClient JSON deserialization?
  4. what does Mono.defer() do?
  5. Can I use SpringMvc and webflux together?
  6. Adding a retry all requests of WebClient
  7. Mono switchIfEmpty() is always called
  8. map vs flatMap in reactor
  9. Using env variable in Spring Boot’s application.properties
  10. @Autowired bean is null when referenced in the constructor of another bean
  11. Running code after Spring Boot starts
  12. How to customise the Jackson JSON mapper implicitly used by Spring Boot?
  13. How to respond with an HTTP 400 error in a Spring MVC @ResponseBody method returning String
  14. Difference between Role and GrantedAuthority in Spring Security
  15. Injecting beans into a class outside the Spring managed context
  16. Right way to write JSON deserializer in Spring or extend it
  17. Transactional saves without calling update method
  18. RESTful Authentication via Spring
  19. Spring classpath prefix difference
  20. Autowiring two beans implementing same interface – how to set default bean to autowire?
  21. Autowired Environment is null
  22. How can I reload properties file in Spring 4 using annotations?
  23. How can I register a secondary servlet with Spring Boot?
  24. Spring Security, secured and none secured access
  25. No String-argument constructor/factory method to deserialize from String value (”)
  26. How do I run a spring boot executable jar in a Production environment?
  27. spring data jpa @query and pageable
  28. Lombok @Builder and JPA Default constructor
  29. How do I create beans programmatically in Spring Boot?
  30. Spring 3.0 making JSON response using jackson message converter

Leave a Comment