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

Author: onobc

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<String> 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<String> 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]

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc

@@ -270,6 +270,8 @@ ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> 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.
 

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java

@@ -226,6 +226,10 @@ public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) {
 		this.deadLetterPolicy = deadLetterPolicy;
 	}
 
+	public ReactiveMessageConsumerBuilderCustomizer<V> getConsumerCustomizer() {
+		return this.consumerCustomizer;
+	}
+
 	public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer<V> consumerCustomizer) {
 		this.consumerCustomizer = consumerCustomizer;
 	}

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java

@@ -110,13 +110,13 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 
 	private final AtomicInteger counter = new AtomicInteger();
 
+	private final List<MethodReactivePulsarListenerEndpoint<?>> 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 void afterSingletonsInstantiated() {
 			}
 			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 @@ protected void processListener(MethodReactivePulsarListenerEndpoint<?> endpoint,
 	@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 @@ private ReactivePulsarListenerContainerFactory<?> resolveContainerFactory(
 
 	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 @@ private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListene
 		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 @@ private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListene
 			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 @@ private void resolveDeadLetterPolicy(MethodReactivePulsarListenerEndpoint<?> end
 		}
 	}
 
+	@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);