From 2024206a28210fae02d24b3b7685161908ffb602 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 18 Nov 2022 16:04:52 -0600 Subject: [PATCH] Update reactive sample with @ReactivePulsarListener (#218) --- .../ReactiveSpringPulsarBootApp.java | 101 +++++++++++------- 1 file changed, 61 insertions(+), 40 deletions(-) diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java index 1ed503e3..750131ef 100644 --- a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java @@ -16,14 +16,11 @@ package org.springframework.pulsar.example; -import java.time.Duration; -import java.util.Collections; - +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.reactive.client.api.MessageResult; -import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec; -import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer; -import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,11 +33,12 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.PulsarListener; -import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; @SpringBootApplication public class ReactiveSpringPulsarBootApp { @@ -50,43 +48,67 @@ public class ReactiveSpringPulsarBootApp { } /** - * Sends POJO messages with a reactive template and receives them with a reactive - * consumer. + * Sends string messages with a reactive template and receives them with a simple + * reactive listener. */ @Configuration(proxyBeanMethods = false) - static class ReactiveSenderWithReactiveConsumer implements ApplicationListener { + static class ReactiveTemplateWithSimpleReactiveListener implements ApplicationListener { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Autowired + private ReactivePulsarTemplate reactivePulsarTemplate; + + @Override + public void onApplicationEvent(ApplicationReadyEvent event) { + Flux.range(0, 10).map((i) -> "sample-message-" + i) + .as(messages -> this.reactivePulsarTemplate.send("sample-reactive-topic1", messages)).subscribe(); + } + + @ReactivePulsarListener(subscriptionName = "sample-reactive-sub1", topics = "sample-reactive-topic1", + consumerCustomizer = "subscriptionInitialPositionEarliest") + public Mono listenSimple(String msg) { + this.logger.info("Simple reactive listener received: {}", msg); + return Mono.empty(); + } + + } + + /** + * Sends POJO messages with a reactive template and receives them with a reactive + * streaming listener. + */ + @Configuration(proxyBeanMethods = false) + static class ReactiveTemplateWithStreamingReactiveListener implements ApplicationListener { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Autowired private ReactivePulsarTemplate reactivePulsarTemplate; - // TODO remove this once the auto-config is available - @Bean - ReactivePulsarConsumerFactory reactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient) { - MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec(); - spec.setTopicNames(Collections.singletonList("sample-reactive-topic1")); - spec.setSubscriptionName("sample-reactive-sub1"); - spec.setConsumerName("sample-reactive-consumer1"); - return new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, spec); - } - - @Bean - ApplicationRunner listenPojo(ReactivePulsarConsumerFactory reactiveConsumerFactory) { - return args -> { - ReactiveMessageConsumer messageConsumer = reactiveConsumerFactory - .createConsumer(Schema.JSON(Foo.class)); - messageConsumer.consumeMany((messageFlux) -> messageFlux.map(MessageResult::acknowledgeAndReturn)) - .take(Duration.ofSeconds(10)).subscribe((msg) -> this.logger.info("Received: {}", msg)); - }; - } - @Override public void onApplicationEvent(ApplicationReadyEvent event) { this.reactivePulsarTemplate.setSchema(Schema.JSON(Foo.class)); - this.reactivePulsarTemplate - .send("sample-reactive-topic1", Flux.range(0, 10).map((i) -> new Foo("Foo-" + i, "Bar-" + i))) - .subscribe(); + Flux.range(0, 10).map((i) -> new Foo("Foo-" + i, "Bar-" + i)) + .as(messages -> this.reactivePulsarTemplate.send("sample-reactive-topic2", messages)).subscribe(); + } + + @ReactivePulsarListener(subscriptionName = "sample-reactive-sub2", topics = "sample-reactive-topic2", + stream = true, schemaType = SchemaType.JSON, consumerCustomizer = "subscriptionInitialPositionEarliest") + public Flux> listenStreaming(Flux> messages) { + return messages + .doOnNext((msg) -> this.logger.info("Streaming reactive listener received: {}", msg.getValue())) + .map(m -> MessageResult.acknowledge(m.getMessageId())); + } + + } + + @Configuration(proxyBeanMethods = false) + static class ConsumerCustomizerConfig { + + @Bean + ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } } @@ -96,20 +118,19 @@ public class ReactiveSpringPulsarBootApp { * listener. */ @Configuration(proxyBeanMethods = false) - static class ReactiveSenderWithImperativeListener { + static class ReactiveTemplateWithImperativeListener { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Bean ApplicationRunner sendSimple(ReactivePulsarTemplate reactivePulsarTemplate) { - return args -> reactivePulsarTemplate - .send("sample-reactive-topic2", Flux.range(0, 10).map((i) -> "msg-from-sendSimple-" + i)) - .subscribe(); + return args -> Flux.range(0, 10).map((i) -> "msg-from-sendSimple-" + i) + .as(messages -> reactivePulsarTemplate.send("sample-reactive-topic3", messages)).subscribe(); } - @PulsarListener(subscriptionName = "sample-reactive-sub2", topics = "sample-reactive-topic2") + @PulsarListener(subscriptionName = "sample-reactive-sub3", topics = "sample-reactive-topic3") void listenSimple(String message) { - this.logger.info("Received: {}", message); + this.logger.info("Imperative listener received: {}", message); } }