From 4944be6fecc2afc61b6150c6769821bef0717149 Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Thu, 17 Nov 2022 00:43:19 +0100 Subject: [PATCH] Add autoconfiguration for ReactivePulsarListener --- .../DocumentConfigurationProperties.java | 1 + ...ReactiveAnnotationDrivenConfiguration.java | 70 ++++++++++ .../PulsarReactiveAutoConfiguration.java | 2 + .../PulsarReactiveProperties.java | 37 ++++++ .../PulsarReactiveAutoConfigurationTests.java | 51 +++++++- .../ReactivePulsarListenerTests.java | 121 ++++++++++++++++++ 6 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java create mode 100644 spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java 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 ab1a8fb0..b036c008 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 @@ -79,6 +79,7 @@ public class DocumentConfigurationProperties extends DefaultTask { }); snippets.add("application-properties.pulsar-reactive-consumer", "Pulsar Reactive Consumer Properties", (c) -> { c.accept("spring.pulsar.reactive.consumer"); + c.accept("spring.pulsar.reactive.listener"); }); snippets.add("application-properties.pulsar-reactive-reader", "Pulsar Reactive Reader Properties", (c) -> { c.accept("spring.pulsar.reactive.reader"); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java new file mode 100644 index 00000000..af249378 --- /dev/null +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java @@ -0,0 +1,70 @@ +/* + * 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.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.EnablePulsar; +import org.springframework.pulsar.config.PulsarListenerBeanNames; +import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; + +/** + * Configuration for Reactive Pulsar annotation-driven support. + * + * @author Christophe Bornet + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass(EnablePulsar.class) +public class PulsarReactiveAnnotationDrivenConfiguration { + + private final PulsarReactiveProperties properties; + + public PulsarReactiveAnnotationDrivenConfiguration(PulsarReactiveProperties properties) { + this.properties = properties; + } + + @Bean + @ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory") + DefaultReactivePulsarListenerContainerFactory reactivePulsarListenerContainerFactory( + ObjectProvider> consumerFactoryProvider) { + + ReactivePulsarContainerProperties containerProperties = new ReactivePulsarContainerProperties<>(); + containerProperties.setSubscriptionType(this.properties.getConsumer().getSubscriptionType()); + + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + PulsarReactiveProperties.Listener listenerProperties = this.properties.getListener(); + map.from(listenerProperties::getSchemaType).to(containerProperties::setSchemaType); + map.from(listenerProperties::getHandlingTimeout).to(containerProperties::setHandlingTimeout); + + return new DefaultReactivePulsarListenerContainerFactory<>(consumerFactoryProvider.getIfAvailable(), + containerProperties); + } + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + @ConditionalOnMissingBean(name = PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME) + static class EnableReactivePulsarConfiguration { + + } + +} 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 index a5516cd5..91d58ade 100644 --- 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 @@ -31,6 +31,7 @@ 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.context.annotation.Import; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarReaderFactory; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; @@ -50,6 +51,7 @@ import com.github.benmanes.caffeine.cache.Caffeine; @AutoConfiguration(after = PulsarAutoConfiguration.class) @ConditionalOnClass({ ReactivePulsarTemplate.class, ReactivePulsarClient.class }) @EnableConfigurationProperties(PulsarReactiveProperties.class) +@Import({ PulsarReactiveAnnotationDrivenConfiguration.class }) public class PulsarReactiveAutoConfiguration { private final PulsarReactiveProperties properties; 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 index ccb77c52..b1a5ad06 100644 --- 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 @@ -35,6 +35,7 @@ import org.apache.pulsar.client.api.Range; import org.apache.pulsar.client.api.RegexSubscriptionMode; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageConsumerSpec; import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageReaderSpec; import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageSenderSpec; @@ -67,6 +68,8 @@ public class PulsarReactiveProperties { private final Reader reader = new Reader(); + private final Listener listener = new Listener(); + public Sender getSender() { return this.sender; } @@ -79,6 +82,10 @@ public class PulsarReactiveProperties { return this.reader; } + public Listener getListener() { + return this.listener; + } + public ReactiveMessageSenderSpec buildReactiveMessageSenderSpec() { return this.sender.buildReactiveMessageSenderSpec(); } @@ -944,4 +951,34 @@ public class PulsarReactiveProperties { } + public static class Listener { + + /** + * SchemaType of the consumed messages. + */ + private SchemaType schemaType; + + /** + * Duration to wait before the message handling times out. + */ + private Duration handlingTimeout = Duration.ofMinutes(2); + + public SchemaType getSchemaType() { + return this.schemaType; + } + + public void setSchemaType(SchemaType schemaType) { + this.schemaType = schemaType; + } + + public Duration getHandlingTimeout() { + return this.handlingTimeout; + } + + public void setHandlingTimeout(Duration handlingTimeout) { + this.handlingTimeout = handlingTimeout; + } + + } + } 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 index ed3436fa..23dc1e18 100644 --- 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 @@ -44,7 +44,13 @@ 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.annotation.EnablePulsar; +import org.springframework.pulsar.annotation.ReactivePulsarBootstrapConfiguration; +import org.springframework.pulsar.annotation.ReactivePulsarListenerAnnotationBeanPostProcessor; import org.springframework.pulsar.config.PulsarClientFactoryBean; +import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarReaderFactory; import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; @@ -77,12 +83,31 @@ class PulsarReactiveAutoConfigurationTests { (context) -> assertThat(context).hasNotFailed().doesNotHaveBean(PulsarReactiveAutoConfiguration.class)); } + @Test + void annotationDrivenConfigurationSkippedWhenEnablePulsarAnnotationNotOnClasspath() { + this.contextRunner.withClassLoader(new FilteredClassLoader(EnablePulsar.class)) + .run((context) -> assertThat(context).hasNotFailed() + .doesNotHaveBean(PulsarReactiveAnnotationDrivenConfiguration.class)); + } + + @Test + void bootstrapConfigurationSkippedWhenCustomReactivePulsarListenerAnnotationProcessorDefined() { + this.contextRunner + .withBean("org.springframework.pulsar.config.internalReactivePulsarListenerAnnotationProcessor", + String.class, () -> "someFauxBean") + .run((context) -> assertThat(context).hasNotFailed() + .doesNotHaveBean(ReactivePulsarBootstrapConfiguration.class)); + } + @Test void defaultBeansAreAutoConfigured() { this.contextRunner.run((context) -> assertThat(context).hasNotFailed() .hasSingleBean(ReactivePulsarTemplate.class).hasSingleBean(ReactivePulsarClient.class) .hasSingleBean(ProducerCacheProvider.class).hasSingleBean(ReactiveMessageSenderCache.class) - .hasSingleBean(ReactivePulsarSenderFactory.class).getBean(ReactivePulsarTemplate.class)); + .hasSingleBean(ReactivePulsarSenderFactory.class).hasSingleBean(ReactivePulsarTemplate.class) + .hasSingleBean(DefaultReactivePulsarListenerContainerFactory.class) + .hasSingleBean(ReactivePulsarListenerAnnotationBeanPostProcessor.class) + .hasSingleBean(ReactivePulsarListenerEndpointRegistry.class)); } @ParameterizedTest @@ -95,6 +120,30 @@ class PulsarReactiveAutoConfigurationTests { .run((context) -> assertThat(context).hasNotFailed().getBean(beanClass).isSameAs(bean)); } + @Test + void customReactivePulsarListenerContainerFactoryIsRespected() { + ReactivePulsarListenerContainerFactory listenerContainerFactory = mock( + ReactivePulsarListenerContainerFactory.class); + this.contextRunner + .withBean("reactivePulsarListenerContainerFactory", ReactivePulsarListenerContainerFactory.class, + () -> listenerContainerFactory) + .run((context) -> assertThat(context).hasNotFailed() + .getBean(ReactivePulsarListenerContainerFactory.class).isSameAs(listenerContainerFactory)); + } + + @Test + void customReactivePulsarListenerAnnotationBeanPostProcessorIsRespected() { + ReactivePulsarListenerAnnotationBeanPostProcessor listenerAnnotationBeanPostProcessor = mock( + ReactivePulsarListenerAnnotationBeanPostProcessor.class); + this.contextRunner + .withBean("org.springframework.pulsar.config.internalReactivePulsarListenerAnnotationProcessor", + ReactivePulsarListenerAnnotationBeanPostProcessor.class, + () -> listenerAnnotationBeanPostProcessor) + .run((context) -> assertThat(context).hasNotFailed() + .getBean(ReactivePulsarListenerAnnotationBeanPostProcessor.class) + .isSameAs(listenerAnnotationBeanPostProcessor)); + } + @Test void beansAreInjectedInReactivePulsarTemplate() { ReactivePulsarSenderFactory senderFactory = mock(ReactivePulsarSenderFactory.class); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java new file mode 100644 index 00000000..fabad72e --- /dev/null +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java @@ -0,0 +1,121 @@ +/* + * 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 java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.pulsar.annotation.ReactivePulsarListener; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Tests for {@link ReactivePulsarListener}. + * + * @author Christophe Bornet + */ +class ReactivePulsarListenerTests implements PulsarTestContainerSupport { + + static CountDownLatch latch1 = new CountDownLatch(1); + static CountDownLatch latch2 = new CountDownLatch(10); + + @Test + void testBasicListener() throws Exception { + SpringApplication app = new SpringApplication(BasicListenerConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + app.setAllowCircularReferences(true); + + try (ConfigurableApplicationContext context = app + .run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) { + @SuppressWarnings("unchecked") + final PulsarTemplate pulsarTemplate = context.getBean(PulsarTemplate.class); + pulsarTemplate.send("hello-pulsar-exclusive", "John Doe"); + final boolean await = latch1.await(20, TimeUnit.SECONDS); + assertThat(await).isTrue(); + } + } + + @Test + void testFluxListener() throws Exception { + SpringApplication app = new SpringApplication(FluxListenerConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + app.setAllowCircularReferences(true); + + try (ConfigurableApplicationContext context = app + .run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) { + @SuppressWarnings("unchecked") + final PulsarTemplate pulsarTemplate = context.getBean(PulsarTemplate.class); + for (int i = 0; i < 10; i++) { + pulsarTemplate.send("hello-pulsar-exclusive", "John Doe"); + } + final boolean await = latch2.await(10, TimeUnit.SECONDS); + assertThat(await).isTrue(); + } + } + + @Configuration + @Import({ PulsarAutoConfiguration.class, PulsarReactiveAutoConfiguration.class }) + static class BasicListenerConfig { + + @ReactivePulsarListener(subscriptionName = "test-exclusive-sub-1", topics = "hello-pulsar-exclusive", + consumerCustomizer = "consumerCustomizer") + public Mono listen(String foo) { + latch1.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + @Configuration + @Import({ PulsarAutoConfiguration.class, PulsarReactiveAutoConfiguration.class }) + static class FluxListenerConfig { + + @ReactivePulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive", + stream = true, consumerCustomizer = "consumerCustomizer") + public Flux> listen(Flux> messages) { + return messages.doOnNext(t -> latch2.countDown()).map(m -> MessageResult.acknowledge(m.getMessageId())); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + +}