SE Reactive Messaging

[danielkec]

API

API is providing same functionality as MP Reactive Messaging does, number of method signatures is reduced to accommodate slightly different requirements of SE Helidon. For connectors is used same api as for MP connectors, which makes connectors reusable for both flawors.

Acknowledgement

Due to no magic approach implicit acknowledgement has been reduced only to methods providing unwrapped payload, as it wouldn't be possible to do manual ack without org.eclipse.microprofile.reactive.messaging.Message wrapper.

Channel<String> channel1 = Channel.create("channel1");

Messaging.builder()
                .publisher(channel1, Multi.just("foo", "bar").map(Message::of))
                // Every message is acked before each invocation of consumer
                .listener(channel1, s -> System.out.println(s))
                .build()
                .start();

Such messages are acked in preprocesses, right before unwrapping.

Kafka connector usage example:

 Channel<ConsumerRecord<String, String>> fromKafka = Channel.<ConsumerRecord<String, String>>builder()
                .name("from-kafka")
                .publisherConfig(KafkaConnector.configBuilder()
                        .bootstrapServers(KAFKA_SERVER)
                        .groupId("test-group")
                        .topic(TEST_SE_TOPIC_1, TEST_SE_TOPIC_2)
                        .autoOffsetReset(KafkaConfigBuilder.AutoOffsetReset.EARLIEST)
                        .enableAutoCommit(false)
                        .keyDeserializer(StringDeserializer.class)
                        .valueDeserializer(StringDeserializer.class)
                        .build()
                )
                .build();
        KafkaConnector kafkaConnector = KafkaConnector.create(Config.empty());
        Messaging messaging = Messaging.builder()
                .connector(kafkaConnector)
                .listener(fromKafka, consumerRecord -> {
                    LOGGER.info("Kafka says: " + consumerRecord.value());
                })
                .build();
messaging.start();

Emitter usage example:

Channel<String> simpleChannel = Channel.create("simple-channel");

Emitter<String> emitter = Emitter.create(simpleChannel);

Messaging messaging = Messaging.builder()
        .emitter(emitter)
        .listener(simpleChannel, System.out::println)
        .build();

messaging.start();

emitter.send(Message.of("test1"));
emitter.send("test2");
                                                     
messaging.stop();

[paulparkinson]

Looks good/consistent. Let's work on some extended TestMessage types/use cases to prove it out. Will contact you...

[danielkec]

/trigger

[danielkec]

Rebased and ready for review

[tomas-langer]

LGTM