Add autoconfig for reactive sender template (#175)

buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java

@@ -74,6 +74,9 @@ public class DocumentConfigurationProperties extends DefaultTask {
 			c.accept("spring.pulsar.listener");
 		});
 		snippets.add("application-properties.pulsar-administration", "Pulsar Administration Properties", (c) -> c.accept("spring.pulsar.administration"));
+		snippets.add("application-properties.pulsar-reactive-sender", "Pulsar Reactive Sender Properties", (c) -> {
+			c.accept("spring.pulsar.reactive.sender");
+		});
 		snippets.writeTo(this.outputDir.toPath());
 	}
 }

settings.gradle

@@ -26,6 +26,7 @@ include 'spring-pulsar'
 include 'spring-pulsar-dependencies'
 include 'spring-pulsar-spring-boot-autoconfigure'
 include 'spring-pulsar-spring-boot-starter'
+include 'spring-pulsar-reactive-spring-boot-starter'
 include 'spring-pulsar-sample-apps:sample-app1'
 include 'spring-pulsar-sample-apps:sample-app2'
 include 'spring-pulsar-docs'

spring-pulsar-dependencies/build.gradle

@@ -28,5 +28,6 @@ dependencies {
 		api "com.google.protobuf:protobuf-java:$protobufJavaVersion"
 		api "org.apache.pulsar:pulsar-client-all:$pulsarVersion"
 		api "org.apache.pulsar:pulsar-client-reactive-adapter:$pulsarClientReactiveVersion"
+		api "org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine:$pulsarClientReactiveVersion"
 	}
 }

spring-pulsar-reactive-spring-boot-starter/build.gradle

@@ -0,0 +1,12 @@
+plugins {
+	id 'org.springframework.pulsar.spring-module'
+}
+
+description = 'Spring Pulsar Reactive Spring Boot Starter'
+
+dependencies {
+	api project (':spring-pulsar')
+	api project (':spring-pulsar-spring-boot-autoconfigure')
+	api 'org.apache.pulsar:pulsar-client-reactive-adapter'
+	api 'org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine'
+}

spring-pulsar-spring-boot-autoconfigure/build.gradle

