Switch to single generic customizer contract

Author: onobc

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/ConcurrentPulsarListenerContainerFactoryCustomizer.java

@@ -178,8 +178,7 @@ private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>>
 	ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 			PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver,
 			TopicResolver topicResolver, ObjectProvider<PulsarAwareTransactionManager> pulsarTransactionManager,
-			ObjectProvider<ConcurrentPulsarListenerContainerFactoryCustomizer> customizersProvider,
-			Environment environment) {
+			PulsarContainerFactoryCustomizers containerFactoryCustomizers, Environment environment) {
 		PulsarContainerProperties containerProperties = new PulsarContainerProperties();
 		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setTopicResolver(topicResolver);
@@ -190,7 +189,7 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
 		ConcurrentPulsarListenerContainerFactory<?> containerFactory = new ConcurrentPulsarListenerContainerFactory<>(
 				pulsarConsumerFactory, containerProperties);
-		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		containerFactoryCustomizers.customize(containerFactory);
 		return containerFactory;
 	}
 
@@ -219,8 +218,7 @@ private void applyReaderBuilderCustomizers(List<ReaderBuilderCustomizer<?>> cust
 	@Bean
 	@ConditionalOnMissingBean(name = "pulsarReaderContainerFactory")
 	DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(PulsarReaderFactory<?> pulsarReaderFactory,
-			SchemaResolver schemaResolver,
-			ObjectProvider<DefaultPulsarReaderContainerFactoryCustomizer> customizersProvider,
+			SchemaResolver schemaResolver, PulsarContainerFactoryCustomizers containerFactoryCustomizers,
 			Environment environment) {
 		PulsarReaderContainerProperties readerContainerProperties = new PulsarReaderContainerProperties();
 		readerContainerProperties.setSchemaResolver(schemaResolver);
@@ -230,7 +228,7 @@ DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(PulsarReader
 		this.propertiesMapper.customizeReaderContainerProperties(readerContainerProperties);
 		DefaultPulsarReaderContainerFactory<?> containerFactory = new DefaultPulsarReaderContainerFactory<>(
 				pulsarReaderFactory, readerContainerProperties);
-		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		containerFactoryCustomizers.customize(containerFactory);
 		return containerFactory;
 	}
 

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/DefaultReactivePulsarListenerContainerFactoryCustomizer.java

@@ -188,4 +188,11 @@ PulsarTopicBuilder pulsarTopicBuilder() {
 				this.properties.getDefaults().getTopic().getNamespace());
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	PulsarContainerFactoryCustomizers pulsarContainerFactoryCustomizers(
+			ObjectProvider<PulsarContainerFactoryCustomizer<?>> customizers) {
+		return new PulsarContainerFactoryCustomizers(customizers.orderedStream().toList());
+	}
+
 }

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfiguration.java

@@ -16,22 +16,24 @@
 
 package org.springframework.boot.autoconfigure.pulsar;
 
-import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
+import org.springframework.pulsar.config.PulsarContainerFactory;
 
 /**
- * Callback interface that can be implemented to customize a
- * {@link DefaultPulsarReaderContainerFactory}.
+ * Callback interface that can be implemented by beans wishing to customize a
+ * {@link PulsarContainerFactory} before it is fully initialized, in particular to tune
+ * its configuration.
  *
+ * @param <T> the type of the {@link PulsarContainerFactory}
  * @author Chris Bono
  * @since 3.4.0
  */
 @FunctionalInterface
-public interface DefaultPulsarReaderContainerFactoryCustomizer {
+public interface PulsarContainerFactoryCustomizer<T extends PulsarContainerFactory<?, ?>> {
 
 	/**
-	 * Customize a {@link DefaultPulsarReaderContainerFactory}.
-	 * @param containerFactory the factory to customize
+	 * Customize the container factory.
+	 * @param containerFactory the {@code PulsarContainerFactory} to customize
 	 */
-	void customize(DefaultPulsarReaderContainerFactory<?> containerFactory);
+	void customize(T containerFactory);
 
 }

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarConfiguration.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2012-2024 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.boot.autoconfigure.pulsar;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.boot.util.LambdaSafe;
+import org.springframework.pulsar.config.PulsarContainerFactory;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+
+/**
+ * Invokes the available {@link PulsarContainerFactoryCustomizer} instances in the context
+ * for a given {@link PulsarConsumerFactory}.
+ *
+ * @author Chris Bono
+ * @since 3.4.0
+ */
+public class PulsarContainerFactoryCustomizers {
+
+	private final List<PulsarContainerFactoryCustomizer<?>> customizers;
+
+	public PulsarContainerFactoryCustomizers(List<? extends PulsarContainerFactoryCustomizer<?>> customizers) {
+		this.customizers = (customizers != null) ? new ArrayList<>(customizers) : Collections.emptyList();
+	}
+
+	/**
+	 * Customize the specified {@link PulsarContainerFactory}. Locates all
+	 * {@link PulsarContainerFactoryCustomizer} beans able to handle the specified
+	 * instance and invoke {@link PulsarContainerFactoryCustomizer#customize} on them.
+	 * @param <T> the type of container factory
+	 * @param containerFactory the container factory to customize
+	 * @return the customized container factory
+	 */
+	@SuppressWarnings("unchecked")
+	public <T extends PulsarContainerFactory<?, ?>> T customize(T containerFactory) {
+		LambdaSafe.callbacks(PulsarContainerFactoryCustomizer.class, this.customizers, containerFactory)
+			.withLogger(PulsarContainerFactoryCustomizers.class)
+			.invoke((customizer) -> customizer.customize(containerFactory));
+		return containerFactory;
+	}
+
+}

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/DefaultPulsarReaderContainerFactoryCustomizer.java renamed to spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarContainerFactoryCustomizer.java

@@ -164,15 +164,14 @@ private void applyMessageConsumerBuilderCustomizers(List<ReactiveMessageConsumer
 	@ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory")
 	DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainerFactory(
 			ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory, SchemaResolver schemaResolver,
-			TopicResolver topicResolver,
-			ObjectProvider<DefaultReactivePulsarListenerContainerFactoryCustomizer> customizersProvider) {
+			TopicResolver topicResolver, PulsarContainerFactoryCustomizers containerFactoryCustomizers) {
 		ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
 		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setTopicResolver(topicResolver);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
 		DefaultReactivePulsarListenerContainerFactory<?> containerFactory = new DefaultReactivePulsarListenerContainerFactory<>(
 				reactivePulsarConsumerFactory, containerProperties);
-		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		containerFactoryCustomizers.customize(containerFactory);
 		return containerFactory;
 	}
 

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarContainerFactoryCustomizers.java

@@ -73,6 +73,7 @@
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.listener.PulsarContainerProperties.TransactionSettings;
+import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
 import org.springframework.test.util.ReflectionTestUtils;
 
@@ -596,15 +597,23 @@ void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
 		@TestConfiguration(proxyBeanMethods = false)
 		static class ListenerContainerFactoryCustomizersConfig {
 
+			@Bean
+			@Order(50)
+			PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> customizerIgnored() {
+				return (__) -> {
+					throw new RuntimeException("should-not-have-matched");
+				};
+			}
+
 			@Bean
 			@Order(200)
-			ConcurrentPulsarListenerContainerFactoryCustomizer customizerFoo() {
+			PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> customizerFoo() {
 				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":foo");
 			}
 
 			@Bean
 			@Order(100)
-			ConcurrentPulsarListenerContainerFactoryCustomizer customizerBar() {
+			PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> customizerBar() {
 				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":bar");
 			}
 
@@ -712,15 +721,23 @@ ReaderBuilderCustomizer<?> customizerBar() {
 		@TestConfiguration(proxyBeanMethods = false)
 		static class ReaderContainerFactoryCustomizersConfig {
 
+			@Bean
+			@Order(50)
+			PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> customizerIgnored() {
+				return (__) -> {
+					throw new RuntimeException("should-not-have-matched");
+				};
+			}
+
 			@Bean
 			@Order(200)
-			DefaultPulsarReaderContainerFactoryCustomizer customizerFoo() {
+			PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> customizerFoo() {
 				return (containerFactory) -> appendToReaderListener(containerFactory, ":foo");
 			}
 
 			@Bean
 			@Order(100)
-			DefaultPulsarReaderContainerFactoryCustomizer customizerBar() {
+			PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> customizerBar() {
 				return (containerFactory) -> appendToReaderListener(containerFactory, ":bar");
 			}
 

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactiveAutoConfiguration.java

@@ -86,6 +86,15 @@ void whenHasUserDefinedConnectionDetailsBeanDoesNotAutoConfigureBean() {
 				.isSameAs(customConnectionDetails));
 	}
 
+	@Test
+	void whenHasUserDefinedContainerFactoryCustomizersBeanDoesNotAutoConfigureBean() {
+		PulsarContainerFactoryCustomizers customizers = mock(PulsarContainerFactoryCustomizers.class);
+		this.contextRunner
+			.withBean("customContainerFactoryCustomizers", PulsarContainerFactoryCustomizers.class, () -> customizers)
+			.run((context) -> assertThat(context).getBean(PulsarContainerFactoryCustomizers.class)
+				.isSameAs(customizers));
+	}
+
 	@Nested
 	class ClientTests {