From 5c7655d9f5e1a5ed7e3149318a50d0904a9b0ca7 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 3 May 2024 11:21:10 -0500 Subject: [PATCH] Add support for bean customizers (#669) This commit adds the notion of `BeanCustomizer` and a generic bean post processor that will apply any unique customizer to its target bean. Included are concrete customizers for PulsarTemplate and ConcurrentPulsarListenerContainerFactory in order to make it easier to configure transactions to these components. --- .../pulsar/annotation/BeanCustomizer.java | 34 +++++++ .../BeanCustomizerPostProcessor.java | 76 +++++++++++++++ .../PulsarBootstrapConfiguration.java | 30 ++++++ ...sarListenerContainerFactoryCustomizer.java | 31 ++++++ .../pulsar/core/PulsarTemplateCustomizer.java | 29 ++++++ .../pulsar/core/PulsarTemplateTests.java | 57 +++++++++++ ...stenerContainerFactoryCustomizerTests.java | 94 +++++++++++++++++++ 7 files changed, 351 insertions(+) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/BeanCustomizerPostProcessor.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryCustomizer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplateCustomizer.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarListenerContainerFactoryCustomizerTests.java 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 { + + } + +}