如何在发送reactor消耗的前100条消息后有条件地发布附加SSE web流量事件kafka主题

Question

我正在使用reactor-kafka库和spring webflux SSE流消费来自kafka主题的数据。当来自主题的所有消息都被消费完时，我需要返回一个特殊的ServerSentEvent，即最大主题偏移量get等于订阅时从0偏移量开始消耗的当前偏移量。以便客户端知道在kafka topic中没有更多的消息。

有没有可能使用web flux来实现这样的目标？也就是说，如果我说在每100个元素从任何有限的元素列表中消费并作为ServerSentEvent发送到SSE流之后，这个SSE流应该再得到一个作为SeverSentEvent的事件，并带有注释“消费”。

Answer 1

我可能错了，但是，这对我来说没有任何意义。

Kafka是事件流(Always on)。

我的意思是，在卡夫卡中没有last message的概念。

您可以在主题消费者端编写一些代码，因此，当在某个时间没有事件到达时，触发操作。

但是，这并不准确，而且确实很有意义。

Answer 2

我不知道这是不是你想要的。

    @GetMapping(path = "/sse/endpoint", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Event> sse() {
        return new KafkaConsumer().receive();
    }

    public class KafkaConsumer {

        private final Sinks.Many<Event> sseEventSender = Sinks.many().multicast()
                .onBackpressureBuffer();

        public Flux<Event> receive() {
            return sseEventSender.asFlux();
        }

        public void consume() {
            if(...) {
                sseEventSender.tryEmitNext(event);
            }
        }
    }