Integrate reactive routes and Mutiny

[cescoffier]

Better integration of reactive routes and mutiny to allow routes to return Uni/Multi.

[murphye]

I was playing with this previously. This seems to work, but I am wondering what improvements you have in mind...

    @Route(path = "/multi", methods = HttpMethod.GET)
    void multi(RoutingExchange ex) {
        Multi<String> source = Multi.createFrom().items("a", "b", "c");

        var response = ex.response();
        response.setChunked(true);
        source.onCompletion().invoke(response::end).subscribe().with(response::write);
    }

[murphye]

Another example using SSE. This will send messages to all consumers of the /stream endpoint from the EventBus.

    @Route(path = "/stream", methods = HttpMethod.GET)
    void stream(RoutingExchange ex) {
        var response = ex.response();
        response.setChunked(true);
        response.headers().add("Content-Type", "text/event-stream;charset=UTF-8");
        response.headers().add("Connection", "keep-alive");

        eventBus.<String>consumer("sample").toMulti()
               .onItem().apply(m -> m.body())
               .onItem().apply(b -> "data: " + b + "\n\n")
               .subscribe().with(response::write);
    }

[cescoffier]

The idea is to return the Uni and Multi directly and avoid the user-managed subscription.

[murphye]

OK. Maybe the functionality could be extended a bit:

1. Automatically handle subscibe and onCompletion
2. Automatically handle the onError and return appropriate response (500 error?)
3. Automatically setChunked(true) for Multi
4. Automatically add data: and \n\n for optional SSE
5. Automatically headers().add text/event-stream and keep-alive for optional SSE
6. Automatically return body() from eventBus.<String>consumer("sample").toMulti() when EventBus is used

[cescoffier]

It's exactly what I had in mind!

[cescoffier]

@mkouba and I have done a first integration - already in master.