Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Channel<String> fromKafka = Channel.create("from-kafka");
KafkaConnector kafkaConnector = KafkaConnector.create();
Messaging messaging = Messaging.builder()
.connector(kafkaConnector)
.subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
//Apply back-pressure, flatMapCompletionStage requests one by one
.flatMapCompletionStage(message -> {
return CompletableFuture.runAsync(() -> {
//Do something lengthy
Multi.timer(300, TimeUnit.MILLISECONDS, executor)
.first()
.await();
//Acknowledge message has been consumed
message.ack();
});
})
.ignore())
.build()
.start();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment