Reactive Web Streaming

Following on from yesterday’s post about reactive programming using Spring Reactive Web, I thought it would be interest to expand upon the example. In this post we’ll take a look at one of the cooler aspects of the new reactive model, result streaming.

What is result streaming?

Result streaming, or data streaming, is a way of returning partial results incrementally as they are gathered rather than in a single burst.

The easiest way to see these differences is to compare these two videos. On the left a list of three items is retrieved in the old style. On the right the list is streamed back to the client.

Behind the scenes, each of the list items is retrieved using this method

private Mono<Response> getSomethingSlow(String name) {
    return Mono.just(Response.of(name)).delaySubscription(Duration.ofSeconds(1));
}

This returns a single response, but adds a second of delay. This is supposed to be roughly analogous to performing a slow database query, or any other kind of high latency request.

The implementation code

Because this is a first class feature of Spring Reactive Web, implementing it is extremely easy.

private Flux<Response> getListStreaming() {
    return Flux.concat(
        getSomethingSlow("test1"),
        getSomethingSlow("test2"),
        getSomethingSlow("test3")
    );
}

The method returns a flux, which the reactive response type similar to a list. It can contain 0 to n items. Unlike a list it can be unbounded, and return results in response to external events like timers. This is in contrast with the Mono type which can return either zero or one item.

To route this back to the user, I used the following router configuration

@Bean
public RouterFunction<ServerResponse> route() {
    return RouterFunctions.route(GET("/list/complete"), r -> getListBlocking())
            .andRoute(GET("/list/streamed"), r -> ServerResponse.ok().contentType(MediaType.APPLICATION_STREAM_JSON).body(getListStreaming(), Response.class));
}

If you don’t like the RouterFunction syntax, you can also use the Spring REST style annotations. The following code produces the same results.

@RestController
public class TestController {

    @GetMapping(value = "/list/streamed", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    private Flux<Response> getListStreaming() {
        return Flux.concat(
            getSomethingSlow("test1"),
            getSomethingSlow("test2"),
            getSomethingSlow("test3")
        );
    }
}

ucosty:io