Add ReactivePulsarSenderTemplate (#164)

build.gradle

@@ -76,3 +76,4 @@ subprojects { subproject ->
 		}
 	}
 }
+

buildSrc/src/main/groovy/io/spring/gradle/convention/RepositoryConventionPlugin.groovy

@@ -47,6 +47,10 @@ class RepositoryConventionPlugin implements Plugin<Project> {
 					}
 					url = 'https://repo.spring.io/snapshot/'
 				}
+				maven {
+					name = 'apache-snapshot'
+					url = 'https://repository.apache.org/content/repositories/snapshots'
+				}
 			}
 			if (isSnapshot || isMilestone) {
 				maven {

spring-pulsar-dependencies/build.gradle

@@ -13,6 +13,7 @@ ext {
 	protobufJavaVersion = '3.21.5'
 	testcontainersVersion = '1.17.3'
 	pulsarVersion = '2.10.1'
+	pulsarClientReactiveVersion = '0.1.0-SNAPSHOT'
 	springBootVersion = '3.0.0-SNAPSHOT'
 }
 
@@ -26,5 +27,6 @@ dependencies {
 		api "com.google.code.findbugs:jsr305:$googleJsr305Version"
 		api "com.google.protobuf:protobuf-java:$protobufJavaVersion"
 		api "org.apache.pulsar:pulsar-client-all:$pulsarVersion"
+		api "org.apache.pulsar:pulsar-client-reactive-adapter:$pulsarClientReactiveVersion"
 	}
 }

spring-pulsar/build.gradle

@@ -9,6 +9,7 @@ dependencies {
 	api 'com.google.protobuf:protobuf-java'
 	api 'io.micrometer:micrometer-observation'
 	api 'org.apache.pulsar:pulsar-client-all'
+	api "org.apache.pulsar:pulsar-client-reactive-adapter"
 	api 'org.springframework:spring-context'
 	api 'org.springframework:spring-messaging'
 	api 'org.springframework:spring-tx'

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java

@@ -0,0 +1,98 @@
+/*
+ * 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.core.reactive;
+
+import java.util.List;
+
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Default implementation of {@link ReactivePulsarSenderFactory}.
+ *
+ * @param <T> reactive sender type.
+ * @author Christophe Bornet
+ */
+public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSenderFactory<T> {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final ReactivePulsarClient reactivePulsarClient;
+
+	private final ReactiveMessageSenderSpec reactiveMessageSenderSpec;
+
+	public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient,
+			ReactiveMessageSenderSpec reactiveMessageSenderSpec) {
+		this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec);
+	}
+
+	public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
+			ReactiveMessageSenderSpec reactiveMessageSenderSpec) {
+		this.reactivePulsarClient = reactivePulsarClient;
+		this.reactiveMessageSenderSpec = reactiveMessageSenderSpec;
+	}
+
+	@Override
+	public ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema) {
+		return doCreateReactiveMessageSender(topic, schema, null, null);
+	}
+
+	@Override
+	public ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema,
+			MessageRouter messageRouter) {
+		return doCreateReactiveMessageSender(topic, schema, messageRouter, null);
+	}
+
+	@Override
+	public ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema,
+			MessageRouter messageRouter, List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
+		return doCreateReactiveMessageSender(topic, schema, messageRouter, customizers);
+	}
+
+	private ReactiveMessageSender<T> doCreateReactiveMessageSender(String topic, Schema<T> schema,
+			MessageRouter messageRouter, List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
+		final String resolvedTopic = ReactiveMessageSenderUtils.resolveTopicName(topic, this);
+		this.logger.trace(() -> String.format("Creating reactive message sender for '%s' topic", resolvedTopic));
+		final ReactiveMessageSenderBuilder<T> sender = this.reactivePulsarClient.messageSender(schema);
+		if (this.reactiveMessageSenderSpec != null) {
+			sender.applySpec(this.reactiveMessageSenderSpec);
+		}
+		sender.topic(resolvedTopic);
+		if (messageRouter != null) {
+			sender.messageRouter(messageRouter);
+		}
+		if (!CollectionUtils.isEmpty(customizers)) {
+			customizers.forEach((c) -> c.customize(sender));
+		}
+		return sender.build();
+	}
+
+	@Override
+	public ReactiveMessageSenderSpec getReactiveMessageSenderSpec() {
+		return this.reactiveMessageSenderSpec;
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java

@@ -0,0 +1,36 @@
+/*
+ * 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.core.reactive;
+
+import org.apache.pulsar.reactive.client.api.MessageSpecBuilder;
+
+/**
+ * The interface to customize a {@link MessageSpecBuilder}.
+ *
+ * @param <T> The message payload type
+ * @author Christophe Bornet
+ */
+@FunctionalInterface
+public interface MessageSpecBuilderCustomizer<T> {
+
+	/**
+	 * Customizes a {@link MessageSpecBuilder}.
+	 * @param messageSpecBuilder the MessageSpecBuilder to customize
+	 */
+	void customize(MessageSpecBuilder<T> messageSpecBuilder);
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java

@@ -0,0 +1,36 @@
+/*
+ * 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.core.reactive;
+
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
+
+/**
+ * The interface to customize a {@link ReactiveMessageSenderBuilder}.
+ *
+ * @param <T> The message payload type
+ * @author Christophe Bornet
+ */
+@FunctionalInterface
+public interface ReactiveMessageSenderBuilderCustomizer<T> {
+
+	/**
+	 * Customizes a {@link ReactiveMessageSenderBuilder}.
+	 * @param reactiveMessageSenderBuilder the builder to customize
+	 */
+	void customize(ReactiveMessageSenderBuilder<T> reactiveMessageSenderBuilder);
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java

@@ -0,0 +1,46 @@
+/*
+ * 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.core.reactive;
+
+import java.util.Optional;
+
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+
+import org.springframework.util.StringUtils;
+
+/**
+ * Common utilities used by reactive sender components.
+ *
+ * @author Christophe Bornet
+ */
+final class ReactiveMessageSenderUtils {
+
+	private ReactiveMessageSenderUtils() {
+	}
+
+	static <T> String resolveTopicName(String userSpecifiedTopic,
+			ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory) {
+		ReactiveMessageSenderSpec reactiveMessageSenderSpec = reactiveMessageSenderFactory
+				.getReactiveMessageSenderSpec();
+		if (StringUtils.hasText(userSpecifiedTopic)) {
+			return userSpecifiedTopic;
+		}
+		return Optional.ofNullable(reactiveMessageSenderSpec).map(ReactiveMessageSenderSpec::getTopicName).orElseThrow(
+				() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java

@@ -0,0 +1,72 @@
+/*
+ * 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.core.reactive;
+
+import java.util.List;
+
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+
+/**
+ * The strategy to create a {@link ReactiveMessageSender} instance(s).
+ *
+ * @param <T> reactive message sender payload type
+ * @author Christophe Bornet
+ */
+public interface ReactivePulsarSenderFactory<T> {
+
+	/**
+	 * Create a reactive message sender.
+	 * @param topic the topic the reactive message sender will send messages to or
+	 * {@code null} to use the default topic
+	 * @param schema the schema of the messages to be sent
+	 * @return the reactive message sender
+	 */
+	ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema);
+
+	/**
+	 * Create a reactive message sender.
+	 * @param topic the topic the reactive message sender will send messages to or
+	 * {@code null} to use the default topic
+	 * @param schema the schema of the messages to be sent
+	 * @param messageRouter the optional message router to use
+	 * @return the reactive message sender
+	 */
+	ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema, MessageRouter messageRouter);
+
+	/**
+	 * Create a reactive message sender.
+	 * @param topic the topic the reactive message sender will send messages to or
+	 * {@code null} to use the default topic
+	 * @param schema the schema of the messages to be sent
+	 * @param messageRouter the optional message router to use
+	 * @param customizers the optional list of customizers to apply to the reactive
+	 * message sender builder
+	 * @return the reactive message sender
+	 */
+	ReactiveMessageSender<T> createReactiveMessageSender(String topic, Schema<T> schema, MessageRouter messageRouter,
+			List<ReactiveMessageSenderBuilderCustomizer<T>> customizers);
+
+	/**
+	 * Return the ReactiveMessageSenderSpec to use when creating reactive senders.
+	 * @return the ReactiveMessageSenderSpec
+	 */
+	ReactiveMessageSenderSpec getReactiveMessageSenderSpec();
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderOperations.java

@@ -0,0 +1,102 @@
+/*
+ * 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.core.reactive;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * The Pulsar reactive send operations contract.
+ *
+ * @param <T> the message payload type
+ * @author Christophe Bornet
+ */
+public interface ReactivePulsarSenderOperations<T> {
+
+	/**
+	 * Sends a message to the default topic in a reactive manner.
+	 * @param message the message to send
+	 * @return the id assigned by the broker to the published message
+	 */
+	Mono<MessageId> send(T message);
+
+	/**
+	 * Sends a message to the specified topic in a reactive manner.
+	 * @param topic the topic to send the message to or {@code null} to send to the
+	 * default topic
+	 * @param message the message to send
+	 * @return the id assigned by the broker to the published message
+	 */
+	Mono<MessageId> send(String topic, T message);
+
+	/**
+	 * Create a {@link SendMessageBuilder builder} for configuring and sending a message
+	 * reactively.
+	 * @param message the payload of the message
+	 * @return the builder to configure and send the message
+	 */
+	SendMessageBuilder<T> newMessage(T message);
+
+	/**
+	 * Builder that can be used to configure and send a message. Provides more options
+	 * than the send methods provided by {@link ReactivePulsarSenderOperations}.
+	 *
+	 * @param <T> the message payload type
+	 */
+	interface SendMessageBuilder<T> {
+
+		/**
+		 * Specify the topic to send the message to.
+		 * @param topic the destination topic
+		 * @return the current builder with the destination topic specified
+		 */
+		SendMessageBuilder<T> withTopic(String topic);
+
+		/**
+		 * Specifies the message customizer to use to further configure the message.
+		 * @param customizer the message customizer
+		 * @return the current builder with the message customizer specified
+		 */
+		SendMessageBuilder<T> withMessageCustomizer(MessageSpecBuilderCustomizer<T> customizer);
+
+		/**
+		 * Specifies the custom message router to use when sending the message.
+		 * @param messageRouter the custom message router
+		 * @return the current builder with the custom message router specified
+		 */
+		SendMessageBuilder<T> withCustomRouter(MessageRouter messageRouter);
+
+		/**
+		 * Specifies the customizer to use to further configure the reactive sender
+		 * builder.
+		 * @param customizer the reactive sender builder customizer
+		 * @return the current builder with the reactive sender builder customizer
+		 * specified
+		 */
+		SendMessageBuilder<T> withSenderCustomizer(ReactiveMessageSenderBuilderCustomizer<T> customizer);
+
+		/**
+		 * Send the message in a reactive manner using the configured specification.
+		 * @return the id assigned by the broker to the published message
+		 */
+		Mono<MessageId> send();
+
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderTemplate.java

@@ -0,0 +1,159 @@
+/*
+ * 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.core.reactive;
+
+import java.util.Collections;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.api.MessageSpec;
+import org.apache.pulsar.reactive.client.api.MessageSpecBuilder;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.pulsar.core.SchemaUtils;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * A thread-safe template for executing high-level reactive Pulsar operations.
+ *
+ * @param <T> the message payload type
+ * @author Christophe Bornet
+ */
+public class ReactivePulsarSenderTemplate<T> implements ReactivePulsarSenderOperations<T> {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory;
+
+	private Schema<T> schema;
+
+	/**
+	 * Construct a template instance with observation configuration.
+	 * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar
+	 * reactive senders
+	 */
+	public ReactivePulsarSenderTemplate(ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory) {
+		this.reactiveMessageSenderFactory = reactiveMessageSenderFactory;
+	}
+
+	@Override
+	public Mono<MessageId> send(T message) {
+		return doSend(null, message, null, null, null);
+	}
+
+	@Override
+	public Mono<MessageId> send(String topic, T message) {
+		return doSend(topic, message, null, null, null);
+	}
+
+	@Override
+	public SendMessageBuilderImpl<T> newMessage(T message) {
+		return new SendMessageBuilderImpl<>(this, message);
+	}
+
+	/**
+	 * Set the schema to use on this template.
+	 * @param schema provides the {@link Schema} used on this template
+	 */
+	public void setSchema(Schema<T> schema) {
+		this.schema = schema;
+	}
+
+	private Mono<MessageId> doSend(String topic, T message,
+			MessageSpecBuilderCustomizer<T> messageSpecBuilderCustomizer, MessageRouter messageRouter,
+			ReactiveMessageSenderBuilderCustomizer<T> customizer) {
+		final String topicName = ReactiveMessageSenderUtils.resolveTopicName(topic, this.reactiveMessageSenderFactory);
+		this.logger.trace(() -> String.format("Sending reative msg to '%s' topic", topicName));
+
+		final ReactiveMessageSender<T> sender = createMessageSender(topic, message, messageRouter, customizer);
+
+		MessageSpecBuilder<T> messageSpecBuilder = MessageSpec.builder(message);
+
+		if (messageSpecBuilderCustomizer != null) {
+			messageSpecBuilderCustomizer.customize(messageSpecBuilder);
+		}
+
+		MessageSpec<T> messageSpec = messageSpecBuilder.build();
+		return sender.sendMessage(Mono.just(messageSpec))
+				.doOnError(
+						ex -> this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName)))
+				.doOnSuccess(msgId -> this.logger.trace(() -> String.format("Sent msg to '%s' topic", topicName)));
+	}
+
+	private ReactiveMessageSender<T> createMessageSender(String topic, T message, MessageRouter messageRouter,
+			ReactiveMessageSenderBuilderCustomizer<T> customizer) {
+		Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
+		return this.reactiveMessageSenderFactory.createReactiveMessageSender(topic, schema, messageRouter,
+				customizer == null ? Collections.emptyList() : Collections.singletonList(customizer));
+	}
+
+	public static class SendMessageBuilderImpl<T> implements SendMessageBuilder<T> {
+
+		private final ReactivePulsarSenderTemplate<T> template;
+
+		private final T message;
+
+		private String topic;
+
+		private MessageSpecBuilderCustomizer<T> messageCustomizer;
+
+		private MessageRouter messageRouter;
+
+		private ReactiveMessageSenderBuilderCustomizer<T> senderCustomizer;
+
+		SendMessageBuilderImpl(ReactivePulsarSenderTemplate<T> template, T message) {
+			this.template = template;
+			this.message = message;
+		}
+
+		@Override
+		public SendMessageBuilderImpl<T> withTopic(String topic) {
+			this.topic = topic;
+			return this;
+		}
+
+		@Override
+		public SendMessageBuilderImpl<T> withMessageCustomizer(MessageSpecBuilderCustomizer<T> messageCustomizer) {
+			this.messageCustomizer = messageCustomizer;
+			return this;
+		}
+
+		@Override
+		public SendMessageBuilderImpl<T> withCustomRouter(MessageRouter messageRouter) {
+			this.messageRouter = messageRouter;
+			return this;
+		}
+
+		@Override
+		public SendMessageBuilderImpl<T> withSenderCustomizer(
+				ReactiveMessageSenderBuilderCustomizer<T> senderCustomizer) {
+			this.senderCustomizer = senderCustomizer;
+			return this;
+		}
+
+		@Override
+		public Mono<MessageId> send() {
+			return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter,
+					this.senderCustomizer);
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java

@@ -0,0 +1,158 @@
+/*
+ * 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.core.reactive;
+
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.apache.pulsar.reactive.client.api.MessageSpec;
+import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Common tests for {@link DefaultReactivePulsarSenderFactory}
+ *
+ * @author Christophe Bornet
+ */
+@SuppressWarnings("unchecked")
+class DefaultReactiveMessageSenderFactoryTests {
+
+	protected final Schema<String> schema = Schema.STRING;
+
+	private ProducerBuilder<String> producerBuilder;
+
+	private PulsarClient pulsarClient;
+
+	@BeforeEach
+	void createPulsarClient() {
+		pulsarClient = mock(PulsarClient.class);
+		producerBuilder = mock(ProducerBuilder.class);
+		Producer<String> producer = mock(Producer.class);
+		TypedMessageBuilder<String> mockMessage = mock(TypedMessageBuilder.class);
+
+		when(mockMessage.sendAsync()).thenReturn(CompletableFuture.completedFuture(MessageId.latest));
+		when(producer.newMessage()).thenReturn(mockMessage);
+		when(producer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+		when(producerBuilder.createAsync()).thenReturn(CompletableFuture.completedFuture(producer));
+		when(pulsarClient.newProducer(schema)).thenReturn(producerBuilder);
+	}
+
+	@Test
+	void createProducerWithSpecificTopic() {
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				null);
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic1", schema);
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic1", null);
+	}
+
+	@Test
+	void createProducerWithSpecificTopicAndMessageRouter() {
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				null);
+		MessageRouter router = mock(MessageRouter.class);
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic1", schema, router);
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic1", router);
+	}
+
+	@Test
+	void createProducerWithDefaultTopic() {
+		MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
+		senderSpec.setTopicName("topic0");
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				senderSpec);
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender(null, schema);
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic0", null);
+	}
+
+	@Test
+	void createProducerWithDefaultTopicAndMessageRouter() {
+		MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
+		senderSpec.setTopicName("topic0");
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				senderSpec);
+		MessageRouter router = mock(MessageRouter.class);
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender(null, schema, router);
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic0", router);
+	}
+
+	@Test
+	void createProducerWithSingleProducerCustomizer() {
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				null);
+		ReactiveMessageSenderBuilderCustomizer<String> customizer = builder -> builder.topic("topic1");
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic0", schema, null,
+				Collections.singletonList(customizer));
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic1", null);
+	}
+
+	@Test
+	void createProducerWithMultipleProducerCustomizer() {
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				null);
+		ReactiveMessageSenderBuilderCustomizer<String> customizer1 = builder -> builder.topic("topic1");
+		MessageRouter router = mock(MessageRouter.class);
+		ReactiveMessageSenderBuilderCustomizer<String> customizer2 = builder -> builder.messageRouter(router);
+		ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic0", schema, null,
+				Arrays.asList(customizer1, customizer2));
+		sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
+		assertSenderHasTopicAndRouter("topic1", router);
+	}
+
+	@Test
+	void createProducerWithNoTopic() {
+		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
+				null);
+		assertThatIllegalArgumentException().isThrownBy(() -> senderFactory.createReactiveMessageSender(null, schema))
+				.withMessageContaining("Topic must be specified when no default topic is configured");
+	}
+
+	protected void assertSenderHasTopicAndRouter(String topic, MessageRouter router) {
+		verify(producerBuilder).topic(topic);
+		if (router != null) {
+			verify(producerBuilder).messageRouter(router);
+		}
+		else {
+			verify(producerBuilder, never()).messageRouter(any());
+		}
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java

@@ -0,0 +1,235 @@
+/*
+ * 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.core.reactive;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Duration;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.springframework.pulsar.core.PulsarTestContainerSupport;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Tests for {@link ReactivePulsarSenderTemplate}.
+ *
+ * @author Christophe Bornet
+ */
+class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
+
+	@Test
+	void sendMessageWithSpecificSchemaTest() throws Exception {
+		String topic = "smt-specific-schema-topic-reactive";
+		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
+				.build()) {
+			try (Consumer<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
+					.subscriptionName("test-specific-schema-subscription").subscribe()) {
+				MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
+				senderSpec.setTopicName(topic);
+				ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client,
+						senderSpec);
+				ReactivePulsarSenderTemplate<Foo> pulsarTemplate = new ReactivePulsarSenderTemplate<>(producerFactory);
+				pulsarTemplate.setSchema(Schema.JSON(Foo.class));
+				Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
+				pulsarTemplate.send(foo).subscribe();
+				assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue)
+						.isEqualTo(foo);
+			}
+		}
+	}
+
+	@ParameterizedTest(name = "{0}")
+	@MethodSource("sendMessageTestProvider")
+	void sendMessageTest(String testName, SendTestArgs testArgs) throws Exception {
+		// Use the test args to construct the params to pass to send handler
+		String topic = testName;
+		String subscription = topic + "-sub";
+		String msgPayload = topic + "-msg";
+		MessageRouter router = null;
+		if (testArgs.useCustomRouter) {
+			router = mock(MessageRouter.class);
+			when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0);
+		}
+		MessageSpecBuilderCustomizer<String> messageCustomizer = null;
+		if (testArgs.useMessageCustomizer) {
+			messageCustomizer = (mb) -> mb.key("foo-key");
+		}
+		ReactiveMessageSenderBuilderCustomizer<String> senderCustomizer = null;
+		if (testArgs.useSenderCustomizer) {
+			senderCustomizer = (sb) -> sb.producerName("foo-producer");
+		}
+
+		if (router != null) {
+			try (PulsarAdmin admin = PulsarAdmin.builder()
+					.serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()).build()) {
+				admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1);
+			}
+		}
+		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
+				.build()) {
+			try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic)
+					.subscriptionName(subscription).subscribe()) {
+				MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
+				if (!testArgs.useSpecificTopic) {
+					senderSpec.setTopicName(topic);
+				}
+				ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
+						senderSpec);
+				ReactivePulsarSenderTemplate<String> pulsarTemplate = new ReactivePulsarSenderTemplate<>(senderFactory);
+				Mono<MessageId> sendResponse;
+				if (testArgs.useSimpleApi) {
+					sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.send(topic, msgPayload)
+							: pulsarTemplate.send(msgPayload);
+				}
+				else {
+					ReactivePulsarSenderTemplate.SendMessageBuilderImpl<String> messageBuilder = pulsarTemplate
+							.newMessage(msgPayload);
+					if (testArgs.useSpecificTopic) {
+						messageBuilder = messageBuilder.withTopic(topic);
+					}
+					if (messageCustomizer != null) {
+						messageBuilder = messageBuilder.withMessageCustomizer(messageCustomizer);
+					}
+					if (router != null) {
+						messageBuilder = messageBuilder.withCustomRouter(router);
+					}
+					if (senderCustomizer != null) {
+						messageBuilder = messageBuilder.withSenderCustomizer(senderCustomizer);
+					}
+					sendResponse = messageBuilder.send();
+				}
+				sendResponse.subscribe();
+
+				Message<String> msg = consumer.receive(3, TimeUnit.SECONDS);
+
+				assertThat(msg).isNotNull();
+				assertThat(msg.getData()).asString().isEqualTo(msgPayload);
+				if (messageCustomizer != null) {
+					assertThat(msg.getKey()).isEqualTo("foo-key");
+				}
+				if (router != null) {
+					verify(router).choosePartition(argThat((Message<String> m) -> m.getTopicName().equals(topic)),
+							any(TopicMetadata.class));
+				}
+				if (senderCustomizer != null) {
+					assertThat(msg.getProducerName()).isEqualTo("foo-producer");
+				}
+				// Make sure the producer was closed by the template (albeit indirectly as
+				// client removes closed producers)
+				await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(client).extracting("producers")
+						.asInstanceOf(InstanceOfAssertFactories.COLLECTION).isEmpty());
+			}
+		}
+	}
+
+	private static Stream<Arguments> sendMessageTestProvider() {
+		return Stream.of(arguments("sendReactiveMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false)),
+				arguments("sendReactiveMessageToDefaultTopicWithSimpleApi",
+						SendTestArgs.useSpecificTopic(false).useSimpleApi(true)),
+				arguments("sendReactiveMessageToDefaultTopicWithRouter",
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true)),
+				arguments("sendReactiveMessageToDefaultTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true)),
+				arguments("sendReactiveMessageToDefaultTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(false).useSenderCustomizer(true)),
+				arguments("sendReactiveMessageToDefaultTopicWithAllOptions",
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true)
+								.useSenderCustomizer(true)),
+				arguments("sendReactiveMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true)),
+				arguments("sendReactiveMessageToSpecificTopicWithSimpleApi",
+						SendTestArgs.useSpecificTopic(true).useSimpleApi(true)),
+				arguments("sendReactiveMessageToSpecificTopicWithRouter",
+						SendTestArgs.useSpecificTopic(true).useCustomRouter(true)),
+				arguments("sendReactiveMessageToSpecificTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true)),
+				arguments("sendReactiveMessageToSpecificTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(true).useSenderCustomizer(true)),
+				arguments("sendReactiveMessageToSpecificTopicWithAllOptions", SendTestArgs.useSpecificTopic(true)
+						.useCustomRouter(true).useMessageCustomizer(true).useSenderCustomizer(true)));
+	}
+
+	static final class SendTestArgs {
+
+		private boolean useSpecificTopic;
+
+		private boolean useCustomRouter;
+
+		private boolean useMessageCustomizer;
+
+		private boolean useSenderCustomizer;
+
+		private boolean useSimpleApi;
+
+		private SendTestArgs(boolean useSpecificTopic) {
+			this.useSpecificTopic = useSpecificTopic;
+		}
+
+		static SendTestArgs useSpecificTopic(boolean useSpecificTopic) {
+			return new SendTestArgs(useSpecificTopic);
+		}
+
+		SendTestArgs useCustomRouter(boolean useCustomRouter) {
+			this.useCustomRouter = useCustomRouter;
+			return this;
+		}
+
+		SendTestArgs useMessageCustomizer(boolean useMessageCustomizer) {
+			this.useMessageCustomizer = useMessageCustomizer;
+			return this;
+		}
+
+		SendTestArgs useSenderCustomizer(boolean useSenderCustomizer) {
+			this.useSenderCustomizer = useSenderCustomizer;
+			return this;
+		}
+
+		SendTestArgs useSimpleApi(boolean useSimpleApi) {
+			this.useSimpleApi = useSimpleApi;
+			return this;
+		}
+
+	}
+
+	record Foo(String foo, String bar) {
+	}
+
+}