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 {