server-sent-events - 如何通过 REST 将传入数据转发到 Quarkus 中的 SSE 流

Question

在我的设置中，我想通过 SSE channel (服务器发送事件)转发某些状态更改。状态更改通过调用 REST 端点启动。因此，我需要将传入的状态更改转发到 SSE 流。

在 Quarkus 中完成此任务的最佳/最简单方法是什么。

我能想到的一种解决方案是使用 EventBus ( https://quarkus.io/guides/reactive-messaging )。 SSE 端点将订阅状态更改并将其推送到 SSE channel 。状态更改端点发布适当的事件。

这是一个可行的解决方案吗？还有其他(更简单的)解决方案吗？在任何情况下我都需要使用 react 性的东西来完成这个吗？

非常感谢任何帮助!

Answer 1

最简单的方法是使用 rxjava 作为流提供程序。首先，您需要添加 rxjava 依赖项。它可以来自 quarkus 中的响应式(Reactive)依赖项，例如 kafka，也可以直接使用它(如果您不需要任何流库):

        <dependency>
            <groupId>io.reactivex.rxjava2</groupId>
            <artifactId>rxjava</artifactId>
            <version>2.2.19</version>
        </dependency>

这是关于如何每秒发送随机 double 值的示例:

    @GET
    @Path("/stream")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @SseElementType("text/plain")
    public Publisher<Double> stream() {
        return Flowable.interval(1, TimeUnit.SECONDS).map(tick -> new Random().nextDouble());
    }

我们创建了新的 Flowable，它将每秒触发一次，并且在每个刻度上我们生成下一个随机双倍。调查有关如何创建 Flowable 的任何其他选项，例如 Flowable.fromFuture()使其适应您的特定代码逻辑。

每次查询此端点时，上面的 P.S 代码都会生成新的 Flowable，我这样做是为了节省空间，在您的情况下，我假设您将拥有一个事件源，您可以构建一次并在每次查询端点时使用相同的实例