Support for Flux<Rendering> and Flux<ModelAndView> for @RequestMapping handler methods

[dsyer]

A handler method returning Flux<T> where T is "something that can render a template" would be quite useful, especially in the light of the popularity and ease of use of things like the @hotwired/turbo and htmx.org JavaScript modules. Those client libraries both have support for "streams" of HTML elements coming from the server, which get transcluded into the "main" page on the client. They also both support SSE streams containing HTML data. It would be nice to be able to render in both styles.

Webflux and MVC currently have support for SSE. E.g. with Webflux you can return Flux<String> or Flux<ServerSentEvent> from a handler method, but in both cases you have to render the data yourself. It would be handy to be able to delegate the rendering to a template engine, so Rendering (WebFlux) and ModelAndView (MVC) seem like a good fit. Thymeleaf also has some native (if a bit clumsy) support via ReactiveDataDriverContextVariable, so there is some prior art there. You could see this feature request as a generalization of that.

Simple example for Turbo on Webflux (for MVC just replace Rendering with ModelAndView) and SSE:

@GetMapping(path = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Rendering> stream() {
    return Flux.interval(Duration.ofSeconds(2)).map(value -> event(value));
}

private Rendering event(long value) {
    return Rendering.view("stream").modelAttribute("value", value)
        .modelAttribute("time", System.currentTimeMillis()).build();
}

with a template (e.g. in mustache but could be thymeleaf etc.):

<turbo-stream action="append" target="load">
    <template>
        <div>Index: {{value}} Time: {{time}}</div>
    </temlate>
</turbo-stream>

The result would be an infinite stream, e.g.:

data: <turbo-stream action="append" target="load">
data: ...

data: <turbo-stream action="append" target="load">
data: ...

...

An example with HTMX and the HTML "stream" would be the same controller but with a different produces media type:

@GetMapping(path = "/updates", produces="text/vnd.turbo-stream.html")
public Flux<Rendering> stream() {
    return Flux.just(event("one"), event("two");
}

private Rendering event(String id) {
    return Rendering.view("update").modelAttribute("id", id)
        .modelAttribute("time", System.currentTimeMillis()).build();
}

with a template (e.g. in mustache but could be thymeleaf etc.):

<div htmx-swap-oob="true" id="{{id}}">
    <div>Time: {{time}}</div>
</div>

The result would be a concatenation of the 2 divs:

<div htmx-swap-oob="true" id="one">
    <div>Time: 1346876956</div>
</div>
<div htmx-swap-oob="true" id="two">
    <div>Time: 1346876987</div>
</div>

[dsyer]

Here's an implementation that works, but probably needs a lot of polish:

@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class CompositeViewRenderer implements HandlerResultHandler {

    private final ViewResolver resolver;

    public CompositeViewRenderer(ViewResolver resolver) {
        this.resolver = resolver;
    }

    @Override
    public boolean supports(HandlerResult result) {
        if (Publisher.class.isAssignableFrom(result.getReturnType().toClass())) {
            if (Rendering.class.isAssignableFrom(result.getReturnType().getGeneric(0).toClass())) {
                return true;
            }
        }
        return false;
    }

    @Override
    public Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
        String[] methodAnnotation = ((InvocableHandlerMethod) result.getHandler())
                .getMethodAnnotation(RequestMapping.class).produces();
        MediaType type = methodAnnotation.length > 0 ? MediaType.valueOf(methodAnnotation[0]) : MediaType.TEXT_HTML;
        exchange.getResponse().getHeaders().setContentType(type);
        boolean sse = MediaType.TEXT_EVENT_STREAM.includes(type);
        @SuppressWarnings("unchecked")
        Flux<Rendering> renderings = Flux.from((Publisher<Rendering>) result.getReturnValue());
        final ExchangeWrapper wrapper = new ExchangeWrapper(exchange);
        return exchange.getResponse().writeAndFlushWith(render(wrapper, renderings)
                .map(buffers -> transform(exchange.getResponse().bufferFactory(), buffers, sse)));
    }

    private Publisher<DataBuffer> transform(DataBufferFactory factory, Publisher<DataBuffer> buffers, boolean sse) {
        if (sse) {
            buffers = Flux.from(buffers).map(buffer -> prefix(buffer, factory.allocateBuffer(buffer.capacity())));
        }
        // Add closing empty lines
        return Flux.from(buffers).map(buffer -> buffer.write("\n\n", StandardCharsets.UTF_8));
    }

    private DataBuffer prefix(DataBuffer buffer, DataBuffer result) {
        String body = buffer.toString(StandardCharsets.UTF_8);
        body = "data:" + body.replace("\n", "\ndata:");
        result.write(body, StandardCharsets.UTF_8);
        DataBufferUtils.release(buffer);
        return result;
    }

    private Flux<Flux<DataBuffer>> render(ExchangeWrapper exchange, Flux<Rendering> renderings) {
        return renderings.flatMap(rendering -> render(exchange, rendering));
    }

    private Publisher<Flux<DataBuffer>> render(ExchangeWrapper exchange, Rendering rendering) {
        Mono<View> view = null;
        if (rendering.view() instanceof View) {
            view = Mono.just((View) rendering.view());
        } else {
            view = resolver.resolveViewName((String) rendering.view(), exchange.getLocaleContext().getLocale());
        }
        return view.flatMap(actual -> actual.render(rendering.modelAttributes(), null, exchange))
                .thenMany(Flux.defer(() -> exchange.release()));
    }

    static class ExchangeWrapper extends ServerWebExchangeDecorator {

        private ResponseWrapper response;

        protected ExchangeWrapper(ServerWebExchange delegate) {
            super(delegate);
            this.response = new ResponseWrapper(super.getResponse());
        }

        @Override
        public ServerHttpResponse getResponse() {
            return this.response;
        }

        public Flux<Flux<DataBuffer>> release() {
            Flux<Flux<DataBuffer>> body = response.getBody();
            this.response = new ResponseWrapper(super.getResponse());
            return body;
        }

    }

    static class ResponseWrapper extends ServerHttpResponseDecorator {

        private Flux<Flux<DataBuffer>> body = Flux.empty();

        public Flux<Flux<DataBuffer>> getBody() {
            return body;
        }

        public ResponseWrapper(ServerHttpResponse delegate) {
            super(delegate);
        }

        @Override
        public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
            return writeAndFlushWith(Mono.just(body));
        }

        @Override
        public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
            Flux<Flux<DataBuffer>> map = Flux.from(body).map(publisher -> Flux.from(publisher));
            this.body = this.body.concatWith(map);
            return Mono.empty();
        }

    }

}

UPDATE: the above happens to work for the Spring Boot MustacheReactiveView but is doesn't work with ThymeleafReactiveView because the response gets committed after the first chunk (really?) so when the second or subsequent chunks try to set response headers (even if they would be the same - e.g. the encoding or content type) you get an exception.

[dsyer]

Here's a sample app with Webflux and Thymeleaf: https://github.com/dsyer/spring-todo-mvc. The tests are green, but it doesn't work in the browser because the wrong stuff is rendered most of the time. Something concurrent there and not thread safe in the app maybe?

UPDATE: I fixed the browser rendering. All working fine now (but still could use some polish I'm sure).

N.B. the main branch is webflux (code like the above), and the webmvc branch is Spring MVC.