[FIXED] Send a ServerSentEvent from another Method

Issue

I'm trying to implement a Server Sent Event Controller for updating my Web Browser Client with the newest Data to display.

This is my current Controller which sends the list of my data every 5 seconds. I want to send a SSE everytime I save my data in another service. I read about using a channel, but how do I consume it with a Flux?

@GetMapping("/images-sse")
fun getImagesAsSSE(
    request: HttpServletRequest
): Flux<ServerSentEvent<MutableList<Image>>> {
    val subdomain = request.serverName.split(".").first()
    return Flux.interval(Duration.ofSeconds(5))
        .map {
            ServerSentEvent.builder<MutableList<Image>>()
                .event("periodic-event")
                .data(weddingService.getBySubdomain(subdomain)?.pictures).build()
        }
}

Solution

Example code for controller:

package sk.qpp;

import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.concurrent.atomic.AtomicLong;

@Controller
@Slf4j
public class ReactiveController {
    record SomeDTO(String name, String address) {
    }

    private final Sinks.Many<SomeDTO> eventSink = Sinks.many().multicast().directBestEffort();

    @RequestMapping(path = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<SomeDTO>> sse() {
        final AtomicLong counter = new AtomicLong(0);
        return eventSink.asFlux()
                .map(e -> ServerSentEvent.builder(e)
                        .id(counter.incrementAndGet() + "")
                        //.event(e.getClass().getName())
                        .build());
    }

    // note, when you want this to work in production, ensure, that http request is not being cached on its way, using POST method for example.
    @ResponseStatus(HttpStatus.OK)
    @ResponseBody
    @GetMapping(path = "/sendSomething", produces = MediaType.TEXT_PLAIN_VALUE)
    public String sendSomething() {
        this.eventSink.emitNext(
                new SomeDTO("name", "address"),
                (signalType, emitResult) -> {
                    log.warn("Some event is being not send to all subscribers. It will vanish...");
                    // returning false, to not retry emitting given data again.
                    return false;
                }
        );
        return "Have a look at /sse endpoint (using \"curl http://localhost/sse\" for example), to see events in realtime.";
    }
}

Sink is used as some "custom flux", where you can put anything (using emitNext), and take from it (using asFlux() method).

After setting up sample controller, open http://localhost:9091/sendSomething in your browser (i.e. do GET request on it) and in console issue command curl http://localhost:9091/sse to see your sse events (after each get request, new should come). It is possible also to see sse events directly in chromium browser. Firefox does try to download and save it to filesystem as file (works also).

Answered By - Lubo
Answer Checked By - Marie Seifert (JavaFixing Admin)