diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizer.java new file mode 100644 index 00000000..24cb7b3a --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizer.java @@ -0,0 +1,34 @@ +/* + * Copyright 2023-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.pulsar.annotation; + +/** + * Customize a bean. + * + * @param the bean type to customize + * @author Chris Bono + */ +@FunctionalInterface +public interface BeanCustomizer { + + /** + * Customize the bean. + * @param bean the bean to customize + */ + void customize(B bean); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizerPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizerPostProcessor.java new file mode 100644 index 00000000..69342f5d --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizerPostProcessor.java @@ -0,0 +1,76 @@ +/* + * Copyright 2023-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.pulsar.annotation; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.core.log.LogAccessor; +import org.springframework.util.CollectionUtils; + +/** + * A {@link BeanPostProcessor} that applies a customizer to beans of a specified type. + *

+ * There must be only one customizer in the application context in order for it to be + * applied. + * + * @param the type of bean to customize + * @param the type of customizer + * @author Chris Bono + */ +class BeanCustomizerPostProcessor> + implements BeanPostProcessor, ApplicationContextAware { + + private final LogAccessor logger = new LogAccessor(getClass()); + + private final Class beanType; + + private final Class customizerType; + + private ApplicationContext applicationContext; + + BeanCustomizerPostProcessor(Class beanType, Class customizerType) { + this.beanType = beanType; + this.customizerType = customizerType; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + if (this.beanType.isInstance(bean)) { + B typedBean = this.beanType.cast(bean); + var customizers = this.applicationContext.getBeansOfType(this.customizerType); + if (CollectionUtils.isEmpty(customizers)) { + return bean; + } + if (customizers.size() > 1) { + this.logger.warn("Found multiple %s beans [%s] - must be only 1 in order to apply" + .formatted(this.customizerType.getSimpleName(), customizers.keySet())); + } + else { + customizers.values().stream().forEach((c) -> c.customize(typedBean)); + } + } + return bean; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java index c8e63753..02553918 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarBootstrapConfiguration.java @@ -16,13 +16,19 @@ package org.springframework.pulsar.annotation; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; +import org.springframework.core.ResolvableType; import org.springframework.core.type.AnnotationMetadata; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactoryCustomizer; import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.config.PulsarReaderEndpointRegistry; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTemplateCustomizer; /** * An {@link ImportBeanDefinitionRegistrar} class that registers a @@ -44,6 +50,30 @@ public class PulsarBootstrapConfiguration implements ImportBeanDefinitionRegistr @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { + if (!registry.containsBeanDefinition("pulsarTemplateCustomizerPostProcessor")) { + var postProcessorType = ResolvableType.forClassWithGenerics(BeanCustomizerPostProcessor.class, + PulsarTemplate.class, PulsarTemplateCustomizer.class); + @SuppressWarnings("unchecked") + var beanDef = BeanDefinitionBuilder + .rootBeanDefinition(postProcessorType, + () -> new BeanCustomizerPostProcessor<>(PulsarTemplate.class, PulsarTemplateCustomizer.class)) + .getBeanDefinition(); + registry.registerBeanDefinition("pulsarTemplateCustomizerPostProcessor", beanDef); + } + + if (!registry.containsBeanDefinition("concurrentContainerFactoryCustomizerPostProcessor")) { + var postProcessorType = ResolvableType.forClassWithGenerics(BeanCustomizerPostProcessor.class, + ConcurrentPulsarListenerContainerFactory.class, + ConcurrentPulsarListenerContainerFactoryCustomizer.class); + @SuppressWarnings("unchecked") + var beanDef = BeanDefinitionBuilder + .rootBeanDefinition(postProcessorType, + () -> new BeanCustomizerPostProcessor<>(ConcurrentPulsarListenerContainerFactory.class, + ConcurrentPulsarListenerContainerFactoryCustomizer.class)) + .getBeanDefinition(); + registry.registerBeanDefinition("concurrentContainerFactoryCustomizerPostProcessor", beanDef); + } + if (!registry .containsBeanDefinition(PulsarAnnotationSupportBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) { registry.registerBeanDefinition( diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryCustomizer.java new file mode 100644 index 00000000..420c7cf4 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryCustomizer.java @@ -0,0 +1,31 @@ +/* + * Copyright 2022-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.pulsar.config; + +import org.springframework.pulsar.annotation.BeanCustomizer; + +/** + * Callback interface that can be implemented to customize a + * {@link ConcurrentPulsarListenerContainerFactory}. + * + * @param The message payload type + * @author Chris Bono + */ +public interface ConcurrentPulsarListenerContainerFactoryCustomizer + extends BeanCustomizer> { + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplateCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplateCustomizer.java new file mode 100644 index 00000000..ab75f47f --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplateCustomizer.java @@ -0,0 +1,29 @@ +/* + * Copyright 2022-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.pulsar.core; + +import org.springframework.pulsar.annotation.BeanCustomizer; + +/** + * Callback interface that can be implemented to customize a {@link PulsarTemplate}. + * + * @param the payload type of the template + * @author Chris Bono + */ +public interface PulsarTemplateCustomizer extends BeanCustomizer> { + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index 176555bf..fe3cdd54 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -55,6 +56,9 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.pulsar.test.support.model.UserRecord; import org.springframework.util.function.ThrowingConsumer; @@ -391,6 +395,59 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { } + @Nested + class PulsarTemplateCustomizerTests { + + @Test + void whenSingleCustomizerAvailableThenItIsApplied() { + var template = mock(PulsarTemplate.class); + var txnProps = mock(TransactionProperties.class); + when(template.transactions()).thenReturn(txnProps); + PulsarTemplateCustomizer customizer = (t) -> t.transactions().setTimeout(Duration.ofSeconds(45)); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(PulsarTemplate.class, () -> template); + appContext.registerBean(PulsarTemplateCustomizer.class, () -> customizer); + appContext.register(PulsarTemplateCustomizerTestsConfig.class); + appContext.refresh(); + verify(txnProps).setTimeout(Duration.ofSeconds(45)); + } + } + + @Test + void whenMultipleCustomizersAvailableThenNoneAreApplied() { + var template = mock(PulsarTemplate.class); + var txnProps = mock(TransactionProperties.class); + when(template.transactions()).thenReturn(txnProps); + PulsarTemplateCustomizer customizer1 = (t) -> t.transactions().setTimeout(Duration.ofSeconds(30)); + PulsarTemplateCustomizer customizer2 = (t) -> t.transactions().setTimeout(Duration.ofSeconds(45)); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(PulsarTemplate.class, () -> template); + appContext.registerBean("customizer1", PulsarTemplateCustomizer.class, () -> customizer1); + appContext.registerBean("customizer2", PulsarTemplateCustomizer.class, () -> customizer2); + appContext.register(PulsarTemplateCustomizerTestsConfig.class); + appContext.refresh(); + verify(txnProps, never()).setTimeout(any(Duration.class)); + } + } + + @Test + void whenNoCustomizersAvaiableThenContextStartsWithoutFailure() { + var template = mock(PulsarTemplate.class); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(PulsarTemplate.class, () -> template); + appContext.register(PulsarTemplateCustomizerTestsConfig.class); + appContext.refresh(); + } + } + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + static class PulsarTemplateCustomizerTestsConfig { + + } + + } + public static class Foo { private String foo; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarListenerContainerFactoryCustomizerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarListenerContainerFactoryCustomizerTests.java new file mode 100644 index 00000000..dec1a7e9 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarListenerContainerFactoryCustomizerTests.java @@ -0,0 +1,94 @@ +/* + * Copyright 2023-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.pulsar.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.annotation.EnablePulsar; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactoryCustomizer; + +/** + * Tests for applying {@link ConcurrentPulsarListenerContainerFactoryCustomizer} to the + * {@link ConcurrentPulsarListenerContainerFactory}. + * + * @author Chris Bono + */ +class ConcurrentPulsarListenerContainerFactoryCustomizerTests { + + @Test + void whenSingleCustomizerAvailableThenItIsApplied() { + var containerFactory = mock(ConcurrentPulsarListenerContainerFactory.class); + var containerProps = new PulsarContainerProperties(); + when(containerFactory.getContainerProperties()).thenReturn(containerProps); + ConcurrentPulsarListenerContainerFactoryCustomizer customizer = ( + cf) -> cf.getContainerProperties().transactions().setTimeout(Duration.ofSeconds(45)); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(ConcurrentPulsarListenerContainerFactory.class, () -> containerFactory); + appContext.registerBean(ConcurrentPulsarListenerContainerFactoryCustomizer.class, () -> customizer); + appContext.register(ConcurrentPulsarListenerContainerFactoryCustomizerTestsConfig.class); + appContext.refresh(); + assertThat(containerProps.transactions().getTimeout()).isEqualTo(Duration.ofSeconds(45)); + } + } + + @Test + void whenMultipleCustomizersAvailableThenNoneAreApplied() { + var containerFactory = mock(ConcurrentPulsarListenerContainerFactory.class); + var containerProps = new PulsarContainerProperties(); + when(containerFactory.getContainerProperties()).thenReturn(containerProps); + ConcurrentPulsarListenerContainerFactoryCustomizer customizer1 = ( + cf) -> cf.getContainerProperties().transactions().setTimeout(Duration.ofSeconds(45)); + ConcurrentPulsarListenerContainerFactoryCustomizer customizer2 = ( + cf) -> cf.getContainerProperties().transactions().setTimeout(Duration.ofSeconds(60)); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(ConcurrentPulsarListenerContainerFactory.class, () -> containerFactory); + appContext.registerBean("customizer1", ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> customizer1); + appContext.registerBean("customizer2", ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> customizer2); + appContext.register(ConcurrentPulsarListenerContainerFactoryCustomizerTestsConfig.class); + appContext.refresh(); + assertThat(containerProps.transactions().getTimeout()).isNull(); + } + } + + @Test + void whenNoCustomizersAvaiableThenContextStartsWithoutFailure() { + var containerFactory = mock(ConcurrentPulsarListenerContainerFactory.class); + try (var appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(ConcurrentPulsarListenerContainerFactory.class, () -> containerFactory); + appContext.register(ConcurrentPulsarListenerContainerFactoryCustomizerTestsConfig.class); + appContext.refresh(); + } + } + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + static class ConcurrentPulsarListenerContainerFactoryCustomizerTestsConfig { + + } + +}