Итак, я использую Quarkus вместе с фреймворком Microprofile Reactive Messaging (с коннекторами SmallRye Kafka) и объектом потоков RxJava2 Flowable для получения / отправки реактивных сообщений. У меня есть микросервис, который использует аннотации @Incoming и @Outgoing для правильного использования каналов для извлечения из задних тем и сообщений pu sh в темы.
Однако теперь я хочу изменить это, чтобы я мог тянуть из Kafka topi c и теперь отправьте полезную нагрузку JSON в конечную точку REST. Насколько мне известно, нет соединителя SmallRye, совместимого с Quarkus HTTP. Кто-нибудь знает какие-либо методы, чтобы заставить это работать?
Пример функции
@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();
}
}