Quarkus从Kafka主题中提取并将JSON有效负载发送到RESTendpoint

Question

因此，我使用Quarkus和Microprofile Reactive Messaging framework（带有SmallRye Kafka连接器）以及RxJava2 Flowable streams对象来进行Reactive消息接收/发送。我有一个微服务，它使用@incoming和@outgoing注释来正确地使用通道来拉回主题并将消息推送到主题。

但是，现在我想修改它，这样我仍然可以从Kafka主题中提取，然后将JSON负载发送到RESTendpoint。据我所知，没有Quarkus HTTP兼容的SmallRye连接器。有没有人碰巧知道有什么方法可以让它工作？

示例函数

    @Incoming("pre-check")
    @Outgoing("post-check")
    @Broadcast
    public Flowable<CustomMessage> publishToApi(CustomMessage customMessage) {

        LOGGER.info("Message received from topic = {}", customMessage);

        if (customMessage.ready) {
            return Flowable.just(customMessage);
        }
        else {
            return Flowable.empty();
        }
    }

Answer 1

删除@outgoing并使用任何HTTP客户端处理消息以将其发送到某个服务器

或者使输出通道成为客户端的响应