What happens when flux is returned from spring web controller?

Question

I am comparatively new to reactive APIs and was curious about what was happening behind the scenes when we return a Flux from a web controller.

According to spring-web documentation

Reactive return values are handled as follows:

A single-value promise is adapted to, similar to using DeferredResult. Examples include Mono (Reactor) or Single (RxJava).

A multi-value stream with a streaming media type (such as application/stream+json or text/event-stream) is adapted to, similar to using ResponseBodyEmitter or SseEmitter. Examples include Flux (Reactor) or Observable (RxJava). Applications can also return Flux or Observable.

A multi-value stream with any other media type (such as application/json) is adapted to, similar to using DeferredResult<List<?>>.

I created two APIs as below:

@GetMapping("/async-deferredresult")
public DeferredResult<List<String>> handleReqDefResult(Model model) {
    LOGGER.info("Received async-deferredresult request");
    DeferredResult<List<String>> output = new DeferredResult<>();

    ForkJoinPool.commonPool().submit(() -> {
        LOGGER.info("Processing in separate thread");
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10000   ; i++) {
            list.add(String.valueOf(i));
        }
        output.setResult(list);
    });

    LOGGER.info("servlet thread freed");
    return output;
}


@GetMapping(value = "/async-flux",produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    LOGGER.info("Received async-deferredresult request");
    List<String> list = new ArrayList<>();
    list.stream();
    for (int i = 0; i < 10000   ; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list);
}

So the exception was that both APIs should behave same as multi-value stream(Flux) should have similar behavior to that of a returning a DeferredResult.
But in API where deferred result was returned, whole list was printed in one go on browser where as in API where Flux was returned the numbers where printed sequentially(one by one).
What exactly is happening when I am returning Flux from controller ?

Kalpa Kalpa · Accepted Answer · 2019-11-23T21:27:08

When we return a Flux from a service endpoint many things can happen. But I assume you want to know what is happening when Flux observed as stream of events from client of this endpoint.

Scenario One: By adding 'application/json' as the content type of the endpoint Spring will communicate to the client to expect JSON body.

@GetMapping(value = "/async-flux", produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    List<String> list = new ArrayList<>();
    for (int i = 0; i < 10000; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list);
}

The output at the client will be the whole set of numbers in one go. And once the response delivered the connection will be closed. Even though you have used Flux as the response type, you are still bound the laws of how HTTP over TCP/IP works. The endpoint got a HTTP request, execute the logic and respond with HTTP response containing final result.

As a result, you do not see the real value of a reactive api.

Scenario Two: By adding 'application/stream+json' as the content type of the endpoint, Spring starts to treat the resulting events of the Flux stream as individual JSON items. When an item is emitted is gets serialised, the HTTP response buffer is flushed, and the connection from the server to client keep open up until the event sequence get completed.

To get that working we can slightly modify your original code as follows.

@GetMapping(value = "/async-flux",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    List<String> list = new ArrayList<>();
    for (int i = 0; i < 10000   ; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list)
            // we have 1 sec delay to demonstrate the difference of behaviour. 
            .delayElements(Duration.ofSeconds(1));
}

This time we can see the real value of reactive api endpoint where it is able to deliver results to it's client as date get available.

You can find more details about how to build reactive REST APIs at https://medium.com/@senanayake.kalpa/building-reactive-rest-apis-in-java-part-1-cd2c34af55c6 https://medium.com/@senanayake.kalpa/building-reactive-rest-apis-in-java-part-2-bd270d4cdf3f

What happens when flux is returned from spring web controller?

1 Answers