从另一个方法发送 ServerSentEvent答案

【问题标题】：Send a ServerSentEvent from another Method从另一个方法发送 ServerSentEvent
【发布时间】：2022-08-14 05:03:58
【问题描述】：

我正在尝试实现一个服务器发送事件控制器，以使用要显示的最新数据更新我的 Web 浏览器客户端。

这是我当前的控制器，它每 5 秒发送一次我的数据列表。每次我将数据保存在另一个服务中时，我都想发送一个 SSE。我阅读了有关使用通道的信息，但是如何使用 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()
        }
}

使用处理器或接收器projectreactor.io/docs/core/release/reference/#processors
不要使用处理器，因为它已被弃用。水槽是个好主意。我会发布样本。

标签： spring-boot kotlin spring-webflux project-reactor

【解决方案1】：

控制器的示例代码：

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 用作一些“自定义通量”，您可以在其中放置任何东西（使用 emitNext），并从中取出（使用 asFlux() 方法）。

设置示例控制器后，在浏览器中打开 http://localhost:9091/sendSomething（即对其执行 GET 请求）并在控制台发出命令 curl http://localhost:9091/sse 以查看您的 sse 事件（在每个 get 请求之后，新的应该会出现） .也可以直接在 chromium 浏览器中查看 sse 事件。 Firefox 确实尝试下载并将其作为文件保存到文件系统（也可以）。

【讨论】：

非常感谢！这真的帮助我让我的代码工作

【解决方案2】：

我终于让它工作了。我还使用 cookie 添加了用户特定的更新。

这是我的 SSE 控制器

@RestController
@RequestMapping("/api/sse")
class SSEController {

    val imageUpdateSink : Sinks.Many<Wedding> = Sinks.many().multicast().directBestEffort()
    @GetMapping("/images")
    fun getImagesAsSSE(
        request: HttpServletRequest
    ): Flux<ServerSentEvent<MutableList<Image>>> {
        val counter: AtomicLong = AtomicLong(0)
        return imageUpdateSink.asFlux()
            .filter { wedding ->
                val folderId = request.cookies.find {cookie ->
                    cookie.name == "folderId"
                }?.value

                folderId == wedding.folderId
            }.map { wedding ->
                     ServerSentEvent.builder<MutableList<Image>>()
                        .event("new-image")
                        .data(
                            wedding.pictures
                        ).id(counter.incrementAndGet().toString())
                        .build()
            }
    }
}

在我的数据更新的服务中：

val updatedImageList = weddingRepository.findByFolderId(imageDTO.folderId)
sseController.imageUpdateSink.tryEmitNext(
    updatedImageList
)

我的 Javascript 看起来像这样：

document.cookie = "folderId=" + [[${wedding.folderId}]]
const evtSource = new EventSource("/api/sse/images")
evtSource.addEventListener("new-image", function(alpineContext){
    return function (event) {
        console.log(event.data)
        alpineContext.images = JSON.parse(event.data)
    };
}(this))

【讨论】：