@@ -10,6 +10,7 @@ dependencies {
 	annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
 
 	optional project (':spring-pulsar')
+	optional 'org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine'
 	implementation 'org.springframework.boot:spring-boot-starter'
 	implementation 'com.google.code.findbugs:jsr305'
 

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java

@@ -0,0 +1,100 @@
+/*
+ * 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.autoconfigure;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
+import org.apache.pulsar.reactive.client.adapter.ProducerCacheProvider;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
+import org.apache.pulsar.reactive.client.producercache.CaffeineProducerCacheProvider;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory;
+import org.springframework.pulsar.core.reactive.ReactivePulsarSenderFactory;
+import org.springframework.pulsar.core.reactive.ReactivePulsarSenderTemplate;
+
+import com.github.benmanes.caffeine.cache.Caffeine;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for Apache Pulsar.
+ *
+ * @author Chris Bono
+ * @author Christophe Bornet
+ */
+@AutoConfiguration(after = PulsarAutoConfiguration.class)
+@ConditionalOnClass(ReactivePulsarClient.class)
+@EnableConfigurationProperties(PulsarReactiveProperties.class)
+public class PulsarReactiveAutoConfiguration {
+
+	private final PulsarReactiveProperties properties;
+
+	public PulsarReactiveAutoConfiguration(PulsarReactiveProperties properties) {
+		this.properties = properties;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ReactivePulsarClient pulsarReactivePulsarClient(PulsarClient pulsarClient) {
+		return AdaptedReactivePulsarClientFactory.create(pulsarClient);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	@ConditionalOnClass(CaffeineProducerCacheProvider.class)
+	@ConditionalOnProperty(name = "spring.pulsar.reactive.sender.cache.enabled", havingValue = "true",
+			matchIfMissing = true)
+	public ProducerCacheProvider pulsarProducerCacheProvider() {
+		PulsarReactiveProperties.Cache cache = this.properties.getSender().getCache();
+		Caffeine<Object, Object> caffeine = Caffeine.newBuilder().expireAfterAccess(cache.getExpireAfterAccess())
+				.maximumSize(cache.getMaximumSize()).initialCapacity(cache.getInitialCapacity());
+		return new CaffeineProducerCacheProvider(caffeine);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	@ConditionalOnProperty(name = "spring.pulsar.reactive.sender.cache.enabled", havingValue = "true",
+			matchIfMissing = true)
+	public ReactiveMessageSenderCache pulsarReactiveMessageSenderCache(
+			ObjectProvider<ProducerCacheProvider> producerCacheProvider) {
+		return producerCacheProvider.stream().findFirst().map(AdaptedReactivePulsarClientFactory::createCache)
+				.orElseGet(AdaptedReactivePulsarClientFactory::createCache);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory(ReactivePulsarClient pulsarReactivePulsarClient,
+			ObjectProvider<ReactiveMessageSenderCache> cache) {
+		return new DefaultReactivePulsarSenderFactory<>(pulsarReactivePulsarClient,
+				this.properties.buildReactiveMessageSenderSpec(), cache.getIfAvailable());
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ReactivePulsarSenderTemplate<?> pulsarReactiveSenderTemplate(
+			ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory) {
+		return new ReactivePulsarSenderTemplate<>(reactivePulsarSenderFactory);
+	}
+
+}

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java

@@ -0,0 +1,320 @@
+/*
+ * 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.autoconfigure;
+
+import java.time.Duration;
+
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerAccessMode;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.PropertyMapper;
+
+/**
+ * Configuration properties for Spring for the Apache Pulsar reactive client.
+ * <p>
+ * Users should refer to Pulsar reactive client documentation for complete descriptions of
+ * these properties.
+ *
+ * @author Christophe Bornet
+ */
+@ConfigurationProperties(prefix = "spring.pulsar.reactive")
+public class PulsarReactiveProperties {
+
+	private final Sender sender = new Sender();
+
+	public Sender getSender() {
+		return this.sender;
+	}
+
+	public ReactiveMessageSenderSpec buildReactiveMessageSenderSpec() {
+		return this.sender.buildReactiveMessageSenderSpec();
+	}
+
+	public static class Sender {
+
+		/**
+		 * Topic the producer will publish to.
+		 */
+		private String topicName;
+
+		/**
+		 * Name for the producer. If not assigned, a unique name is generated.
+		 */
+		private String producerName;
+
+		/**
+		 * Time before a message has to be acknowledged by the broker in milliseconds.
+		 */
+		private Duration sendTimeout = Duration.ofSeconds(30);
+
+		/**
+		 * Maximum number of pending messages for the producer.
+		 */
+		private Integer maxPendingMessages = 1000;
+
+		/**
+		 * Maximum number of pending messages across all the partitions.
+		 */
+		private Integer maxPendingMessagesAcrossPartitions = 50000;
+
+		/**
+		 * Message routing mode for a partitioned producer.
+		 */
+		private MessageRoutingMode messageRoutingMode = MessageRoutingMode.RoundRobinPartition;
+
+		/**
+		 * Message hashing scheme to choose the partition to which the message is
+		 * published.
+		 */
+		private HashingScheme hashingScheme = HashingScheme.JavaStringHash;
+
+		/**
+		 * Action the producer will take in case of encryption failure.
+		 */
+		private ProducerCryptoFailureAction cryptoFailureAction = ProducerCryptoFailureAction.FAIL;
+
+		/**
+		 * Time period within which the messages sent will be batched in milliseconds.
+		 */
+		private Duration batchingMaxPublishDelay = Duration.ofMillis(1);
+
+		/**
+		 * Maximum number of messages to be batched.
+		 */
+		private Integer batchingMaxMessages = 1000;
+
+		/**
+		 * Whether to automatically batch messages.
+		 */
+		private Boolean batchingEnabled = true;
+
+		/**
+		 * Whether to split large-size messages into multiple chunks.
+		 */
+		private Boolean chunkingEnabled = false;
+
+		/**
+		 * Message compression type.
+		 */
+		private CompressionType compressionType;
+
+		/**
+		 * Name of the initial subscription of the topic.
+		 */
+		private String initialSubscriptionName;
+
+		/**
+		 * Type of access to the topic the producer requires.
+		 */
+		private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
+
+		private final Cache cache = new Cache();
+
+		public String getTopicName() {
+			return this.topicName;
+		}
+
+		public void setTopicName(String topicName) {
+			this.topicName = topicName;
+		}
+
+		public String getProducerName() {
+			return this.producerName;
+		}
+
+		public void setProducerName(String producerName) {
+			this.producerName = producerName;
+		}
+
+		public Duration getSendTimeout() {
+			return this.sendTimeout;
+		}
+
+		public void setSendTimeout(Duration sendTimeout) {
+			this.sendTimeout = sendTimeout;
+		}
+
+		public Integer getMaxPendingMessages() {
+			return this.maxPendingMessages;
+		}
+
+		public void setMaxPendingMessages(Integer maxPendingMessages) {
+			this.maxPendingMessages = maxPendingMessages;
+		}
+
+		public Integer getMaxPendingMessagesAcrossPartitions() {
+			return this.maxPendingMessagesAcrossPartitions;
+		}
+
+		public void setMaxPendingMessagesAcrossPartitions(Integer maxPendingMessagesAcrossPartitions) {
+			this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions;
+		}
+
+		public MessageRoutingMode getMessageRoutingMode() {
+			return this.messageRoutingMode;
+		}
+
+		public void setMessageRoutingMode(MessageRoutingMode messageRoutingMode) {
+			this.messageRoutingMode = messageRoutingMode;
+		}
+
+		public HashingScheme getHashingScheme() {
+			return this.hashingScheme;
+		}
+
+		public void setHashingScheme(HashingScheme hashingScheme) {
+			this.hashingScheme = hashingScheme;
+		}
+
+		public ProducerCryptoFailureAction getCryptoFailureAction() {
+			return this.cryptoFailureAction;
+		}
+
+		public void setCryptoFailureAction(ProducerCryptoFailureAction cryptoFailureAction) {
+			this.cryptoFailureAction = cryptoFailureAction;
+		}
+
+		public Duration getBatchingMaxPublishDelay() {
+			return this.batchingMaxPublishDelay;
+		}
+
+		public void setBatchingMaxPublishDelay(Duration batchingMaxPublishDelay) {
+			this.batchingMaxPublishDelay = batchingMaxPublishDelay;
+		}
+
+		public Integer getBatchingMaxMessages() {
+			return this.batchingMaxMessages;
+		}
+
+		public void setBatchingMaxMessages(Integer batchingMaxMessages) {
+			this.batchingMaxMessages = batchingMaxMessages;
+		}
+
+		public Boolean getBatchingEnabled() {
+			return this.batchingEnabled;
+		}
+
+		public void setBatchingEnabled(Boolean batchingEnabled) {
+			this.batchingEnabled = batchingEnabled;
+		}
+
+		public Boolean getChunkingEnabled() {
+			return this.chunkingEnabled;
+		}
+
+		public void setChunkingEnabled(Boolean chunkingEnabled) {
+			this.chunkingEnabled = chunkingEnabled;
+		}
+
+		public CompressionType getCompressionType() {
+			return this.compressionType;
+		}
+
+		public void setCompressionType(CompressionType compressionType) {
+			this.compressionType = compressionType;
+		}
+
+		public String getInitialSubscriptionName() {
+			return this.initialSubscriptionName;
+		}
+
+		public void setInitialSubscriptionName(String initialSubscriptionName) {
+			this.initialSubscriptionName = initialSubscriptionName;
+		}
+
+		public ProducerAccessMode getProducerAccessMode() {
+			return this.producerAccessMode;
+		}
+
+		public void setProducerAccessMode(ProducerAccessMode producerAccessMode) {
+			this.producerAccessMode = producerAccessMode;
+		}
+
+		public Cache getCache() {
+			return this.cache;
+		}
+
+		public ReactiveMessageSenderSpec buildReactiveMessageSenderSpec() {
+			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+
+			MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
+
+			map.from(this::getTopicName).to(spec::setTopicName);
+			map.from(this::getProducerName).to(spec::setProducerName);
+			map.from(this::getSendTimeout).to(spec::setSendTimeout);
+			map.from(this::getMaxPendingMessages).to(spec::setMaxPendingMessages);
+			map.from(this::getMaxPendingMessagesAcrossPartitions).to(spec::setMaxPendingMessagesAcrossPartitions);
+			map.from(this::getMessageRoutingMode).to(spec::setMessageRoutingMode);
+			map.from(this::getHashingScheme).to(spec::setHashingScheme);
+			map.from(this::getCryptoFailureAction).to(spec::setCryptoFailureAction);
+			map.from(this::getBatchingMaxPublishDelay).to(spec::setBatchingMaxPublishDelay);
+			map.from(this::getBatchingMaxMessages).to(spec::setBatchingMaxMessages);
+			map.from(this::getBatchingEnabled).to(spec::setBatchingEnabled);
+			map.from(this::getChunkingEnabled).to(spec::setChunkingEnabled);
+			map.from(this::getCompressionType).to(spec::setCompressionType);
+			map.from(this::getInitialSubscriptionName).to(spec::setInitialSubscriptionName);
+			map.from(this::getProducerAccessMode).to(spec::setAccessMode);
+
+			return spec;
+		}
+
+	}
+
+	public static class Cache {
+
+		/** Time period to expire unused entries in the cache. */
+		private Duration expireAfterAccess = Duration.ofMinutes(1);
+
+		/** Maximum size of cache (entries). */
+		private Long maximumSize = 1000L;
+
+		/** Initial size of cache. */
+		private Integer initialCapacity = 50;
+
+		public Duration getExpireAfterAccess() {
+			return this.expireAfterAccess;
+		}
+
+		public void setExpireAfterAccess(Duration expireAfterAccess) {
+			this.expireAfterAccess = expireAfterAccess;
+		}
+
+		public Long getMaximumSize() {
+			return this.maximumSize;
+		}
+
+		public void setMaximumSize(Long maximumSize) {
+			this.maximumSize = maximumSize;
+		}
+
+		public Integer getInitialCapacity() {
+			return this.initialCapacity;
+		}
+
+		public void setInitialCapacity(Integer initialCapacity) {
+			this.initialCapacity = initialCapacity;
+		}
+
+	}
+
+}

spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1,2 @@
 org.springframework.pulsar.autoconfigure.PulsarAutoConfiguration
+org.springframework.pulsar.autoconfigure.PulsarReactiveAutoConfiguration

spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java

@@ -0,0 +1,202 @@
+/*
+ * 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.autoconfigure;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
+import org.apache.pulsar.reactive.client.adapter.ProducerCacheProvider;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
+import org.apache.pulsar.reactive.client.producercache.CaffeineProducerCacheProvider;
+import org.assertj.core.api.AbstractObjectAssert;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.pulsar.config.PulsarClientFactoryBean;
+import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory;
+import org.springframework.pulsar.core.reactive.ReactivePulsarSenderFactory;
+import org.springframework.pulsar.core.reactive.ReactivePulsarSenderTemplate;
+
+/**
+ * Autoconfiguration tests for {@link PulsarReactiveAutoConfiguration}.
+ *
+ * @author Christophe Bornet
+ */
+@SuppressWarnings("unchecked")
+class PulsarReactiveAutoConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(PulsarAutoConfiguration.class))
+			.withConfiguration(AutoConfigurations.of(PulsarReactiveAutoConfiguration.class));
+
+	@Test
+	void autoConfigurationSkippedWhenReactivePulsarClientNotOnClasspath() {
+		this.contextRunner.withClassLoader(new FilteredClassLoader(ReactivePulsarClient.class)).run(
+				(context) -> assertThat(context).hasNotFailed().doesNotHaveBean(PulsarReactiveAutoConfiguration.class));
+	}
+
+	@Test
+	void defaultBeansAreAutoConfigured() {
+		this.contextRunner.run((context) -> assertThat(context).hasNotFailed()
+				.hasSingleBean(ReactivePulsarSenderTemplate.class).hasSingleBean(ReactivePulsarClient.class)
+				.hasSingleBean(ProducerCacheProvider.class).hasSingleBean(ReactiveMessageSenderCache.class)
+				.hasSingleBean(ReactivePulsarSenderFactory.class).getBean(ReactivePulsarSenderTemplate.class));
+	}
+
+	@ParameterizedTest
+	@ValueSource(classes = { ReactivePulsarClient.class, ProducerCacheProvider.class, ReactiveMessageSenderCache.class,
+			ReactivePulsarSenderFactory.class, ReactivePulsarSenderTemplate.class })
+	<T> void customBeanIsRespected(Class<T> beanClass) {
+		T bean = mock(beanClass);
+		this.contextRunner.withBean(beanClass.getName(), beanClass, () -> bean)
+				.run((context) -> assertThat(context).hasNotFailed().getBean(beanClass).isSameAs(bean));
+	}
+
+	@Test
+	void beansAreInjectedInReactivePulsarTemplate() {
+		ReactivePulsarSenderFactory<?> senderFactory = mock(ReactivePulsarSenderFactory.class);
+		this.contextRunner
+				.withBean("customReactivePulsarSenderFactory", ReactivePulsarSenderFactory.class, () -> senderFactory)
+				.run((context -> assertThat(context).hasNotFailed().getBean(ReactivePulsarSenderTemplate.class)
+						.extracting("reactiveMessageSenderFactory")
+						.asInstanceOf(InstanceOfAssertFactories.type(ReactivePulsarSenderFactory.class))
+						.isSameAs(senderFactory)));
+	}
+
+	@Test
+	@SuppressWarnings("rawtypes")
+	void beansAreInjectedInReactivePulsarSenderFactory() throws Exception {
+		ReactivePulsarClient client = mock(ReactivePulsarClient.class);
+		try (ReactiveMessageSenderCache cache = mock(ReactiveMessageSenderCache.class)) {
+			this.contextRunner.withPropertyValues("spring.pulsar.reactive.sender.topic-name=test-topic")
+					.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
+					.withBean("customReactiveMessageSenderCache", ReactiveMessageSenderCache.class, () -> cache)
+					.run((context -> {
+						AbstractObjectAssert<? extends AbstractObjectAssert<?, DefaultReactivePulsarSenderFactory>, DefaultReactivePulsarSenderFactory> senderFactory = assertThat(
+								context).hasNotFailed().getBean(DefaultReactivePulsarSenderFactory.class);
+						senderFactory.extracting(DefaultReactivePulsarSenderFactory::getReactiveMessageSenderSpec)
+								.extracting(ReactiveMessageSenderSpec::getTopicName).isEqualTo("test-topic");
+						senderFactory.extracting("reactivePulsarClient",
+								InstanceOfAssertFactories.type(ReactivePulsarClient.class)).isSameAs(client);
+						senderFactory
+								.extracting("reactiveMessageSenderCache",
+										InstanceOfAssertFactories.type(ReactiveMessageSenderCache.class))
+								.isSameAs(cache);
+					}));
+		}
+	}
+
+	@Test
+	void beansAreInjectedInReactiveMessageSenderCache() throws Exception {
+		try (ProducerCacheProvider provider = mock(ProducerCacheProvider.class)) {
+			this.contextRunner.withBean("customProducerCacheProvider", ProducerCacheProvider.class, () -> provider)
+					.run((context -> {
+						var senderFactory = assertThat(context).hasNotFailed()
+								.getBean(ReactiveMessageSenderCache.class);
+						senderFactory.extracting("cacheProvider")
+								.asInstanceOf(InstanceOfAssertFactories.type(ProducerCacheProvider.class))
+								.isSameAs(provider);
+					}));
+		}
+	}
+
+	@Test
+	@SuppressWarnings("rawtypes")
+	void beansAreInjectedInReactivePulsarClient() throws Exception {
+		try (PulsarClient client = mock(PulsarClient.class)) {
+			PulsarClientFactoryBean factoryBean = new PulsarClientFactoryBean(null) {
+				@Override
+				protected PulsarClient createInstance() {
+					return client;
+				}
+			};
+			this.contextRunner.withBean("customPulsarClient", PulsarClientFactoryBean.class, () -> factoryBean)
+					.run((context -> assertThat(context).hasNotFailed().getBean(ReactivePulsarClient.class)
+							.extracting("reactivePulsarResourceAdapter")
+							.extracting("pulsarClientSupplier", InstanceOfAssertFactories.type(Supplier.class))
+							.extracting(Supplier::get).isSameAs(client)));
+		}
+	}
+
+	@Nested
+	class SenderCacheAutoConfigurationTests {
+
+		@Test
+		void caffeineCacheUsedByDefault() {
+			contextRunner.run(this::assertCaffeineProducerCacheProvider);
+		}
+
+		@Test
+		void caffeineCacheCanBeConfigured() {
+			contextRunner
+					.withPropertyValues("spring.pulsar.reactive.sender.cache.expire-after-access=100s",
+							"spring.pulsar.reactive.sender.cache.maximum-size=5150",
+							"spring.pulsar.reactive.sender.cache.initial-capacity=200")
+					.run((context) -> assertCaffeineProducerCacheProvider(context).extracting("cache")
+							.extracting("cache").hasFieldOrPropertyWithValue("maximum", 5150L)
+							.hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100)));
+		}
+
+		@Test
+		void defaultClientCacheIsUsedIfCaffeineProducerCacheProviderNotOnClasspath() {
+			ReactiveMessageSenderCache cache = AdaptedReactivePulsarClientFactory.createCache();
+			try (MockedStatic<AdaptedReactivePulsarClientFactory> mockedClientFactory = Mockito
+					.mockStatic(AdaptedReactivePulsarClientFactory.class)) {
+				mockedClientFactory.when(AdaptedReactivePulsarClientFactory::createCache).thenReturn(cache);
+				contextRunner.withClassLoader(new FilteredClassLoader(CaffeineProducerCacheProvider.class))
+						.run((context) -> assertThat(context).hasNotFailed()
+								.doesNotHaveBean(ProducerCacheProvider.class)
+								.hasSingleBean(ReactiveMessageSenderCache.class)
+								.getBean(ReactiveMessageSenderCache.class).isSameAs(cache));
+				mockedClientFactory.verify(AdaptedReactivePulsarClientFactory::createCache);
+			}
+		}
+
+		@Test
+		void cacheCanBeDisabled() {
+			contextRunner.withPropertyValues("spring.pulsar.reactive.sender.cache.enabled=false").run((context -> {
+				assertThat(context).hasNotFailed().doesNotHaveBean(ProducerCacheProvider.class)
+						.doesNotHaveBean(ReactiveMessageSenderCache.class);
+			}));
+		}
+
+		private AbstractObjectAssert<?, ProducerCacheProvider> assertCaffeineProducerCacheProvider(
+				AssertableApplicationContext context) {
+			return assertThat(context).hasNotFailed().hasSingleBean(ProducerCacheProvider.class)
+					.hasSingleBean(ReactiveMessageSenderCache.class).getBean(ProducerCacheProvider.class)
+					.isExactlyInstanceOf(CaffeineProducerCacheProvider.class);
+		}
+
+	}
+
+}

spring-pulsar-spring-boot-autoconfigure/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker

@@ -0,0 +1 @@
+mock-maker-inline

spring-pulsar/build.gradle

@@ -9,7 +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.apache.pulsar:pulsar-client-reactive-adapter'
 	api 'org.springframework:spring-context'
 	api 'org.springframework:spring-messaging'
 	api 'org.springframework:spring-tx'