Let users control reactive SSE stream completion [SPR-14578]

[spring-projects-issues]

Jose Antonio Iñigo opened SPR-14578 and commented

Given this Spring 4.2 server sent events code example:

@RestController
public class AccountController {

    @RequestMapping("/accounts/alerts")
    public SseEmitter getAccountAlerts() {

        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

        Thread t1 = new Thread(() ->{
            try {
                int i = 0;
                while(++i<=3){
                    Thread.sleep(1000);
                    emitter.send(new Alert((long)i, "Alert message"+i));
                }
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        });
        t1.start();

        return emitter;
    }

}

It could be translated to spring reactive like this:

@RestController
public class AccountsController {

    @GetMapping(value="/accounts/alerts", produces="text/event-stream")
    public Flux<Alert> getAccountAlerts() {
    @GetMapping(value="/accounts/alertsStreaming", produces="text/event-stream")
    public Flux<Alert> getAccountAlertsNoPathVariableStreaming() {
        return Flux.range(1, 3)
            .map((Integer i) -> {
                    return new Alert((long)i, "Alert message"+i);
                })
            .delayMillis(1000)
            .log();
    }
} 

In the spring 4.2 case, after sending the 3 events, emitter.complete() frees the HTTP connection.

However, in the reactive version, the connection is held open 27 seconds after the 3 events are sent (in 3 seconds). Shouldn't it be closed when the Flux completes?

---

Affects: 5.0 M1

Referenced from: commits https://github.com/spring-projects/spring-framework/commit/06d4bb6a1a42a50cba2f0e39f8ef405fd46f53cd

[spring-projects-issues]

Sébastien Deleuze commented

Thanks for raising this interesting question, there is indeed currently different behaviors about how complete signal is handled between Spring Framework 4.x SseEmitter and Spring Framework 5.x Flux<?>.

One important point to have in mind is that SSE clients expect the connection to remain open. If the connection closes, they automatically re-open it multiple times, leading to the server to re-send the data to the client again and again.

With SseEmitter, the call to complete() is manual and closes the HTTP connection. In the reactive version SseEventHttpMessageWriter indeed keeps the connection opened in order to avoid sending multiple times the data to the browser when Flux is a cold stream.

I initially made that choice in M1 because in the case of Flux, the complete signal is less explicit than with SseEmitter, and I thought it would be confusing for the users to see the reconnect + event duplication for non infinite streams, but this is of course open for discussion. The ideal solution would be to prevent the browser to reconnect if the Flux has completed, according to the spec this is doable by returning a 204 - NO CONTENT status code when it reconnects, but since we have no way to link both requests that does not seems possible.

That let us with a choice between current behavior or closing the connection as you suggested and let the user perform a concatWith(Flux.never()) if he want to avoid the reconnect behavior with cold streams.

Any thoughts Jose Antonio Iñigo Rossen Stoyanchev?

[spring-projects-issues]

Jose Antonio Iñigo commented

Thanks for the explanation, I know see why it behaves that way.

I would like to point out some things:

• If I understood it well, according to the spec, even with a 204 status code the clients will always reconnect:

HTTP 204 No Content, and 205 Reset Content responses are equivalent to 200 OK responses with the right MIME type but no content, and thus must reset the connection.

When a user agent is to reset the connection, the user agent must set the readyState attribute to CONNECTING, queue a task to fire a simple event named error at the EventSource object, and then fetch the event source resource again after a delay equal to the reconnection time of the event source, from the same origin as the original request triggered by the EventSource() constructor. Only if the user agent resets the connection does the connection get opened anew!

So, using EventSource, even in the Spring 4.2 emmiter.complete() case, the browser reconnects when the stream has finished. When working with cold observables I guess we'll see the same behaviour. So finishing the connection from the server side doesn't really matter in this situation, since the client will always reset (restart) it.

• Now the case that interests me: from a reactive REST service client point of view (org.springframework.web.client.reactive.WebClient) you may be interested in receiving the results as they are being streamed. In case you request something from a service that returns a finite observable, the desired behaviour would be to have the connection closed when you get all the results. What is happening now is that the client hangs until the server closes the connection due to the default timeout. So in this case we need the server to close the connection when the Flux completes.

[spring-projects-issues]

Sébastien Deleuze commented

I am not sure yet what is the right default behavior yet, but having WebClient returning a Flux that never completes for SSE is maybe something that makes sense. If we build a dedicated SSE client extractor, we could even imagine it reconnects to the server if the connection is lost, as specified in the spec.

If you want to just retreive a finite stream, one possibility would be to use application/json, since with Spring Web Reactive the same endpoint returning Flux<Foo> can be requested with application/json media type (works even if this is a stream, and here you will get the complete event) or with text/event-stream if you want the SSE behavior.

I think SSE is really designed for hot streams, and we need to find what would be the best default behavior for "collection like streams that completes" : keep the connection opened as currently, close the connection that will be reopen by the client as you suggested, log a warning or an error when a complete signal is received on an SSE endpoint since that could lead to some unexpected behavior on client side (multiple reconnection, duplicated data) ... Let's see what is Rossen Stoyanchev take on this when he will be back.

[spring-projects-issues]

Sébastien Deleuze commented

After discussion with the team, we agree that the best default behavior is to let users control themselves the complete signal, even if it could be confusing for newcomers. This commit removes the automatic concatWith(Flux.never()).

[spring-projects-issues]

Jose Antonio Iñigo commented

Thank you guys!