Add ReactivePulsarMessageListenerContainer (#198)

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<Foo> 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));
 			};
 		}

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'

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 <T> message type.
+ * @author Christophe Bornet
+ */
+public non-sealed class DefaultReactivePulsarMessageListenerContainer<T>
+		implements ReactivePulsarMessageListenerContainer<T> {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final ReactivePulsarConsumerFactory<T> pulsarConsumerFactory;
+
+	private final ReactivePulsarContainerProperties<T> pulsarContainerProperties;
+
+	private boolean autoStartup = true;
+
+	private final Object lifecycleMonitor = new Object();
+
+	private final AtomicBoolean running = new AtomicBoolean(false);
+
+	private ReactiveMessageConsumerBuilderCustomizer<T> consumerCustomizer;
+
+	private ReactiveMessagePipeline pipeline;
+
+	public DefaultReactivePulsarMessageListenerContainer(ReactivePulsarConsumerFactory<T> pulsarConsumerFactory,
+			ReactivePulsarContainerProperties<T> pulsarContainerProperties) {
+		this.pulsarConsumerFactory = pulsarConsumerFactory;
+		this.pulsarContainerProperties = pulsarContainerProperties;
+	}
+
+	public ReactivePulsarConsumerFactory<T> getReactivePulsarConsumerFactory() {
+		return this.pulsarConsumerFactory;
+	}
+
+	public ReactivePulsarContainerProperties<T> 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<T> getConsumerCustomizer() {
+		return this.consumerCustomizer;
+	}
+
+	@Override
+	public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer<T> 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<T> containerProperties) {
+		ReactiveMessageConsumerBuilderCustomizer<T> 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<ReactiveMessageConsumerBuilderCustomizer<T>> customizers = new ArrayList<>();
+		customizers.add(customizer);
+		if (this.consumerCustomizer != null) {
+			customizers.add(this.consumerCustomizer);
+		}
+
+		ReactiveMessageConsumer<T> consumer = getReactivePulsarConsumerFactory()
+				.createConsumer(containerProperties.getSchema(), customizers);
+		ReactiveMessagePipelineBuilder<T> pipelineBuilder = ApiImplementationFactory
+				.createReactiveMessageHandlerPipelineBuilder(consumer);
+		Object messageHandler = containerProperties.getMessageHandler();
+		ReactiveMessagePipeline pipeline;
+		if (messageHandler instanceof ReactivePulsarStreamingHandler<?>) {
+			pipeline = pipelineBuilder
+					.streamingMessageHandler(((ReactivePulsarStreamingHandler<T>) messageHandler)::received).build();
+		}
+		else {
+			ReactiveMessagePipelineBuilder.OneByOneMessagePipelineBuilder<T> messagePipelineBuilder = pipelineBuilder
+					.messageHandler(((ReactivePulsarOneByOneMessageHandler<T>) 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;
+	}
+
+}

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 <T> message type.
+ * @author Christophe Bornet
+ */
+public class ReactivePulsarContainerProperties<T> {
+
+	private List<String> topics;
+
+	private Pattern topicsPattern;
+
+	private String subscriptionName;
+
+	private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
+
+	private Schema<T> 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<T> getSchema() {
+		return this.schema;
+	}
+
+	public void setSchema(Schema<T> schema) {
+		this.schema = schema;
+	}
+
+	public List<String> getTopics() {
+		return this.topics;
+	}
+
+	public void setTopics(List<String> 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;
+	}
+
+}

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 {
+
+}

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 <T> message type.
+ * @author Christophe Bornet
+ */
+public sealed interface ReactivePulsarMessageListenerContainer<T>
+		extends SmartLifecycle, DisposableBean permits DefaultReactivePulsarMessageListenerContainer {
+
+	void setupMessageHandler(ReactivePulsarMessageHandler messageListener);
+
+	@Override
+	default void destroy() {
+		stop();
+	}
+
+	default void setAutoStartup(boolean autoStartup) {
+		// empty
+	}
+
+	default ReactivePulsarContainerProperties<T> getContainerProperties() {
+		throw new UnsupportedOperationException("This container doesn't support retrieving its properties");
+	}
+
+	void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer<T> consumerCustomizer);
+
+}

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 <T> message payload type
+ * @author Christophe Bornet
+ */
+public non-sealed interface ReactivePulsarOneByOneMessageHandler<T> 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<Void> received(Message<T> message);
+
+}

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 <T> message payload type
+ * @author Christophe Bornet
+ */
+public non-sealed interface ReactivePulsarStreamingHandler<T> 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<MessageResult<Void>> received(Flux<Message<T>> messages);
+
+}

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<Integer> dlqContainer = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, dlqContainerProperties);
 		dlqContainer.start();

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<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				reactivePulsarClient, config);
+		// Ensure subscription is created
+		pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10));
+		CountDownLatch latch = new CountDownLatch(1);
+		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties.setMessageHandler(
+				(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
+		prodConfig.setTopicName(topic);
+		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
+				reactivePulsarClient, prodConfig, null);
+		ReactivePulsarSenderTemplate<String> 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<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				reactivePulsarClient, config);
+		// Ensure subscription is created
+		pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10));
+		CountDownLatch latch = new CountDownLatch(5);
+		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties.setMessageHandler((ReactivePulsarStreamingHandler<String>) (msg) -> msg.map(m -> {
+			latch.countDown();
+			return MessageResult.acknowledge(m.getMessageId());
+		}));
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
+		prodConfig.setTopicName(topic);
+		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
+				reactivePulsarClient, prodConfig, null);
+		ReactivePulsarSenderTemplate<String> 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<String> 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<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties.setMessageHandler(
+				(ReactivePulsarOneByOneMessageHandler<String>) (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<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
+		prodConfig.setTopicName(topic);
+		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
+				reactivePulsarClient, prodConfig, null);
+		ReactivePulsarSenderTemplate<String> 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<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				reactivePulsarClient, config);
+
+		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties
+				.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		DefaultReactivePulsarMessageListenerContainer<String> 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<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				reactivePulsarClient, config);
+
+		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties
+				.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
+		DefaultReactivePulsarMessageListenerContainer<String> 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<String> 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<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties.setMessageHandler(
+				(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		pulsarContainerProperties.setTopicsPattern("persistent://public/default/drpmlct-017-.*");
+		pulsarContainerProperties.setSubscriptionName(subscriptionName);
+		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		container.start();
+		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
+		prodConfig.setTopicName(topic);
+		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
+				reactivePulsarClient, prodConfig, null);
+		ReactivePulsarSenderTemplate<String> 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<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				reactivePulsarClient, config);
+		ReactiveMessageConsumer<String> 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<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
+		pulsarContainerProperties.setMessageHandler((ReactivePulsarStreamingHandler<String>) (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<String> 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<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
+				reactivePulsarClient, prodConfig, null);
+		ReactivePulsarSenderTemplate<String> 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();
+	}
+
+}