如何持续使用来自Apache Pulsar的消息？

Question

如何使用Akka Streams不断消耗来自Apache Pulsar的消息并打印每条消息？

下面是我从pulsar4s库中找到的示例代码。而不是将消息发布到另一个主题，如何打印消耗的消息？

val consumerFn = () => client.consumer(ConsumerConfig(Seq(intopic), Subscription("mysub")))
val producerFn = () => client.producer(ProducerConfig(outtopic))

val control = source(consumerFn, Some(MessageId.earliest))
                .map { consumerMessage => ProducerMessage(consumerMessage.data) }
                .to(sink(producerFn)).run()

Answer 1

您可以简单地使用Sink.foreach(println))

例如

source(consumerFn, Some(MessageId.earliest)).
.runWith(Sink.foreach(println))