diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java index 2fee9254..63bec101 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java +++ b/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()); } } diff --git a/settings.gradle b/settings.gradle index 212ed9f3..49a22913 100644 --- a/settings.gradle +++ b/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' diff --git a/spring-pulsar-dependencies/build.gradle b/spring-pulsar-dependencies/build.gradle index a53aa188..de51e4df 100644 --- a/spring-pulsar-dependencies/build.gradle +++ b/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" } } diff --git a/spring-pulsar-reactive-spring-boot-starter/build.gradle b/spring-pulsar-reactive-spring-boot-starter/build.gradle new file mode 100644 index 00000000..53dbebdd --- /dev/null +++ b/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' +} diff --git a/spring-pulsar-spring-boot-autoconfigure/build.gradle b/spring-pulsar-spring-boot-autoconfigure/build.gradle index a91d8566..00683e1e 100644 --- a/spring-pulsar-spring-boot-autoconfigure/build.gradle +++ b/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' diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java new file mode 100644 index 00000000..f9728e42 --- /dev/null +++ b/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 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) { + return producerCacheProvider.stream().findFirst().map(AdaptedReactivePulsarClientFactory::createCache) + .orElseGet(AdaptedReactivePulsarClientFactory::createCache); + } + + @Bean + @ConditionalOnMissingBean + public ReactivePulsarSenderFactory reactivePulsarSenderFactory(ReactivePulsarClient pulsarReactivePulsarClient, + ObjectProvider cache) { + return new DefaultReactivePulsarSenderFactory<>(pulsarReactivePulsarClient, + this.properties.buildReactiveMessageSenderSpec(), cache.getIfAvailable()); + } + + @Bean + @ConditionalOnMissingBean + public ReactivePulsarSenderTemplate pulsarReactiveSenderTemplate( + ReactivePulsarSenderFactory reactivePulsarSenderFactory) { + return new ReactivePulsarSenderTemplate<>(reactivePulsarSenderFactory); + } + +} diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java new file mode 100644 index 00000000..b56c840f --- /dev/null +++ b/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. + *

+ * 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; + } + + } + +} diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index dc6964b1..93aa9162 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/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 diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java new file mode 100644 index 00000000..5593018a --- /dev/null +++ b/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 }) + void customBeanIsRespected(Class 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, 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 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 assertCaffeineProducerCacheProvider( + AssertableApplicationContext context) { + return assertThat(context).hasNotFailed().hasSingleBean(ProducerCacheProvider.class) + .hasSingleBean(ReactiveMessageSenderCache.class).getBean(ProducerCacheProvider.class) + .isExactlyInstanceOf(CaffeineProducerCacheProvider.class); + } + + } + +} diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/spring-pulsar-spring-boot-autoconfigure/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker new file mode 100644 index 00000000..1f0955d4 --- /dev/null +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker @@ -0,0 +1 @@ +mock-maker-inline diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index fc6d839c..dbea388a 100644 --- a/spring-pulsar/build.gradle +++ b/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'