From 9de215bc66f1c211121fd82b4d5722d2256ca04d Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 13 Nov 2023 19:42:19 -0600 Subject: [PATCH] Add specialized builder customizers for listener annotations (#487) Introduces specializations of the consumer builders that can be used by the annotation listeners. This differentiates between the existing consumer builders that are picked up by Spring Boot autoconfiguration and used to customizer the consumer factories globally. * Add PulsarListenerConsumerBuilderCustomizer * Add PulsarReaderReaderBuilderCustomizer * Add ReactivePulsarListenerMessageConsumerBuilderCustomizer Resolves #486 --- .../modules/ROOT/pages/reference/pulsar.adoc | 10 ++--- .../ROOT/pages/reference/reactive-pulsar.adoc | 6 +-- ...arListenerAnnotationBeanPostProcessor.java | 18 ++++---- ...tenerMessageConsumerBuilderCustomizer.java | 43 ++++++++++++++++++ ...ctiveMessageConsumerBuilderCustomizer.java | 14 +++++- .../listener/ReactivePulsarListenerTests.java | 12 ++--- .../pulsar/annotation/PulsarListener.java | 7 +-- ...arListenerAnnotationBeanPostProcessor.java | 16 +++---- ...lsarListenerConsumerBuilderCustomizer.java | 45 +++++++++++++++++++ ...lsarReaderAnnotationBeanPostProcessor.java | 17 ++++--- .../PulsarReaderReaderBuilderCustomizer.java | 45 +++++++++++++++++++ .../core/ConsumerBuilderCustomizer.java | 13 +++++- .../pulsar/core/ReaderBuilderCustomizer.java | 11 ++++- .../pulsar/listener/PulsarListenerTests.java | 4 +- .../pulsar/reader/PulsarReaderTests.java | 4 +- 15 files changed, 213 insertions(+), 52 deletions(-) create mode 100644 spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerMessageConsumerBuilderCustomizer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConsumerBuilderCustomizer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderReaderBuilderCustomizer.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 aa060a93..2d294c05 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 @@ -1029,12 +1029,12 @@ At that point, the message is acknowledged by the container, and the listener is === Consumer Customization on PulsarListener Spring for Apache Pulsar provides a convenient way to customize the consumer created by the container used by the `PulsarListener`. -Applications can provide a bean for `ConsumerBuilderCustomizer`. +Applications can provide a bean for `PulsarListenerConsumerBuilderCustomizer`. Here is an example. [source, java] ---- @Bean -public ConsumerBuilderCustomizer myCustomizer() { +public PulsarListenerConsumerBuilderCustomizer myCustomizer() { return cb -> { cb.subscriptionName("modified-subscription-name"); }; @@ -1107,8 +1107,8 @@ 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 `ReaderBuilderCustomizer` in Spring for Apache Pulsar. -You can provide a `@Bean` from `ReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below. +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. [source, java] ---- @@ -1119,7 +1119,7 @@ void read(String message) { } @Bean -public ReaderBuilderCustomizer myCustomizer() { +public PulsarReaderReaderBuilderCustomizer myCustomizer() { return readerBuilder -> { readerBuilder.startMessageId(messageId); // the first message read is after this message id. // Any other customizations on the readerBuilder 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 773cbef1..878e3ae2 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 @@ -245,7 +245,7 @@ Spring Boot provides this consumer factory which can be configured with any of t [[reactive-consumer-customizer]] ==== Consumer Customization -You can specify a `ReactiveMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages. +You can specify a `ReactivePulsarListenerMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages. WARNING: Use with caution as this gives full access to the consumer builder and invoking some of its methods (such as `create`) may have unintended side effects. @@ -260,7 +260,7 @@ Mono listen(String message) { } @Bean -ReactiveMessageConsumerBuilderCustomizer myConsumerCustomizer() { +ReactivePulsarListenerMessageConsumerBuilderCustomizer myConsumerCustomizer() { return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } ---- @@ -273,7 +273,7 @@ The following customizer example uses direct Pulsar consumer properties: [source, java] ---- @Bean -ReactiveMessageConsumerBuilderCustomizer directConsumerPropsCustomizer() { +ReactivePulsarListenerMessageConsumerBuilderCustomizer directConsumerPropsCustomizer() { return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1"); } ---- 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 152d89ac..b4a7ad2b 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 @@ -53,7 +53,6 @@ import org.springframework.pulsar.reactive.config.MethodReactivePulsarListenerEn import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpoint; import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; -import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -277,19 +276,20 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra } } - @SuppressWarnings("unchecked") + @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveConsumerCustomizer(MethodReactivePulsarListenerEndpoint endpoint, ReactivePulsarListener reactivePulsarListener) { - Object customizer = resolveExpression(reactivePulsarListener.consumerCustomizer()); - if (customizer instanceof ReactiveMessageConsumerBuilderCustomizer) { - endpoint.setConsumerCustomizer((ReactiveMessageConsumerBuilderCustomizer) customizer); + Object consumerCustomizer = resolveExpression(reactivePulsarListener.consumerCustomizer()); + if (consumerCustomizer instanceof ReactivePulsarListenerMessageConsumerBuilderCustomizer customizer) { + endpoint.setConsumerCustomizer(customizer::customize); } else { - String consumerCustomizerBeanName = resolveExpressionAsString(reactivePulsarListener.consumerCustomizer(), + String customizerBeanName = resolveExpressionAsString(reactivePulsarListener.consumerCustomizer(), "consumerCustomizer"); - if (StringUtils.hasText(consumerCustomizerBeanName)) { - endpoint.setConsumerCustomizer(this.beanFactory.getBean(consumerCustomizerBeanName, - ReactiveMessageConsumerBuilderCustomizer.class)); + if (StringUtils.hasText(customizerBeanName)) { + var customizer = this.beanFactory.getBean(customizerBeanName, + ReactivePulsarListenerMessageConsumerBuilderCustomizer.class); + endpoint.setConsumerCustomizer(customizer::customize); } } } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerMessageConsumerBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerMessageConsumerBuilderCustomizer.java new file mode 100644 index 00000000..c26102c0 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerMessageConsumerBuilderCustomizer.java @@ -0,0 +1,43 @@ +/* + * 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.config.annotation; + +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; + +/** + * Callback interface that can be implemented by a bean to customize the + * {@link ReactiveMessageConsumerBuilder builder} that is used to create the underlying + * Pulsar reactive message consumer used by a {@link ReactivePulsarListener} to receive + * messages. + *

+ * Unlike the {@link ReactiveMessageConsumerBuilder} which is applied to all created + * reactive message consumer builders, this customizer is only applied to the individual + * consumer builder(s) of the {@code @ReactivePulsarListener(s)} it is associated with. + * + * @param The message payload type + * @author Chris Bono + */ +@FunctionalInterface +public interface ReactivePulsarListenerMessageConsumerBuilderCustomizer { + + /** + * Customize the {@link ReactiveMessageConsumerBuilder}. + * @param reactiveMessageConsumerBuilder the builder to customize + */ + void customize(ReactiveMessageConsumerBuilder reactiveMessageConsumerBuilder); + +} diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java index b18b2c19..9635c0c0 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java @@ -19,7 +19,17 @@ package org.springframework.pulsar.reactive.core; import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; /** - * The interface to customize a {@link ReactiveMessageConsumerBuilder}. + * Callback interface that can be implemented to customize the + * {@link ReactiveMessageConsumerBuilder builder} that is used by the + * {@link ReactivePulsarConsumerFactory} to create consumers. + *

+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be + * used as default configuration by the {@link DefaultReactivePulsarConsumerFactory} and + * as such will apply to all created consumers. + *

+ * The consumer factory also supports passing in a specific instance of this callback when + * {@link ReactivePulsarConsumerFactory#createConsumer creating a consumer} and as such + * the passed in customizer only applies to the single created consumer. * * @param The message payload type * @author Christophe Bornet @@ -28,7 +38,7 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; public interface ReactiveMessageConsumerBuilderCustomizer { /** - * Customizes a {@link ReactiveMessageConsumerBuilder}. + * Customize the {@link ReactiveMessageConsumerBuilder}. * @param reactiveMessageConsumerBuilder the builder to customize */ void customize(ReactiveMessageConsumerBuilder reactiveMessageConsumerBuilder); diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index 92ec4f58..854667bb 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java @@ -69,8 +69,8 @@ import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContaine 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.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2; import org.springframework.pulsar.support.PulsarHeaders; @@ -145,7 +145,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Bean - ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + ReactivePulsarListenerMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } @@ -207,7 +207,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Bean - ReactiveMessageConsumerBuilderCustomizer listen1Customizer() { + ReactivePulsarListenerMessageConsumerBuilderCustomizer listen1Customizer() { return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } @@ -218,7 +218,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Bean - ReactiveMessageConsumerBuilderCustomizer listen2Customizer() { + ReactivePulsarListenerMessageConsumerBuilderCustomizer listen2Customizer() { return b -> b.topics(List.of("topic-2")) .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } @@ -231,7 +231,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Bean - ReactiveMessageConsumerBuilderCustomizer listen3Customizer() { + ReactivePulsarListenerMessageConsumerBuilderCustomizer listen3Customizer() { return b -> b.topicsPatternAutoDiscoveryPeriod(Duration.ofSeconds(5)) .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } @@ -331,7 +331,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Bean - ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + ReactivePulsarListenerMessageConsumerBuilderCustomizer consumerCustomizer() { return b -> b.negativeAckRedeliveryDelay(Duration.ofSeconds(1)) .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java index fc741484..8435cd65 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java @@ -28,7 +28,6 @@ import org.apache.pulsar.common.schema.SchemaType; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; -import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.listener.AckMode; /** @@ -210,8 +209,10 @@ public @interface PulsarListener { /** * The bean name or a 'SpEL' expression that resolves to a - * {@link ConsumerBuilderCustomizer} to use to configure the consumer. - * @return the bean name or empty string to not configure the consumer. + * {@link PulsarListenerConsumerBuilderCustomizer} to use to configure the underlying + * consumer. + * @return the bean name or SpEL expression to the customizer or an empty string to + * not customize the consumer */ String consumerCustomizer() default ""; 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 8accd0ab..b271b1af 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 @@ -51,7 +51,6 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpoint; import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; -import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.listener.PulsarConsumerErrorHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -268,18 +267,19 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar } } - @SuppressWarnings({ "rawtypes" }) + @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveConsumerCustomizer(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener) { Object consumerCustomizer = resolveExpression(pulsarListener.consumerCustomizer()); - if (consumerCustomizer instanceof ConsumerBuilderCustomizer) { - endpoint.setConsumerBuilderCustomizer((ConsumerBuilderCustomizer) consumerCustomizer); + if (consumerCustomizer instanceof PulsarListenerConsumerBuilderCustomizer customizer) { + endpoint.setConsumerBuilderCustomizer(customizer::customize); } else { - String consumerCustomizerBeanName = resolveExpressionAsString(pulsarListener.consumerCustomizer(), + String customizerBeanName = resolveExpressionAsString(pulsarListener.consumerCustomizer(), "consumerCustomizer"); - if (StringUtils.hasText(consumerCustomizerBeanName)) { - endpoint.setConsumerBuilderCustomizer( - this.beanFactory.getBean(consumerCustomizerBeanName, ConsumerBuilderCustomizer.class)); + if (StringUtils.hasText(customizerBeanName)) { + var customizer = this.beanFactory.getBean(customizerBeanName, + PulsarListenerConsumerBuilderCustomizer.class); + endpoint.setConsumerBuilderCustomizer(customizer::customize); } } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConsumerBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConsumerBuilderCustomizer.java new file mode 100644 index 00000000..5dd2e1c1 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConsumerBuilderCustomizer.java @@ -0,0 +1,45 @@ +/* + * 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.annotation; + +import org.apache.pulsar.client.api.ConsumerBuilder; + +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; + +/** + * Callback interface that can be implemented by a bean to customize the + * {@link ConsumerBuilder} that is used to create the underlying Pulsar consumer used by a + * {@link PulsarListener} to receive messages. + * + *

+ * Unlike the {@link ConsumerBuilderCustomizer} which is applied to all created consumer + * builders, this customizer is only applied to the individual consumer builder(s) of the + * {@code @PulsarListener(s)} it is associated with. + * + * @param The message payload type + * @author Chris Bono + */ +@FunctionalInterface +public interface PulsarListenerConsumerBuilderCustomizer { + + /** + * Customize the {@link ConsumerBuilder}. + * @param consumerBuilder the builder to customize + */ + void customize(ConsumerBuilder consumerBuilder); + +} 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 5d3ca48e..20baba6c 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 @@ -44,8 +44,6 @@ import org.springframework.pulsar.config.PulsarReaderContainerFactory; import org.springframework.pulsar.config.PulsarReaderEndpoint; import org.springframework.pulsar.config.PulsarReaderEndpointRegistrar; import org.springframework.pulsar.config.PulsarReaderEndpointRegistry; -import org.springframework.pulsar.core.ConsumerBuilderCustomizer; -import org.springframework.pulsar.core.ReaderBuilderCustomizer; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -232,17 +230,18 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn resolveReaderCustomizer(endpoint, pulsarReader); } + @SuppressWarnings({ "rawtypes", "unchecked" }) private void resolveReaderCustomizer(MethodPulsarReaderEndpoint endpoint, PulsarReader pulsarReader) { Object readerCustomizer = resolveExpression(pulsarReader.readerCustomizer()); - if (readerCustomizer instanceof ConsumerBuilderCustomizer) { - endpoint.setReaderBuilderCustomizer((ReaderBuilderCustomizer) readerCustomizer); + if (readerCustomizer instanceof PulsarReaderReaderBuilderCustomizer customizer) { + endpoint.setReaderBuilderCustomizer(customizer::customize); } else { - String readerCustomizerBeanName = resolveExpressionAsString(pulsarReader.readerCustomizer(), - "readerCustomizer"); - if (StringUtils.hasText(readerCustomizerBeanName)) { - endpoint.setReaderBuilderCustomizer( - this.beanFactory.getBean(readerCustomizerBeanName, ReaderBuilderCustomizer.class)); + String customizerBeanName = resolveExpressionAsString(pulsarReader.readerCustomizer(), "readerCustomizer"); + if (StringUtils.hasText(customizerBeanName)) { + var customizer = this.beanFactory.getBean(customizerBeanName, + PulsarReaderReaderBuilderCustomizer.class); + endpoint.setReaderBuilderCustomizer(customizer::customize); } } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderReaderBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderReaderBuilderCustomizer.java new file mode 100644 index 00000000..f3e8682b --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderReaderBuilderCustomizer.java @@ -0,0 +1,45 @@ +/* + * 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.annotation; + +import org.apache.pulsar.client.api.ReaderBuilder; + +import org.springframework.pulsar.core.ReaderBuilderCustomizer; + +/** + * Callback interface that can be implemented by a bean to customize the + * {@link ReaderBuilder} that is used to create the underlying Pulsar reader used by a + * {@link PulsarReader @PulsarReader} to receive messages. + * + *

+ * Unlike the {@link ReaderBuilderCustomizer} which is applied to all created reader + * builders, this customizer is only applied to the individual reader builder(s) of the + * {@code @PulsarReader(s)} it is associated with. + * + * @param The message payload type + * @author Chris Bono + */ +@FunctionalInterface +public interface PulsarReaderReaderBuilderCustomizer { + + /** + * Customize the {@link ReaderBuilder}. + * @param readerBuilder the builder to customize + */ + void customize(ReaderBuilder readerBuilder); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java index d8f004a4..28af223e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java @@ -19,7 +19,16 @@ package org.springframework.pulsar.core; import org.apache.pulsar.client.api.ConsumerBuilder; /** - * The interface to customize a {@link ConsumerBuilder}. + * Callback interface that can be implemented to customize the {@link ConsumerBuilder} + * that is used by the {@link PulsarConsumerFactory} to create consumers. + *

+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be + * used as default configuration by the {@link DefaultPulsarConsumerFactory} and as such + * will apply to all created consumers. + *

+ * The consumer factory also supports passing in a specific instance of this callback when + * {@link PulsarConsumerFactory#createConsumer creating a consumer} and as such the passed + * in customizer only applies to the single created consumer. * * @param The message payload type * @author Christophe Bornet @@ -28,7 +37,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder; public interface ConsumerBuilderCustomizer { /** - * Customizes a {@link ConsumerBuilder}. + * Customize the {@link ConsumerBuilder}. * @param consumerBuilder the builder to customize */ void customize(ConsumerBuilder consumerBuilder); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ReaderBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ReaderBuilderCustomizer.java index 6d7db0e2..bf713803 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ReaderBuilderCustomizer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ReaderBuilderCustomizer.java @@ -19,7 +19,16 @@ package org.springframework.pulsar.core; import org.apache.pulsar.client.api.ReaderBuilder; /** - * The interface to customize a {@link ReaderBuilder}. + * Callback interface that can be implemented to customize the {@link ReaderBuilder} that + * is used by the {@link PulsarReaderFactory} to create readers. + *

+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be + * used as default configuration by the {@link DefaultPulsarReaderFactory} and as such + * will apply to all created readers. + *

+ * The reader factory also supports passing in a specific instance of this callback when + * {@link PulsarReaderFactory#createReader creating a reader} and as such the passed in + * customizer only applies to the single created reader. * * @param The message payload type * @author Soby Chacko 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 893e3725..80cb9c12 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 @@ -57,10 +57,10 @@ import org.springframework.context.annotation.Configuration; import org.springframework.messaging.handler.annotation.Header; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustomizer; import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; 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; @@ -1095,7 +1095,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } @Bean - public ConsumerBuilderCustomizer myCustomizer() { + public PulsarListenerConsumerBuilderCustomizer myCustomizer() { return cb -> cb.subscriptionName("test-changed-subscription-name"); } 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 97f461b8..90c04197 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 @@ -35,6 +35,7 @@ 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; @@ -43,7 +44,6 @@ 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.core.ReaderBuilderCustomizer; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -204,7 +204,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { } @Bean - public ReaderBuilderCustomizer myCustomizer(PulsarTemplate pulsarTemplate) { + public PulsarReaderReaderBuilderCustomizer myCustomizer(PulsarTemplate pulsarTemplate) { return cb -> { for (int i = 0; i < 10; i++) { try {