From a5ddf77ced9d9c392efe9de1068fceefa6a706ef Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Thu, 10 Nov 2022 19:52:53 +0100 Subject: [PATCH] Add ReactivePulsarMessageListenerContainer (#198) --- .../ReactiveSpringPulsarBootApp.java | 4 +- spring-pulsar/build.gradle | 1 + ...eactivePulsarMessageListenerContainer.java | 193 ++++++++++ .../ReactivePulsarContainerProperties.java | 128 +++++++ .../ReactivePulsarMessageHandler.java | 26 ++ ...eactivePulsarMessageListenerContainer.java | 50 +++ .../ReactivePulsarOneByOneMessageHandler.java | 40 +++ .../ReactivePulsarStreamingHandler.java | 43 +++ ...ltPulsarMessageListenerContainerTests.java | 6 +- ...vePulsarMessageListenerContainerTests.java | 330 ++++++++++++++++++ 10 files changed, 815 insertions(+), 6 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java 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 c09cb2f0..30cc19cd 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 @@ -76,9 +76,7 @@ public class ReactiveSpringPulsarBootApp { return args -> { ReactiveMessageConsumer messageConsumer = reactiveConsumerFactory .createConsumer(Schema.JSON(Foo.class)); - messageConsumer - .consumeMany((messageFlux) -> messageFlux.map( - (message) -> MessageResult.acknowledge(message.getMessageId(), message.getValue()))) + messageConsumer.consumeMany((messageFlux) -> messageFlux.map(MessageResult::acknowledgeAndReturn)) .take(Duration.ofSeconds(10)).subscribe((msg) -> this.logger.info("Received: {}", msg)); }; } diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index 096c7137..8e69e022 100644 --- a/spring-pulsar/build.gradle +++ b/spring-pulsar/build.gradle @@ -31,6 +31,7 @@ dependencies { testImplementation 'io.micrometer:micrometer-tracing-bridge-brave' testImplementation 'io.micrometer:micrometer-tracing-test' testImplementation 'io.micrometer:micrometer-tracing-integration-test' + testImplementation 'io.projectreactor:reactor-test' testImplementation 'org.assertj:assertj-core' testImplementation 'org.awaitility:awaitility' testImplementation 'org.hamcrest:hamcrest' diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java new file mode 100644 index 00000000..aa6c8516 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java @@ -0,0 +1,193 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer; +import org.apache.pulsar.reactive.client.api.ReactiveMessagePipeline; +import org.apache.pulsar.reactive.client.api.ReactiveMessagePipelineBuilder; +import org.apache.pulsar.reactive.client.internal.api.ApiImplementationFactory; + +import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.util.CollectionUtils; + +/** + * Default implementation for {@link ReactivePulsarMessageListenerContainer}. + * + * @param message type. + * @author Christophe Bornet + */ +public non-sealed class DefaultReactivePulsarMessageListenerContainer + implements ReactivePulsarMessageListenerContainer { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final ReactivePulsarConsumerFactory pulsarConsumerFactory; + + private final ReactivePulsarContainerProperties pulsarContainerProperties; + + private boolean autoStartup = true; + + private final Object lifecycleMonitor = new Object(); + + private final AtomicBoolean running = new AtomicBoolean(false); + + private ReactiveMessageConsumerBuilderCustomizer consumerCustomizer; + + private ReactiveMessagePipeline pipeline; + + public DefaultReactivePulsarMessageListenerContainer(ReactivePulsarConsumerFactory pulsarConsumerFactory, + ReactivePulsarContainerProperties pulsarContainerProperties) { + this.pulsarConsumerFactory = pulsarConsumerFactory; + this.pulsarContainerProperties = pulsarContainerProperties; + } + + public ReactivePulsarConsumerFactory getReactivePulsarConsumerFactory() { + return this.pulsarConsumerFactory; + } + + public ReactivePulsarContainerProperties getContainerProperties() { + return this.pulsarContainerProperties; + } + + @Override + public boolean isRunning() { + return this.running.get(); + } + + protected void setRunning(boolean running) { + this.running.set(running); + } + + @Override + public void setupMessageHandler(ReactivePulsarMessageHandler messageHandler) { + this.pulsarContainerProperties.setMessageHandler(messageHandler); + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @Override + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public ReactiveMessageConsumerBuilderCustomizer getConsumerCustomizer() { + return this.consumerCustomizer; + } + + @Override + public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer consumerCustomizer) { + this.consumerCustomizer = consumerCustomizer; + } + + @Override + public final void start() { + synchronized (this.lifecycleMonitor) { + if (!isRunning()) { + Objects.requireNonNull(this.pulsarContainerProperties.getMessageHandler(), + "A ReactivePulsarMessageHandler must be provided"); + doStart(); + } + } + } + + @Override + public void stop() { + synchronized (this.lifecycleMonitor) { + if (isRunning()) { + doStop(); + } + } + } + + private void doStart() { + setRunning(true); + this.pipeline = startPipeline(this.pulsarContainerProperties); + } + + public void doStop() { + try { + this.logger.info("Closing Pulsar Reactive pipeline."); + this.pipeline.close(); + } + catch (Exception e) { + this.logger.error(e, () -> "Error closing Pulsar Reactive pipeline."); + } + finally { + setRunning(false); + } + } + + @SuppressWarnings({ "unchecked" }) + private ReactiveMessagePipeline startPipeline(ReactivePulsarContainerProperties containerProperties) { + ReactiveMessageConsumerBuilderCustomizer customizer = (builder) -> { + if (containerProperties.getSubscriptionType() != null) { + builder.subscriptionType(containerProperties.getSubscriptionType()); + } + if (containerProperties.getSubscriptionName() != null) { + builder.subscriptionName(containerProperties.getSubscriptionName()); + } + if (!CollectionUtils.isEmpty(containerProperties.getTopics())) { + builder.topicNames(containerProperties.getTopics()); + } + if (containerProperties.getTopicsPattern() != null) { + builder.topicsPattern(containerProperties.getTopicsPattern()); + } + }; + + List> customizers = new ArrayList<>(); + customizers.add(customizer); + if (this.consumerCustomizer != null) { + customizers.add(this.consumerCustomizer); + } + + ReactiveMessageConsumer consumer = getReactivePulsarConsumerFactory() + .createConsumer(containerProperties.getSchema(), customizers); + ReactiveMessagePipelineBuilder pipelineBuilder = ApiImplementationFactory + .createReactiveMessageHandlerPipelineBuilder(consumer); + Object messageHandler = containerProperties.getMessageHandler(); + ReactiveMessagePipeline pipeline; + if (messageHandler instanceof ReactivePulsarStreamingHandler) { + pipeline = pipelineBuilder + .streamingMessageHandler(((ReactivePulsarStreamingHandler) messageHandler)::received).build(); + } + else { + ReactiveMessagePipelineBuilder.OneByOneMessagePipelineBuilder messagePipelineBuilder = pipelineBuilder + .messageHandler(((ReactivePulsarOneByOneMessageHandler) messageHandler)::received) + .handlingTimeout(containerProperties.getHandlingTimeout()); + if (containerProperties.getConcurrency() > 0) { + pipeline = messagePipelineBuilder.concurrent().concurrency(containerProperties.getConcurrency()) + .maxInflight(containerProperties.getMaxInFlight()).build(); + } + else { + pipeline = pipelineBuilder.build(); + } + } + pipeline.start(); + return pipeline; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java new file mode 100644 index 00000000..ae3063cd --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java @@ -0,0 +1,128 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import java.time.Duration; +import java.util.List; +import java.util.regex.Pattern; + +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; + +/** + * Contains runtime properties for a reactive listener container. + * + * @param message type. + * @author Christophe Bornet + */ +public class ReactivePulsarContainerProperties { + + private List topics; + + private Pattern topicsPattern; + + private String subscriptionName; + + private SubscriptionType subscriptionType = SubscriptionType.Exclusive; + + private Schema schema; + + private ReactivePulsarMessageHandler messageHandler; + + private Duration handlingTimeout = Duration.ofMinutes(2); + + private int concurrency = 0; + + private int maxInFlight = 0; + + public ReactivePulsarMessageHandler getMessageHandler() { + return this.messageHandler; + } + + public void setMessageHandler(ReactivePulsarMessageHandler messageHandler) { + this.messageHandler = messageHandler; + } + + public SubscriptionType getSubscriptionType() { + return this.subscriptionType; + } + + public void setSubscriptionType(SubscriptionType subscriptionType) { + this.subscriptionType = subscriptionType; + } + + public Schema getSchema() { + return this.schema; + } + + public void setSchema(Schema schema) { + this.schema = schema; + } + + public List getTopics() { + return this.topics; + } + + public void setTopics(List topics) { + this.topics = topics; + } + + public Pattern getTopicsPattern() { + return this.topicsPattern; + } + + public void setTopicsPattern(Pattern topicsPattern) { + this.topicsPattern = topicsPattern; + } + + public void setTopicsPattern(String topicsPattern) { + this.topicsPattern = Pattern.compile(topicsPattern); + } + + public String getSubscriptionName() { + return this.subscriptionName; + } + + public void setSubscriptionName(String subscriptionName) { + this.subscriptionName = subscriptionName; + } + + public Duration getHandlingTimeout() { + return this.handlingTimeout; + } + + public void setHandlingTimeout(Duration handlingTimeout) { + this.handlingTimeout = handlingTimeout; + } + + public int getConcurrency() { + return this.concurrency; + } + + public void setConcurrency(int concurrency) { + this.concurrency = concurrency; + } + + public int getMaxInFlight() { + return this.maxInFlight; + } + + public void setMaxInFlight(int maxInFlight) { + this.maxInFlight = maxInFlight; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java new file mode 100644 index 00000000..94c46db7 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java @@ -0,0 +1,26 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +/** + * Reactive message handler used by {@link DefaultReactivePulsarMessageListenerContainer}. + * + * @author Christophe Bornet + */ +public sealed interface ReactivePulsarMessageHandler permits ReactivePulsarOneByOneMessageHandler, ReactivePulsarStreamingHandler { + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java new file mode 100644 index 00000000..50452f59 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java @@ -0,0 +1,50 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; + +/** + * Internal abstraction used by the framework representing a reactive message listener + * container. Not meant to be implemented externally. + * + * @param message type. + * @author Christophe Bornet + */ +public sealed interface ReactivePulsarMessageListenerContainer + extends SmartLifecycle, DisposableBean permits DefaultReactivePulsarMessageListenerContainer { + + void setupMessageHandler(ReactivePulsarMessageHandler messageListener); + + @Override + default void destroy() { + stop(); + } + + default void setAutoStartup(boolean autoStartup) { + // empty + } + + default ReactivePulsarContainerProperties getContainerProperties() { + throw new UnsupportedOperationException("This container doesn't support retrieving its properties"); + } + + void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer consumerCustomizer); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java new file mode 100644 index 00000000..92f50f72 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java @@ -0,0 +1,40 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.reactive.client.api.ReactiveMessagePipelineBuilder; +import org.reactivestreams.Publisher; + +/** + * Message handler class with a {@link #received} method for use in + * {@link ReactiveMessagePipelineBuilder#messageHandler}. + * + * @param message payload type + * @author Christophe Bornet + */ +public non-sealed interface ReactivePulsarOneByOneMessageHandler extends ReactivePulsarMessageHandler { + + /** + * Callback passed to {@link ReactiveMessagePipelineBuilder#messageHandler} that will + * be called for each received message. + * @param message the message received + * @return a completed {@link Publisher} when the callback is done. + */ + Publisher received(Message message); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java new file mode 100644 index 00000000..b070cecb --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java @@ -0,0 +1,43 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.ReactiveMessagePipelineBuilder; +import org.reactivestreams.Publisher; + +import reactor.core.publisher.Flux; + +/** + * Message handler class with a {@link #received} method for use in + * {@link ReactiveMessagePipelineBuilder#streamingMessageHandler}. + * + * @param message payload type + * @author Christophe Bornet + */ +public non-sealed interface ReactivePulsarStreamingHandler extends ReactivePulsarMessageHandler { + + /** + * Callback passed to {@link ReactiveMessagePipelineBuilder#streamingMessageHandler} + * that will be applied to the flux of received message. + * @param messages the messages received + * @return a completed {@link Publisher} when the callback is done. + */ + Publisher> received(Flux> messages); + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java index 5ed5eed2..0455e2d4 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java @@ -215,8 +215,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS config.put("topicNames", Collections.singleton("dpmlct-016")); config.put("subscriptionName", "dpmlct-sb-016"); config.put("ackTimeoutMillis", 1); - DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1).deadLetterTopic("dlq-topic") - .build(); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) + .deadLetterTopic("dpmlct-016-dlq-topic").build(); config.put("deadLetterPolicy", deadLetterPolicy); final PulsarClient pulsarClient = PulsarClient.builder() @@ -232,7 +232,7 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS .setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> dlqLatch.countDown()); dlqContainerProperties.setSchema(Schema.INT32); dlqContainerProperties.setSubscriptionType(SubscriptionType.Shared); - dlqContainerProperties.setTopics(new String[] { "dlq-topic" }); + dlqContainerProperties.setTopics(new String[] { "dpmlct-016-dlq-topic" }); DefaultPulsarMessageListenerContainer dlqContainer = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, dlqContainerProperties); dlqContainer.start(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java new file mode 100644 index 00000000..532e8d9e --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java @@ -0,0 +1,330 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener.reactive; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.pulsar.client.api.DeadLetterPolicy; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec; +import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec; +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer; +import org.apache.pulsar.reactive.client.api.ReactiveMessagePipeline; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; + +import org.springframework.pulsar.core.PulsarTestContainerSupport; +import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; +import org.springframework.pulsar.core.reactive.ReactivePulsarSenderTemplate; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * Tests for {@link DefaultReactivePulsarMessageListenerContainer} + * + * @author Christophe Bornet + */ +class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestContainerSupport { + + @Test + void messageHandlerListener() throws Exception { + String topic = "drpmlct-012"; + MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec(); + config.setTopicNames(Collections.singletonList(topic)); + config.setSubscriptionName("drpmlct-sb-012"); + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, config); + // Ensure subscription is created + pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10)); + CountDownLatch latch = new CountDownLatch(1); + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties.setMessageHandler( + (ReactivePulsarOneByOneMessageHandler) (msg) -> Mono.fromRunnable(latch::countDown)); + pulsarContainerProperties.setSchema(Schema.STRING); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec(); + prodConfig.setTopicName(topic); + DefaultReactivePulsarSenderFactory pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>( + reactivePulsarClient, prodConfig, null); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe").subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + container.stop(); + pulsarClient.close(); + } + + @Test + void streamingHandlerListener() throws Exception { + String topic = "drpmlct-013"; + MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec(); + config.setTopicNames(Collections.singletonList(topic)); + config.setSubscriptionName("drpmlct-sb-013"); + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, config); + // Ensure subscription is created + pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10)); + CountDownLatch latch = new CountDownLatch(5); + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties.setMessageHandler((ReactivePulsarStreamingHandler) (msg) -> msg.map(m -> { + latch.countDown(); + return MessageResult.acknowledge(m.getMessageId()); + })); + pulsarContainerProperties.setSchema(Schema.STRING); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec(); + prodConfig.setTopicName(topic); + DefaultReactivePulsarSenderFactory pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>( + reactivePulsarClient, prodConfig, null); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(pulsarProducerFactory); + Flux.range(0, 5).map(i -> "hello john doe" + i).as(pulsarTemplate::send).subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + container.stop(); + pulsarClient.close(); + } + + @Test + void containerProperties() throws Exception { + String topic = "drpmlct-sb-014"; + String subscriptionName = "drpmlct-sb-014"; + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, null); + // Ensure subscription is created + pulsarConsumerFactory + .createConsumer(Schema.STRING, + Collections.singletonList( + c -> c.topicNames(Collections.singletonList(topic)).subscriptionName(subscriptionName))) + .consumeNothing().block(Duration.ofSeconds(10)); + CountDownLatch latch = new CountDownLatch(1); + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties.setMessageHandler( + (ReactivePulsarOneByOneMessageHandler) (msg) -> Mono.fromRunnable(latch::countDown)); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setTopics(List.of(topic)); + pulsarContainerProperties.setSubscriptionName(subscriptionName); + pulsarContainerProperties.setConcurrency(5); + pulsarContainerProperties.setMaxInFlight(6); + pulsarContainerProperties.setHandlingTimeout(Duration.ofMillis(7)); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec(); + prodConfig.setTopicName(topic); + DefaultReactivePulsarSenderFactory pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>( + reactivePulsarClient, prodConfig, null); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe").subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(container).extracting("pipeline", InstanceOfAssertFactories.type(ReactiveMessagePipeline.class)) + .hasFieldOrPropertyWithValue("concurrency", 5).hasFieldOrPropertyWithValue("maxInflight", 6) + .hasFieldOrPropertyWithValue("handlingTimeout", Duration.ofMillis(7)); + + container.stop(); + pulsarClient.close(); + } + + @Test + void defaultSubscriptionType() throws Exception { + String topic = "drpmlct-015"; + MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec(); + config.setTopicNames(Collections.singletonList(topic)); + config.setSubscriptionName("drpmlct-sb-015"); + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, config); + + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties + .setMessageHandler((ReactivePulsarOneByOneMessageHandler) (msg) -> Mono.empty()); + pulsarContainerProperties.setSchema(Schema.STRING); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + + Thread.sleep(2_000); + + StepVerifier + .create(pulsarConsumerFactory + .createConsumer(Schema.STRING, + Collections.singletonList(c -> c.subscriptionType(SubscriptionType.Shared))) + .consumeNothing()) + .expectError().verify(Duration.ofSeconds(10)); + + container.stop(); + pulsarClient.close(); + } + + @Test + void containerSubscriptionType() throws Exception { + String topic = "drpmlct-016"; + MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec(); + config.setTopicNames(Collections.singletonList(topic)); + config.setSubscriptionName("drpmlct-sb-016"); + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, config); + + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties + .setMessageHandler((ReactivePulsarOneByOneMessageHandler) (msg) -> Mono.empty()); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + + Thread.sleep(2_000); + + StepVerifier + .create(pulsarConsumerFactory + .createConsumer(Schema.STRING, + Collections.singletonList(c -> c.subscriptionType(SubscriptionType.Shared))) + .consumeNothing()) + .expectComplete().verify(Duration.ofSeconds(10)); + + container.stop(); + pulsarClient.close(); + } + + @Test + void containerTopicsPattern() throws Exception { + String topic = "drpmlct-017-foo"; + String subscriptionName = "drpmlct-sb-017"; + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, null); + // Ensure subscription is created + pulsarConsumerFactory + .createConsumer(Schema.STRING, + Collections.singletonList( + c -> c.topicNames(Collections.singletonList(topic)).subscriptionName(subscriptionName))) + .consumeNothing().block(Duration.ofSeconds(10)); + CountDownLatch latch = new CountDownLatch(1); + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties.setMessageHandler( + (ReactivePulsarOneByOneMessageHandler) (msg) -> Mono.fromRunnable(latch::countDown)); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setTopicsPattern("persistent://public/default/drpmlct-017-.*"); + pulsarContainerProperties.setSubscriptionName(subscriptionName); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.start(); + MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec(); + prodConfig.setTopicName(topic); + DefaultReactivePulsarSenderFactory pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>( + reactivePulsarClient, prodConfig, null); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe").subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + + container.stop(); + pulsarClient.close(); + } + + @Test + void consumerCustomizer() throws Exception { + String topic = "drpmlct-018"; + String deadLetterTopic = "drpmlct-018-dlq-topic"; + MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec(); + config.setTopicNames(Collections.singletonList(topic)); + config.setSubscriptionName("drpmlct-sb-018"); + config.setNegativeAckRedeliveryDelay(Duration.ZERO); + PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build(); + ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient); + DefaultReactivePulsarConsumerFactory pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>( + reactivePulsarClient, config); + ReactiveMessageConsumer dlqConsumer = pulsarConsumerFactory.createConsumer(Schema.STRING, + Collections.singletonList(b -> b.topicNames(Collections.singletonList(deadLetterTopic)))); + + // Ensure subscriptions are created + pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10)); + dlqConsumer.consumeNothing().block(Duration.ofSeconds(10)); + + CountDownLatch latch = new CountDownLatch(6); + ReactivePulsarContainerProperties pulsarContainerProperties = new ReactivePulsarContainerProperties<>(); + pulsarContainerProperties.setMessageHandler((ReactivePulsarStreamingHandler) (msg) -> msg.map(m -> { + latch.countDown(); + if (m.getValue().endsWith("4")) { + return MessageResult.negativeAcknowledge(m.getMessageId()); + } + return MessageResult.acknowledge(m.getMessageId()); + })); + pulsarContainerProperties.setSchema(Schema.STRING); + pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared); + DefaultReactivePulsarMessageListenerContainer container = new DefaultReactivePulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) + .deadLetterTopic(deadLetterTopic).build(); + container.setConsumerCustomizer(b -> b.deadLetterPolicy(deadLetterPolicy)); + container.start(); + MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec(); + prodConfig.setTopicName(topic); + DefaultReactivePulsarSenderFactory pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>( + reactivePulsarClient, prodConfig, null); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(pulsarProducerFactory); + Flux.range(0, 5).map(i -> "hello john doe" + i).as(pulsarTemplate::send).subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + + CountDownLatch dlqLatch = new CountDownLatch(1); + dlqConsumer.consumeOne(message -> { + if (message.getValue().endsWith("4")) { + dlqLatch.countDown(); + } + return Mono.just(MessageResult.acknowledge(message.getMessageId())); + }).block(); + + assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + container.stop(); + pulsarClient.close(); + } + +}