From 836d385d8696f28f1229ade22efd3fa6adf15720 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 20 Nov 2023 13:32:25 -0600 Subject: [PATCH] Automatically set customizer on listeners (#495) When there is only a single customizer and a single listener defined in the application then the customizer will automatically be associated with the listener. This works for @PulsarListener, @PulsarReader, and @ReactivePulsarListener. See #480 --- .../modules/ROOT/pages/reference/pulsar.adoc | 24 +- .../ROOT/pages/reference/reactive-pulsar.adoc | 2 + .../MethodReactivePulsarListenerEndpoint.java | 4 + ...arListenerAnnotationBeanPostProcessor.java | 35 ++- ...ReactivePulsarListenerCustomizerTests.java | 238 ++++++++++++++++++ ...arListenerAnnotationBeanPostProcessor.java | 38 +-- ...lsarReaderAnnotationBeanPostProcessor.java | 26 +- .../config/MethodPulsarListenerEndpoint.java | 4 + .../config/MethodPulsarReaderEndpoint.java | 4 + .../PulsarListenerCustomizerTests.java | 194 ++++++++++++++ .../pulsar/listener/PulsarListenerTests.java | 98 +------- .../listener/PulsarListenerTestsBase.java | 102 ++++++++ .../reader/PulsarReaderCustomizerTests.java | 196 +++++++++++++++ .../pulsar/reader/PulsarReaderTests.java | 65 +---- .../pulsar/reader/PulsarReaderTestsBase.java | 87 +++++++ 15 files changed, 929 insertions(+), 188 deletions(-) create mode 100644 spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerCustomizerTests.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerCustomizerTests.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderCustomizerTests.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTestsBase.java diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc index bbadfca5..ce8e730a 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc @@ -299,6 +299,26 @@ void listen(String message) { TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties +==== Customizing the ConsumerBuilder + +You can customize any fields available through `ConsumerBuilder` using a `PulsarListenerConsumerBuilderCustomizer` by providing a `@Bean` of type `PulsarListenerConsumerBuilderCustomizer` and then making it available to the `PulsarListener` as shown below. + +[source, java] +---- +@PulsarListener(topics = "hello-topic", consumerCustomizer = "myCustomizer") +public void listen(String message) { + System.out.println("Message Received: " + message); +} + +@Bean +PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return (builder) -> builder.consumerName("myConsumer"); +} +---- + +TIP: If your application only has a single `@PulsarListener` and a single `PulsarListenerConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied. + + [[schema-info-listener-imperative]] :listener-class: PulsarListener include::schema-info/schema-info-listener.adoc[leveloffset=+1] @@ -1112,7 +1132,7 @@ Suppose you want the reader to start reading messages arbitrarily from a topic o ==== Customizing the ReaderBuilder You can customize any fields available through `ReaderBuilder` using a `PulsarReaderReaderBuilderCustomizer` in Spring for Apache Pulsar. -You can provide a `@Bean` of type `PulsarReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below. +You can provide a `@Bean` of type `PulsarReaderReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below. [source, java] ---- @@ -1131,6 +1151,8 @@ public PulsarReaderReaderBuilderCustomizer myCustomizer() { } ---- +TIP: If your application only has a single `@PulsarReader` and a single `PulsarReaderReaderBuilderCustomizer` bean registered then the customizer will be automatically applied. + [[topic-resolution-process-imperative]] == Topic Resolution include::topic-resolution.adoc[leveloffset=+1] diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc index 36ae2f5d..69a16738 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc @@ -270,6 +270,8 @@ ReactivePulsarListenerMessageConsumerBuilderCustomizer myConsumerCustomi } ---- +TIP: If your application only has a single `@ReactivePulsarListener` and a single `ReactivePulsarListenerMessageConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied. + You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder. This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `ReactivePulsarListener` methods whose configuration varies. diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java index d91f97af..4534f68b 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java @@ -226,6 +226,10 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul this.deadLetterPolicy = deadLetterPolicy; } + public ReactiveMessageConsumerBuilderCustomizer getConsumerCustomizer() { + return this.consumerCustomizer; + } + public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer consumerCustomizer) { this.consumerCustomizer = consumerCustomizer; } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java index 5420001d..65ce39d2 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java @@ -110,13 +110,13 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra private final AtomicInteger counter = new AtomicInteger(); + private final List> processedEndpoints = new ArrayList<>(); + @Override public void afterSingletonsInstantiated() { this.registrar.setBeanFactory(this.beanFactory); - this.beanFactory.getBeanProvider(PulsarListenerConfigurer.class) .forEach(c -> c.configurePulsarListeners(this.registrar)); - if (this.registrar.getEndpointRegistry() == null) { if (this.endpointRegistry == null) { Assert.state(this.beanFactory != null, @@ -127,12 +127,11 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra } this.registrar.setEndpointRegistry(this.endpointRegistry); } - if (this.defaultContainerFactoryBeanName != null) { this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName); } addFormatters(this.messageHandlerMethodFactory.getDefaultFormattingConversionService()); - + postProcessEndpointsBeforeRegistration(); // Actually register all listeners this.registrar.afterPropertiesSet(); } @@ -201,14 +200,11 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra @Nullable private ReactivePulsarListenerContainerFactory resolveContainerFactory( ReactivePulsarListener ReactivePulsarListener, Object factoryTarget, String beanName) { - String containerFactory = ReactivePulsarListener.containerFactory(); if (!StringUtils.hasText(containerFactory)) { return null; } - ReactivePulsarListenerContainerFactory factory = null; - Object resolved = resolveExpression(containerFactory); if (resolved instanceof ReactivePulsarListenerContainerFactory) { return (ReactivePulsarListenerContainerFactory) resolved; @@ -230,7 +226,6 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListenerEndpoint endpoint, ReactivePulsarListener reactivePulsarListener, Object bean, String[] topics, String topicPattern) { - endpoint.setBean(bean); endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); endpoint.setSubscriptionName(getEndpointSubscriptionName(reactivePulsarListener)); @@ -239,7 +234,6 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra endpoint.setTopicPattern(topicPattern); resolveSubscriptionType(endpoint, reactivePulsarListener); endpoint.setSchemaType(reactivePulsarListener.schemaType()); - String concurrency = reactivePulsarListener.concurrency(); if (StringUtils.hasText(concurrency)) { endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency")); @@ -249,16 +243,15 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra endpoint.setUseKeyOrderedProcessing( resolveExpressionAsBoolean(useKeyOrderedProcessing, "useKeyOrderedProcessing")); } - String autoStartup = reactivePulsarListener.autoStartup(); if (StringUtils.hasText(autoStartup)) { endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup")); } endpoint.setFluxListener(reactivePulsarListener.stream()); endpoint.setBeanFactory(this.beanFactory); - resolveDeadLetterPolicy(endpoint, reactivePulsarListener); resolveConsumerCustomizer(endpoint, reactivePulsarListener); + this.processedEndpoints.add(endpoint); } private void resolveSubscriptionType(MethodReactivePulsarListenerEndpoint endpoint, @@ -286,9 +279,29 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra } } + @SuppressWarnings("unchecked") + protected void postProcessEndpointsBeforeRegistration() { + if (this.processedEndpoints.size() == 1) { + MethodReactivePulsarListenerEndpoint endpoint = this.processedEndpoints.get(0); + if (endpoint.getConsumerCustomizer() != null) { + return; + } + this.beanFactory.getBeanProvider(ReactivePulsarListenerMessageConsumerBuilderCustomizer.class) + .ifUnique((customizer) -> { + this.logger.info(() -> String + .format("Setting the only registered ReactivePulsarListenerMessageConsumerBuilderCustomizer " + + "on the only registered @ReactivePulsarListener (%s)", endpoint.getId())); + endpoint.setConsumerCustomizer(customizer::customize); + }); + } + } + @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveConsumerCustomizer(MethodReactivePulsarListenerEndpoint endpoint, ReactivePulsarListener reactivePulsarListener) { + if (!StringUtils.hasText(reactivePulsarListener.consumerCustomizer())) { + return; + } Object consumerCustomizer = resolveExpression(reactivePulsarListener.consumerCustomizer()); if (consumerCustomizer instanceof ReactivePulsarListenerMessageConsumerBuilderCustomizer customizer) { endpoint.setConsumerCustomizer(customizer::customize); diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerCustomizerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerCustomizerTests.java new file mode 100644 index 00000000..1a9c8e42 --- /dev/null +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerCustomizerTests.java @@ -0,0 +1,238 @@ +/* + * Copyright 2022-2023 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.reactive.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import java.util.Collections; + +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory; +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.assertj.core.api.ObjectAssert; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.core.DefaultPulsarClientFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.core.PulsarProducerFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithMultipleListenersAndSingleCustomizer.WithMultipleListenersAndSingleCustomizerConfig; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithSingleListenerAndMultipleCustomizers.WithSingleListenerAndMultipleCustomizersConfig; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithSingleListenerAndSingleCustomizer.WithSingleListenerAndSingleCustomizerConfig; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * Tests for the customizers on the {@link ReactivePulsarListener} annotation. + * + * @author Chris Bono + */ +@SpringJUnitConfig +@DirtiesContext +@SuppressWarnings({ "unchecked", "rawtypes" }) +class ReactivePulsarListenerCustomizerTests implements PulsarTestContainerSupport { + + private ObjectAssert assertContainer( + ReactivePulsarListenerEndpointRegistry registry, String containerId) { + return assertThat(registry.getListenerContainer(containerId)).isNotNull() + .isInstanceOf(DefaultReactivePulsarMessageListenerContainer.class) + .asInstanceOf(InstanceOfAssertFactories.type(DefaultReactivePulsarMessageListenerContainer.class)); + } + + @Configuration(proxyBeanMethods = false) + @EnableReactivePulsar + static class TopLevelConfig { + + @Bean + PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { + return new DefaultPulsarProducerFactory<>(pulsarClient); + } + + @Bean + PulsarClient pulsarClient() throws PulsarClientException { + return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient(); + } + + @Bean + ReactivePulsarClient pulsarReactivePulsarClient(PulsarClient pulsarClient) { + return AdaptedReactivePulsarClientFactory.create(pulsarClient); + } + + @Bean + PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + return new PulsarTemplate<>(pulsarProducerFactory); + } + + @Bean + ReactivePulsarConsumerFactory pulsarConsumerFactory(ReactivePulsarClient pulsarClient) { + return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, Collections.emptyList()); + } + + @Bean + ReactivePulsarListenerContainerFactory reactivePulsarListenerContainerFactory( + ReactivePulsarConsumerFactory pulsarConsumerFactory) { + return new DefaultReactivePulsarListenerContainerFactory<>(pulsarConsumerFactory, + new ReactivePulsarContainerProperties<>()); + } + + @Bean + PulsarAdministration pulsarAdministration() { + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleListenerAndSingleCustomizerConfig.class) + class WithSingleListenerAndSingleCustomizer { + + private static ReactivePulsarListenerMessageConsumerBuilderCustomizer MY_CUSTOMIZER = mock( + ReactivePulsarListenerMessageConsumerBuilderCustomizer.class); + + @Test + void customizerIsAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) { + assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> { + var builder = mock(ReactiveMessageConsumerBuilder.class); + container.getConsumerCustomizer().customize(builder); + verify(MY_CUSTOMIZER).customize(builder); + }); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleListenerAndSingleCustomizerConfig { + + @ReactivePulsarListener(id = "singleListenerSingleCustomizer-id", + topics = "singleListenerSingleCustomizer-topic") + void listen(String ignored) { + } + + @Bean + ReactivePulsarListenerMessageConsumerBuilderCustomizer myCustomizer() { + return MY_CUSTOMIZER; + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithMultipleListenersAndSingleCustomizerConfig.class) + class WithMultipleListenersAndSingleCustomizer { + + private static ReactivePulsarListenerMessageConsumerBuilderCustomizer MY_CUSTOMIZER = mock( + ReactivePulsarListenerMessageConsumerBuilderCustomizer.class); + + @Test + void customizerIsNotAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) { + assertContainer(registry, "multiListenerSingleCustomizer1-id").satisfies((container) -> { + var builder = mock(ReactiveMessageConsumerBuilder.class); + container.getConsumerCustomizer().customize(builder); + verify(MY_CUSTOMIZER, never()).customize(builder); + }); + assertContainer(registry, "multiListenerSingleCustomizer2-id").satisfies((container) -> { + var builder = mock(ReactiveMessageConsumerBuilder.class); + container.getConsumerCustomizer().customize(builder); + verify(MY_CUSTOMIZER, never()).customize(builder); + }); + } + + @Configuration(proxyBeanMethods = false) + static class WithMultipleListenersAndSingleCustomizerConfig { + + @ReactivePulsarListener(id = "multiListenerSingleCustomizer1-id", + topics = "multiListenerSingleCustomizer1-topic") + void listen1(String ignored) { + } + + @ReactivePulsarListener(id = "multiListenerSingleCustomizer2-id", + topics = "multiListenerSingleCustomizer2-topic") + void listen2(String ignored) { + } + + @Bean + ReactivePulsarListenerMessageConsumerBuilderCustomizer myCustomizer() { + return MY_CUSTOMIZER; + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleListenerAndMultipleCustomizersConfig.class) + class WithSingleListenerAndMultipleCustomizers { + + private static ReactivePulsarListenerMessageConsumerBuilderCustomizer MY_CUSTOMIZER = mock( + ReactivePulsarListenerMessageConsumerBuilderCustomizer.class); + + private static ReactivePulsarListenerMessageConsumerBuilderCustomizer MY_CUSTOMIZER2 = mock( + ReactivePulsarListenerMessageConsumerBuilderCustomizer.class); + + @Test + void customizerIsNotAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) { + assertContainer(registry, "singleListenerMultiCustomizers-id").satisfies((container) -> { + var builder = mock(ReactiveMessageConsumerBuilder.class); + container.getConsumerCustomizer().customize(builder); + verify(MY_CUSTOMIZER, never()).customize(builder); + verify(MY_CUSTOMIZER2, never()).customize(builder); + }); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleListenerAndMultipleCustomizersConfig { + + @ReactivePulsarListener(id = "singleListenerMultiCustomizers-id", + topics = "singleListenerMultiCustomizers-topic") + void listen(String ignored) { + } + + @Bean + ReactivePulsarListenerMessageConsumerBuilderCustomizer myCustomizer1() { + return MY_CUSTOMIZER; + } + + @Bean + ReactivePulsarListenerMessageConsumerBuilderCustomizer myCustomizer2() { + return MY_CUSTOMIZER2; + } + + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java index cc96bba0..2399e71f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java @@ -108,13 +108,13 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar private final AtomicInteger counter = new AtomicInteger(); + private final List> processedEndpoints = new ArrayList<>(); + @Override public void afterSingletonsInstantiated() { this.registrar.setBeanFactory(this.beanFactory); - this.beanFactory.getBeanProvider(PulsarListenerConfigurer.class) .forEach(c -> c.configurePulsarListeners(this.registrar)); - if (this.registrar.getEndpointRegistry() == null) { if (this.endpointRegistry == null) { Assert.state(this.beanFactory != null, @@ -125,12 +125,11 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar } this.registrar.setEndpointRegistry(this.endpointRegistry); } - if (this.defaultContainerFactoryBeanName != null) { this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName); } - addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService); + postProcessEndpointsBeforeRegistration(); // Actually register all listeners this.registrar.afterPropertiesSet(); } @@ -167,7 +166,6 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar Method methodToUse = checkProxy(method, bean); MethodPulsarListenerEndpoint endpoint = new MethodPulsarListenerEndpoint<>(); endpoint.setMethod(methodToUse); - String beanRef = pulsarListener.beanRef(); this.listenerScope.addListener(beanRef, bean); String[] topics = resolveTopics(pulsarListener); @@ -178,27 +176,21 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar protected void processListener(MethodPulsarListenerEndpoint endpoint, PulsarListener PulsarListener, Object bean, String beanName, String[] topics, String topicPattern) { - processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics, topicPattern); - String containerFactory = resolve(PulsarListener.containerFactory()); PulsarListenerContainerFactory listenerContainerFactory = resolveContainerFactory(PulsarListener, containerFactory, beanName); - this.registrar.registerEndpoint(endpoint, listenerContainerFactory); } @Nullable private PulsarListenerContainerFactory resolveContainerFactory(PulsarListener PulsarListener, Object factoryTarget, String beanName) { - String containerFactory = PulsarListener.containerFactory(); if (!StringUtils.hasText(containerFactory)) { return null; } - PulsarListenerContainerFactory factory = null; - Object resolved = resolveExpression(containerFactory); if (resolved instanceof PulsarListenerContainerFactory) { return (PulsarListenerContainerFactory) resolved; @@ -219,7 +211,6 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener, Object bean, String[] topics, String topicPattern) { - endpoint.setBean(bean); endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); endpoint.setSubscriptionName(getEndpointSubscriptionName(pulsarListener)); @@ -229,12 +220,10 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar resolveSubscriptionType(endpoint, pulsarListener); endpoint.setSchemaType(pulsarListener.schemaType()); endpoint.setAckMode(pulsarListener.ackMode()); - String concurrency = pulsarListener.concurrency(); if (StringUtils.hasText(concurrency)) { endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency")); } - String autoStartup = pulsarListener.autoStartup(); if (StringUtils.hasText(autoStartup)) { endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup")); @@ -242,12 +231,12 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar resolvePulsarProperties(endpoint, pulsarListener.properties()); endpoint.setBatchListener(pulsarListener.batch()); endpoint.setBeanFactory(this.beanFactory); - resolveNegativeAckRedeliveryBackoff(endpoint, pulsarListener); resolveAckTimeoutRedeliveryBackoff(endpoint, pulsarListener); resolveDeadLetterPolicy(endpoint, pulsarListener); resolvePulsarConsumerErrorHandler(endpoint, pulsarListener); resolveConsumerCustomizer(endpoint, pulsarListener); + this.processedEndpoints.add(endpoint); } private void resolveSubscriptionType(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener) { @@ -275,8 +264,27 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar } } + @SuppressWarnings("unchecked") + protected void postProcessEndpointsBeforeRegistration() { + if (this.processedEndpoints.size() == 1) { + MethodPulsarListenerEndpoint endpoint = this.processedEndpoints.get(0); + if (endpoint.getConsumerBuilderCustomizer() != null) { + return; + } + this.beanFactory.getBeanProvider(PulsarListenerConsumerBuilderCustomizer.class).ifUnique((customizer) -> { + this.logger + .info(() -> String.format("Setting the only registered PulsarListenerConsumerBuilderCustomizer " + + "on the only registered @PulsarListener (%s)", endpoint.getId())); + endpoint.setConsumerBuilderCustomizer(customizer::customize); + }); + } + } + @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveConsumerCustomizer(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener) { + if (!StringUtils.hasText(pulsarListener.consumerCustomizer())) { + return; + } Object consumerCustomizer = resolveExpression(pulsarListener.consumerCustomizer()); if (consumerCustomizer instanceof PulsarListenerConsumerBuilderCustomizer customizer) { endpoint.setConsumerBuilderCustomizer(customizer::customize); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java index 20baba6c..37bbfa37 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java @@ -100,13 +100,13 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn private final AtomicInteger counter = new AtomicInteger(); + private final List> processedEndpoints = new ArrayList<>(); + @Override public void afterSingletonsInstantiated() { this.registrar.setBeanFactory(this.beanFactory); - this.beanFactory.getBeanProvider(PulsarReaderConfigurer.class) .forEach(c -> c.configurePulsarReaders(this.registrar)); - if (this.registrar.getEndpointRegistry() == null) { if (this.endpointRegistry == null) { Assert.state(this.beanFactory != null, @@ -117,11 +117,10 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn } this.registrar.setEndpointRegistry(this.endpointRegistry); } - if (this.defaultContainerFactoryBeanName != null) { this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName); } - + postProcessEndpointsBeforeRegistration(); // Register all readers this.registrar.afterPropertiesSet(); } @@ -228,10 +227,29 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn endpoint.setBeanFactory(this.beanFactory); resolveReaderCustomizer(endpoint, pulsarReader); + this.processedEndpoints.add(endpoint); + } + + @SuppressWarnings("unchecked") + protected void postProcessEndpointsBeforeRegistration() { + if (this.processedEndpoints.size() == 1) { + MethodPulsarReaderEndpoint endpoint = this.processedEndpoints.get(0); + if (endpoint.getReaderBuilderCustomizer() != null) { + return; + } + this.beanFactory.getBeanProvider(PulsarReaderReaderBuilderCustomizer.class).ifUnique((customizer) -> { + this.logger.info(() -> String.format("Setting the only registered PulsarReaderReaderBuilderCustomizer " + + "on the only registered @PulsarReader (%s)", endpoint.getId())); + endpoint.setReaderBuilderCustomizer(customizer::customize); + }); + } } @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveReaderCustomizer(MethodPulsarReaderEndpoint endpoint, PulsarReader pulsarReader) { + if (!StringUtils.hasText(pulsarReader.readerCustomizer())) { + return; + } Object readerCustomizer = resolveExpression(pulsarReader.readerCustomizer()); if (readerCustomizer instanceof PulsarReaderReaderBuilderCustomizer customizer) { endpoint.setReaderBuilderCustomizer(customizer::customize); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index ef1fe268..8e4dde68 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -247,6 +247,10 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo this.ackTimeoutRedeliveryBackoff = ackTimeoutRedeliveryBackoff; } + public ConsumerBuilderCustomizer getConsumerBuilderCustomizer() { + return this.consumerBuilderCustomizer; + } + public void setConsumerBuilderCustomizer(ConsumerBuilderCustomizer consumerBuilderCustomizer) { this.consumerBuilderCustomizer = consumerBuilderCustomizer; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java index cc62e1d6..258af324 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java @@ -199,6 +199,10 @@ public class MethodPulsarReaderEndpoint extends AbstractPulsarReaderEndpoint< this.messageHandlerMethodFactory = messageHandlerMethodFactory; } + public ReaderBuilderCustomizer getReaderBuilderCustomizer() { + return this.readerBuilderCustomizer; + } + public void setReaderBuilderCustomizer(ReaderBuilderCustomizer readerBuilderCustomizer) { this.readerBuilderCustomizer = readerBuilderCustomizer; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerCustomizerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerCustomizerTests.java new file mode 100644 index 00000000..01011086 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerCustomizerTests.java @@ -0,0 +1,194 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.ConsumerBuilder; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.assertj.core.api.ObjectAssert; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustomizer; +import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; +import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithCustomizerOnListener.WithCustomizerOnListenerConfig; +import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithMultipleListenersAndSingleCustomizer.WithMultipleListenersAndSingleCustomizerConfig; +import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithSingleListenerAndMultipleCustomizers.WithSingleListenerAndMultipleCustomizersConfig; +import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithSingleListenerAndSingleCustomizer.WithSingleListenerAndSingleCustomizerConfig; +import org.springframework.test.context.ContextConfiguration; + +/** + * Tests setting the consumer customizer on the {@link PulsarListener @PulsarListener}. + * + * @author Chris Bono + */ +@SuppressWarnings({ "unchecked", "rawtypes" }) +class PulsarListenerCustomizerTests extends PulsarListenerTestsBase { + + private ObjectAssert assertContainer( + PulsarListenerEndpointRegistry registry, String containerId) { + return assertThat(registry.getListenerContainer(containerId)).isNotNull() + .isInstanceOf(AbstractPulsarMessageListenerContainer.class) + .asInstanceOf(InstanceOfAssertFactories.type(AbstractPulsarMessageListenerContainer.class)); + } + + @Nested + @ContextConfiguration(classes = WithCustomizerOnListenerConfig.class) + class WithCustomizerOnListener { + + private static final CountDownLatch latch = new CountDownLatch(1); + + @Test + void overridesDefaultCustomizer() throws Exception { + pulsarTemplate.send("overrides-default-topic", "hello"); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithCustomizerOnListenerConfig { + + @PulsarListener(topics = "overrides-default-topic", consumerCustomizer = "myCustomizer") + void listen(String ignored, Consumer consumer) { + assertThat(consumer.getConsumerName()).isEqualTo("fromMyCustomizer"); + latch.countDown(); + } + + @Bean + ConsumerBuilderCustomizer defaultCustomizer() { + return (cb) -> cb.consumerName("fromDefaultCustomizer"); + } + + @Bean + PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return (cb) -> cb.consumerName("fromMyCustomizer"); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleListenerAndSingleCustomizerConfig.class) + class WithSingleListenerAndSingleCustomizer { + + private static PulsarListenerConsumerBuilderCustomizer MY_CUSTOMIZER = mock( + PulsarListenerConsumerBuilderCustomizer.class); + + @Test + void customizerIsAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) { + assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> { + var builder = mock(ConsumerBuilder.class); + container.getConsumerBuilderCustomizer().customize(builder); + verify(MY_CUSTOMIZER).customize(builder); + }); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleListenerAndSingleCustomizerConfig { + + @PulsarListener(id = "singleListenerSingleCustomizer-id", topics = "singleListenerSingleCustomizer-topic") + void listen(String ignored) { + } + + @Bean + PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return MY_CUSTOMIZER; + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithMultipleListenersAndSingleCustomizerConfig.class) + class WithMultipleListenersAndSingleCustomizer { + + @Test + void customizerIsNotAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) { + assertContainer(registry, "multiListenerSingleCustomizer1-id") + .extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer) + .isNull(); + assertContainer(registry, "multiListenerSingleCustomizer2-id") + .extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer) + .isNull(); + } + + @Configuration(proxyBeanMethods = false) + static class WithMultipleListenersAndSingleCustomizerConfig { + + @PulsarListener(id = "multiListenerSingleCustomizer1-id", topics = "multiListenerSingleCustomizer1-topic") + void listen1(String ignored) { + } + + @PulsarListener(id = "multiListenerSingleCustomizer2-id", topics = "multiListenerSingleCustomizer2-topic") + void listen2(String ignored) { + } + + @Bean + PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return mock(PulsarListenerConsumerBuilderCustomizer.class); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleListenerAndMultipleCustomizersConfig.class) + class WithSingleListenerAndMultipleCustomizers { + + @Test + void customizerIsNotAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) { + assertContainer(registry, "singleListenerMultiCustomizers-id") + .extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer) + .isNull(); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleListenerAndMultipleCustomizersConfig { + + @PulsarListener(id = "singleListenerMultiCustomizers-id", topics = "singleListenerMultiCustomizers-topic") + void listen(String ignored) { + } + + @Bean + PulsarListenerConsumerBuilderCustomizer myCustomizer1() { + return mock(PulsarListenerConsumerBuilderCustomizer.class); + } + + @Bean + PulsarListenerConsumerBuilderCustomizer myCustomizer2() { + return mock(PulsarListenerConsumerBuilderCustomizer.class); + } + + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index d63fe83e..6753ed67 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -34,8 +34,6 @@ import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Messages; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; @@ -53,7 +51,6 @@ import org.awaitility.Awaitility; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; @@ -66,25 +63,17 @@ import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactor import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.ConsumerBuilderCustomizer; -import org.springframework.pulsar.core.DefaultPulsarClientFactory; -import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.DefaultTopicResolver; -import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; -import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; -import org.springframework.pulsar.core.PulsarTopic; import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.core.TopicResolver; import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig; import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig; import org.springframework.pulsar.support.PulsarHeaders; -import org.springframework.pulsar.test.support.PulsarTestContainerSupport; -import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.backoff.FixedBackOff; /** @@ -92,60 +81,7 @@ import org.springframework.util.backoff.FixedBackOff; * @author Alexander Preuß * @author Chris Bono */ -@SpringJUnitConfig -@DirtiesContext -public class PulsarListenerTests implements PulsarTestContainerSupport { - - @Autowired - PulsarTemplate pulsarTemplate; - - @Autowired - private PulsarClient pulsarClient; - - @Configuration(proxyBeanMethods = false) - @EnablePulsar - public static class TopLevelConfig { - - @Bean - public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1"); - } - - @Bean - public PulsarClient pulsarClient() throws PulsarClientException { - return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient(); - } - - @Bean - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - return new PulsarTemplate<>(pulsarProducerFactory); - } - - @Bean - public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, - ObjectProvider> defaultConsumerCustomizersProvider) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, - defaultConsumerCustomizersProvider.orderedStream().toList()); - } - - @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( - PulsarConsumerFactory pulsarConsumerFactory) { - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, - new PulsarContainerProperties()); - } - - @Bean - PulsarAdministration pulsarAdministration() { - return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); - } - - @Bean - PulsarTopic partitionedTopic() { - return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); - } - - } +class PulsarListenerTests extends PulsarListenerTestsBase { @Nested @ContextConfiguration(classes = PulsarListenerBasicTestCases.TestPulsarListenersForBasicScenario.class) @@ -1080,38 +1016,6 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } - @Nested - @ContextConfiguration(classes = PulsarListenerCustomizerTests.WithCustomizerConfig.class) - class PulsarListenerCustomizerTests { - - private static final CountDownLatch latch = new CountDownLatch(1); - - @Test - void withCustomizerOverridingSubscriptionName() throws Exception { - pulsarTemplate.send("with-customizer-listener-topic", "hello"); - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - } - - @EnablePulsar - @Configuration - static class WithCustomizerConfig { - - @PulsarListener(id = "with-customizer-listener", subscriptionName = "with-customizer-listener-subscription", - topics = "with-customizer-listener-topic", consumerCustomizer = "myCustomizer") - void listen(String ignored, Consumer consumer) { - assertThat(consumer.getSubscription()).isEqualTo("test-changed-subscription-name"); - latch.countDown(); - } - - @Bean - public PulsarListenerConsumerBuilderCustomizer myCustomizer() { - return cb -> cb.subscriptionName("test-changed-subscription-name"); - } - - } - - } - @Nested class SubscriptionTypeTests { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java new file mode 100644 index 00000000..7cdd0692 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java @@ -0,0 +1,102 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.EnablePulsar; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; +import org.springframework.pulsar.config.PulsarListenerContainerFactory; +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; +import org.springframework.pulsar.core.DefaultPulsarClientFactory; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.pulsar.core.PulsarProducerFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Soby Chacko + * @author Alexander Preuß + * @author Chris Bono + */ +@SpringJUnitConfig +@DirtiesContext +abstract class PulsarListenerTestsBase implements PulsarTestContainerSupport { + + @Autowired + protected PulsarTemplate pulsarTemplate; + + @Autowired + protected PulsarClient pulsarClient; + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + static class TopLevelConfig { + + @Bean + PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { + return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1"); + } + + @Bean + PulsarClient pulsarClient() throws PulsarClientException { + return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient(); + } + + @Bean + PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + return new PulsarTemplate<>(pulsarProducerFactory); + } + + @Bean + public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, + ObjectProvider> defaultConsumerCustomizersProvider) { + return new DefaultPulsarConsumerFactory<>(pulsarClient, + defaultConsumerCustomizersProvider.orderedStream().toList()); + } + + @Bean + PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarConsumerFactory pulsarConsumerFactory) { + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, + new PulsarContainerProperties()); + } + + @Bean + PulsarAdministration pulsarAdministration() { + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); + } + + @Bean + PulsarTopic partitionedTopic() { + return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderCustomizerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderCustomizerTests.java new file mode 100644 index 00000000..68f35af9 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderCustomizerTests.java @@ -0,0 +1,196 @@ +/* + * Copyright 2022-2023 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.reader; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.pulsar.client.api.ReaderBuilder; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.assertj.core.api.ObjectAssert; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.PulsarReader; +import org.springframework.pulsar.annotation.PulsarReaderReaderBuilderCustomizer; +import org.springframework.pulsar.config.PulsarReaderEndpointRegistry; +import org.springframework.pulsar.core.ReaderBuilderCustomizer; +import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithCustomizerOnReader.WithCustomizerOnReaderConfig; +import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithMultipleReadersAndSingleCustomizer.WithMultipleReadersAndSingleCustomizerConfig; +import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithSingleReaderAndMultipleCustomizers.WithSingleReaderAndMultipleCustomizersConfig; +import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithSingleReaderAndSingleCustomizer.WithSingleReaderAndSingleCustomizerConfig; +import org.springframework.test.context.ContextConfiguration; + +/** + * Tests setting the consumer customizer on the {@link PulsarReader @PulsarReader}. + * + * @author Chris Bono + */ +@SuppressWarnings({ "unchecked", "rawtypes" }) +class PulsarReaderCustomizerTests extends PulsarReaderTestsBase { + + private ObjectAssert assertContainer(PulsarReaderEndpointRegistry registry, + String containerId) { + return assertThat(registry.getReaderContainer(containerId)).isNotNull() + .isInstanceOf(AbstractPulsarMessageReaderContainer.class) + .asInstanceOf(InstanceOfAssertFactories.type(AbstractPulsarMessageReaderContainer.class)); + } + + @Nested + @ContextConfiguration(classes = WithCustomizerOnReaderConfig.class) + class WithCustomizerOnReader { + + private static final CountDownLatch latch = new CountDownLatch(1); + + @Test + void overridesDefaultCustomizer() throws Exception { + pulsarTemplate.send("myCustomizerTopic", "hello"); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithCustomizerOnReaderConfig { + + @PulsarReader(id = "overrides-default-id", readerCustomizer = "myCustomizer", startMessageId = "earliest") + void listen(String ignored) { + latch.countDown(); + } + + @Bean + ReaderBuilderCustomizer defaultCustomizer() { + return (rb) -> rb.topic("defaultCustomizerTopic"); + } + + @Bean + PulsarReaderReaderBuilderCustomizer myCustomizer() { + return (rb) -> rb.topic("myCustomizerTopic"); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleReaderAndSingleCustomizerConfig.class) + class WithSingleReaderAndSingleCustomizer { + + private static PulsarReaderReaderBuilderCustomizer MY_CUSTOMIZER = mock( + PulsarReaderReaderBuilderCustomizer.class); + + @Test + void customizerIsAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) { + assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> { + var builder = mock(ReaderBuilder.class); + container.getReaderBuilderCustomizer().customize(builder); + verify(MY_CUSTOMIZER).customize(builder); + }); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleReaderAndSingleCustomizerConfig { + + @PulsarReader(id = "singleListenerSingleCustomizer-id", topics = "singleListenerSingleCustomizer-topic", + startMessageId = "earliest") + void listen(String ignored) { + } + + @Bean + PulsarReaderReaderBuilderCustomizer myCustomizer() { + return MY_CUSTOMIZER; + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithMultipleReadersAndSingleCustomizerConfig.class) + class WithMultipleReadersAndSingleCustomizer { + + @Test + void customizerIsNotAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) { + assertContainer(registry, "multiReaderSingleCustomizer1-id") + .extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer) + .isNull(); + assertContainer(registry, "multiReaderSingleCustomizer2-id") + .extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer) + .isNull(); + } + + @Configuration(proxyBeanMethods = false) + static class WithMultipleReadersAndSingleCustomizerConfig { + + @PulsarReader(id = "multiReaderSingleCustomizer1-id", topics = "multiReaderSingleCustomizer1-topic", + startMessageId = "earliest") + void listen1(String ignored) { + } + + @PulsarReader(id = "multiReaderSingleCustomizer2-id", topics = "multiReaderSingleCustomizer2-topic", + startMessageId = "earliest") + void listen2(String ignored) { + } + + @Bean + PulsarReaderReaderBuilderCustomizer myCustomizer() { + return mock(PulsarReaderReaderBuilderCustomizer.class); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSingleReaderAndMultipleCustomizersConfig.class) + class WithSingleReaderAndMultipleCustomizers { + + @Test + void customizerIsNotAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) { + assertContainer(registry, "singleReaderMultiCustomizers-id") + .extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer) + .isNull(); + } + + @Configuration(proxyBeanMethods = false) + static class WithSingleReaderAndMultipleCustomizersConfig { + + @PulsarReader(id = "singleReaderMultiCustomizers-id", topics = "singleReaderMultiCustomizers-topic", + startMessageId = "earliest") + void listen(String ignored) { + } + + @Bean + PulsarReaderReaderBuilderCustomizer myCustomizer1() { + return mock(PulsarReaderReaderBuilderCustomizer.class); + } + + @Bean + PulsarReaderReaderBuilderCustomizer myCustomizer2() { + return mock(PulsarReaderReaderBuilderCustomizer.class); + } + + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java index 90c04197..78a69b80 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java @@ -24,30 +24,18 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarReader; import org.springframework.pulsar.annotation.PulsarReaderReaderBuilderCustomizer; -import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory; -import org.springframework.pulsar.config.PulsarReaderContainerFactory; -import org.springframework.pulsar.core.DefaultPulsarClientFactory; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.DefaultPulsarReaderFactory; -import org.springframework.pulsar.core.PulsarProducerFactory; -import org.springframework.pulsar.core.PulsarReaderFactory; import org.springframework.pulsar.core.PulsarTemplate; -import org.springframework.pulsar.test.support.PulsarTestContainerSupport; -import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * Tests for {@link PulsarReader}. @@ -55,47 +43,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * @author Soby Chacko * @author Chris Bono */ -@SpringJUnitConfig -@DirtiesContext -public class PulsarReaderTests implements PulsarTestContainerSupport { - - @Autowired - PulsarTemplate pulsarTemplate; - - @Autowired - private PulsarClient pulsarClient; - - @Configuration(proxyBeanMethods = false) - @EnablePulsar - public static class TopLevelConfig { - - @Bean - public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient); - } - - @Bean - public PulsarClient pulsarClient() throws PulsarClientException { - return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient(); - } - - @Bean - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - return new PulsarTemplate<>(pulsarProducerFactory); - } - - @Bean - public PulsarReaderFactory pulsarReaderFactory(PulsarClient pulsarClient) { - return new DefaultPulsarReaderFactory<>(pulsarClient); - } - - @Bean - PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory pulsarReaderFactory) { - return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, - new PulsarReaderContainerProperties()); - } - - } +public class PulsarReaderTests extends PulsarReaderTestsBase { @Nested @ContextConfiguration(classes = StartMessageIdEarliest.PulsarReaderStartMessageIdEarliest.class) @@ -109,8 +57,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar - @Configuration + @Configuration(proxyBeanMethods = false) static class PulsarReaderStartMessageIdEarliest { @PulsarReader(id = "pulsarReaderBasicScenario-id-1", topics = "pulsarReaderBasicScenario-topic-1", @@ -162,8 +109,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar - @Configuration + @Configuration(proxyBeanMethods = false) static class PulsarReaderStartMessageIdLatest { @PulsarReader(id = "pulsarReaderBasicScenario-id-3", topics = "pulsarReaderBasicScenario-topic-3", @@ -185,11 +131,10 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @Test void startMessageIdProvidedThroughReaderCustomizer() throws Exception { - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar - @Configuration + @Configuration(proxyBeanMethods = false) static class WithCustomizerConfig { int currentIndex = 5; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTestsBase.java new file mode 100644 index 00000000..3cd32efe --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTestsBase.java @@ -0,0 +1,87 @@ +/* + * Copyright 2023 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.reader; + +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.EnablePulsar; +import org.springframework.pulsar.annotation.PulsarReader; +import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory; +import org.springframework.pulsar.config.PulsarReaderContainerFactory; +import org.springframework.pulsar.core.DefaultPulsarClientFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultPulsarReaderFactory; +import org.springframework.pulsar.core.PulsarProducerFactory; +import org.springframework.pulsar.core.PulsarReaderFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * Tests for {@link PulsarReader}. + * + * @author Soby Chacko + * @author Chris Bono + */ +@SpringJUnitConfig +@DirtiesContext +class PulsarReaderTestsBase implements PulsarTestContainerSupport { + + @Autowired + protected PulsarTemplate pulsarTemplate; + + @Autowired + protected PulsarClient pulsarClient; + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + static class TopLevelConfig { + + @Bean + PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { + return new DefaultPulsarProducerFactory<>(pulsarClient); + } + + @Bean + PulsarClient pulsarClient() throws PulsarClientException { + return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient(); + } + + @Bean + PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + return new PulsarTemplate<>(pulsarProducerFactory); + } + + @Bean + PulsarReaderFactory pulsarReaderFactory(PulsarClient pulsarClient) { + return new DefaultPulsarReaderFactory<>(pulsarClient); + } + + @Bean + PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory pulsarReaderFactory) { + return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, + new PulsarReaderContainerProperties()); + } + + } + +}