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

Question

在我的设置中，我希望通过SSE通道（服务器发送的事件）转发某些状态更改。状态更改是通过调用RESTendpoint来启动的。因此，我需要将传入的状态更改转发到SSE流。

在Quarkus中实现这一点的最佳/最简单方法是什么。

我能想到的一个解决方案是使用EventBus（https://quarkus.io/guides/reactive-messaging）。SSEendpoint将订阅状态更改并通过SSE通道推送它。状态更改endpoint发布适当的事件。

这是可行的解决方案吗？是否有其他（更简单的）解决方案？我是否需要在任何情况下使用反应性的东西来实现这一点？

任何帮助都非常感谢！

Answer 1

Dmytro，谢谢你为我指明了正确的方向。我选择了与静态编程语言相关的Mutiny。我的代码现在如下所示：

data class DeviceStatus(var status: Status = Status.OFFLINE) {
    enum class Status {OFFLINE, CONNECTED, ANALYZING, MAINTENANCE}
}

@ApplicationScoped
class DeviceStatusService {
    var deviceStatusProcessor: PublishProcessor<DeviceStatus> = PublishProcessor.create()
    var deviceStatusQueue: Flowable<DeviceStatus> = Flowable.fromPublisher(deviceStatusProcessor)

    fun pushDeviceStatus(deviceStatus: DeviceStatus) {
        deviceStatusProcessor.onNext(deviceStatus)
    }

    fun getStream(): Multi<DeviceStatus> {
        return Multi.createFrom().publisher(deviceStatusQueue)
    }
}

@Path("/deviceStatus")
class DeviceStatusResource {
    private val LOGGER: Logger = Logger.getLogger("DeviceStatusResource")

    @Inject
    @field: Default
    lateinit var deviceStatusService: DeviceStatusService

    @POST
    @Consumes(MediaType.APPLICATION_JSON)
    fun status(status: DeviceStatus): Response {
        LOGGER.info("POST /deviceStatus " + status.status);
        deviceStatusService.pushDeviceStatus(status)
        return Response.ok().build();
    }

    @GET
    @Path("/eventStream")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @SseElementType(MediaType.APPLICATION_JSON)
    fun stream(): Multi<DeviceStatus>? {
        return deviceStatusService.getStream()
    }
}

作为最小设置，该服务可以直接使用deviceStatusProcess作为发布者。但是，Flowable增加了缓冲。欢迎对实现发表评论。

Answer 2

最简单的方法是使用rxjava作为流提供者。首先，您需要添加rxjava依赖项。它可以来自quarkus中的反应依赖项，如kafka，也可以直接使用它（如果您不需要任何流媒体库）：

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

下面是如何每秒发送随机双精度值的示例：

    @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、 每次查询此endpoint时，上面的代码都将生成新的可流动的，我这样做是为了节省空间，在您的情况下，我假设您将有一个事件源，您可以构建一次，并在每次查询endpoint时使用相同的实例