From ecd0a30c8375034f414ecda74e5a811e6bd0664c Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 11 Jul 2022 13:06:14 -0500 Subject: [PATCH] Minor cleanup and refactoring (#24) --- .../PulsarAnnotationDrivenConfiguration.java | 11 +-- .../autoconfigure/PulsarListenerTests.java | 5 +- .../pulsar/annotation/EnablePulsar.java | 5 +- .../PulsarBootstrapConfiguration.java | 13 ++-- .../pulsar/annotation/PulsarListener.java | 36 ++++++++- ...arListenerAnnotationBeanPostProcessor.java | 78 +++++++------------ .../annotation/PulsarListenerConfigurer.java | 20 ++++- .../pulsar/annotation/PulsarListeners.java | 10 ++- ...efaultPulsarListenerContainerFactory.java} | 3 +- .../config/PulsarClientFactoryBean.java | 10 ++- ...tils.java => PulsarListenerBeanNames.java} | 16 ++-- .../core/DefaultPulsarProducerFactory.java | 22 ++---- .../pulsar/core/PulsarProducerFactory.java | 25 +++++- src/api/overview.html | 8 +- src/dist/notice.txt | 4 +- 15 files changed, 156 insertions(+), 110 deletions(-) rename spring-pulsar/src/main/java/org/springframework/pulsar/config/{PulsarListenerContainerFactoryImpl.java => DefaultPulsarListenerContainerFactory.java} (94%) rename spring-pulsar/src/main/java/org/springframework/pulsar/config/{PulsarListenerConfigUtils.java => PulsarListenerBeanNames.java} (66%) diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index 20ed964d..959727cd 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -22,8 +22,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; -import org.springframework.pulsar.config.PulsarListenerConfigUtils; -import org.springframework.pulsar.config.PulsarListenerContainerFactoryImpl; +import org.springframework.pulsar.config.DefaultPulsarListenerContainerFactory; +import org.springframework.pulsar.config.PulsarListenerBeanNames; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.listener.PulsarContainerProperties; @@ -31,6 +31,7 @@ import org.springframework.pulsar.listener.PulsarContainerProperties; * Configuration for Pulsar annotation-driven support. * * @author Soby Chacko + * @author Chris Bono */ @Configuration(proxyBeanMethods = false) @ConditionalOnClass(EnablePulsar.class) @@ -44,9 +45,9 @@ public class PulsarAnnotationDrivenConfiguration { @Bean @ConditionalOnMissingBean(name = "pulsarListenerContainerFactory") - PulsarListenerContainerFactoryImpl pulsarListenerContainerFactory( + DefaultPulsarListenerContainerFactory pulsarListenerContainerFactory( ObjectProvider> pulsarConsumerFactory) { - PulsarListenerContainerFactoryImpl factory = new PulsarListenerContainerFactoryImpl<>(); + DefaultPulsarListenerContainerFactory factory = new DefaultPulsarListenerContainerFactory<>(); final PulsarConsumerFactory pulsarConsumerFactory1 = pulsarConsumerFactory.getIfAvailable(); factory.setPulsarConsumerFactory(pulsarConsumerFactory1); @@ -68,7 +69,7 @@ public class PulsarAnnotationDrivenConfiguration { @Configuration(proxyBeanMethods = false) @EnablePulsar - @ConditionalOnMissingBean(name = PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME) + @ConditionalOnMissingBean(name = PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME) static class EnableKafkaConfiguration { } diff --git a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java index 88fdf109..22529bee 100644 --- a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java +++ b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java @@ -33,7 +33,10 @@ import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.PulsarTemplate; /** + * Tests for {@link PulsarListener}. + * * @author Soby Chacko + * @author Chris Bono */ class PulsarListenerTests extends AbstractContainerBaseTests { @@ -86,7 +89,7 @@ class PulsarListenerTests extends AbstractContainerBaseTests { @Import(PulsarAutoConfiguration.class) public static class BatchListenerConfig { - @PulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive", batch = "true") + @PulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive", batch = true) public void listen(List foo) { foo.forEach(t -> latch2.countDown()); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/EnablePulsar.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/EnablePulsar.java index 724ec06a..14c0fbe6 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/EnablePulsar.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/EnablePulsar.java @@ -25,10 +25,10 @@ import java.lang.annotation.Target; import org.springframework.context.annotation.Import; /** - * Enable Pulsar listener annotated endpoints that are created under the covers by a - * {@link org.springframework.pulsar.config.AbstractPulsarListenerContainerFactory}. + * Enables detection of {@link PulsarListener} annotations on any Spring-managed bean in the container. * * @author Soby Chacko + * @author Chris Bono */ @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @@ -36,4 +36,3 @@ import org.springframework.context.annotation.Import; @Import(PulsarListenerConfigurationSelector.class) public @interface EnablePulsar { } - 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 bde6dc7e..65c7c553 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 @@ -20,7 +20,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; import org.springframework.core.type.AnnotationMetadata; -import org.springframework.pulsar.config.PulsarListenerConfigUtils; +import org.springframework.pulsar.config.PulsarListenerBeanNames; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; /** @@ -32,6 +32,7 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; * annotation. * * @author Soby Chacko + * @author Chris Bono * * @see PulsarListenerAnnotationBeanPostProcessor * @see PulsarListenerEndpointRegistry @@ -41,15 +42,13 @@ public class PulsarBootstrapConfiguration implements ImportBeanDefinitionRegistr @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { - if (!registry.containsBeanDefinition( - PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) { - - registry.registerBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME, + if (!registry.containsBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) { + registry.registerBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME, new RootBeanDefinition(PulsarListenerAnnotationBeanPostProcessor.class)); } - if (!registry.containsBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) { - registry.registerBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, + if (!registry.containsBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) { + registry.registerBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, new RootBeanDefinition(PulsarListenerEndpointRegistry.class)); } } 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 05610016..f5a05855 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 @@ -25,6 +25,8 @@ import java.lang.annotation.Target; 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; /** * Annotation that marks a method to be the target of a Pulsar message listener on the @@ -43,6 +45,7 @@ import org.springframework.messaging.handler.annotation.MessageMapping; *

* * @author Soby Chacko + * @author Chris Bono */ @Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) @@ -55,7 +58,7 @@ public @interface PulsarListener { *

If none is specified an auto-generated id is used. *

SpEL {@code #{...}} and property place holders {@code ${...}} are supported. * @return the {@code id} for the container managing for this endpoint. - * @see org.springframework.pulsar.config.PulsarListenerEndpointRegistry#getListenerContainer(String) + * @see PulsarListenerEndpointRegistry#getListenerContainer(String) */ String id() default ""; @@ -74,8 +77,13 @@ public @interface PulsarListener { SchemaType schemaType() default SchemaType.NONE; /** - * Specific container factory to use on this listener. - * @return {@code containerFactory} to use on this Pulsar listener. + * The bean name of the {@link PulsarListenerContainerFactory} + * to use to create the message listener container responsible to serve this endpoint. + *

+ * If not specified, the default container factory is used, if any. If a SpEL + * expression is provided ({@code #{...}}), the expression can either evaluate to a + * container factory instance or a bean name. + * @return the container factory bean name. */ String containerFactory() default ""; @@ -108,7 +116,7 @@ public @interface PulsarListener { * * @return whether this listener is in batch mode or not. */ - String batch() default ""; + boolean batch() default false; /** * A pseudo bean name used in SpEL expressions within this annotation to reference @@ -120,5 +128,25 @@ public @interface PulsarListener { */ String beanRef() default "__listener"; + /** + * Pulsar consumer properties; they will supersede any properties with the same name + * defined in the consumer factory (if the consumer factory supports property overrides). + *

+ * Supported Syntax + *

The supported syntax for key-value pairs is the same as the + * syntax defined for entries in a Java + * {@linkplain java.util.Properties#load(java.io.Reader) properties file}: + *

    + *
  • {@code key=value}
    • + *
  • {@code key:value}
    • + *
  • {@code key value}
    • + *
+ * {@code group.id} and {@code client.id} are ignored. + *

SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * SpEL expressions must resolve to a {@link String}, a @{link String[]} or a + * {@code Collection} where each member of the array or collection is a + * property name + value with the above formats. + * @return the properties. + */ String[] properties() 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 0f0e706e..fde119f5 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 @@ -47,7 +47,6 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.InitializingBean; -import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.ObjectFactory; import org.springframework.beans.factory.SmartInitializingSingleton; @@ -61,7 +60,6 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.MethodIntrospector; -import org.springframework.core.OrderComparator; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.core.annotation.AnnotationUtils; @@ -80,8 +78,9 @@ import org.springframework.messaging.handler.annotation.support.MessageHandlerMe import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.pulsar.config.MethodPulsarListenerEndpoint; -import org.springframework.pulsar.config.PulsarListenerConfigUtils; +import org.springframework.pulsar.config.PulsarListenerBeanNames; 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.util.Assert; @@ -92,12 +91,12 @@ import org.springframework.validation.Validator; /** * Bean post-processor that registers methods annotated with {@link PulsarListener} * to be invoked by a Pulsar message listener container created under the covers - * by a {@link org.springframework.pulsar.config.PulsarListenerContainerFactory} + * by a {@link PulsarListenerContainerFactory} * according to the parameters of the annotation. * *

Annotated methods can use flexible arguments as defined by {@link PulsarListener}. * - *

This post-processor is automatically registered by Spring's {@link EnablePulsar} + *

This post-processor is automatically registered by the {@link EnablePulsar} * annotation. * *

Auto-detect any {@link PulsarListenerConfigurer} instances in the container, @@ -109,13 +108,14 @@ import org.springframework.validation.Validator; * @param the value type. * * @author Soby Chacko + * @author Chris Bono * * @see PulsarListener * @see EnablePulsar * @see PulsarListenerConfigurer * @see PulsarListenerEndpointRegistrar * @see PulsarListenerEndpointRegistry - * @see org.springframework.pulsar.config.PulsarListenerEndpoint + * @see PulsarListenerEndpoint * @see MethodPulsarListenerEndpoint */ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton { @@ -136,17 +136,20 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost private static final String GENERATED_ID_PREFIX = "org.springframework.Pulsar.PulsarListenerEndpointContainer#"; private ApplicationContext applicationContext; + private BeanFactory beanFactory; + private BeanExpressionResolver resolver; + private BeanExpressionContext expressionContext; + private PulsarListenerEndpointRegistry endpointRegistry; private String defaultContainerFactoryBeanName = DEFAULT_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME; private final PulsarListenerEndpointRegistrar registrar = new PulsarListenerEndpointRegistrar(); - private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory = - new PulsarHandlerMethodFactoryAdapter(); + private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory = new PulsarHandlerMethodFactoryAdapter(); private Charset charset = StandardCharsets.UTF_8; @@ -179,23 +182,20 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost } @Override - public void afterPropertiesSet() throws Exception { + public void afterPropertiesSet() { buildEnhancer(); } private void buildEnhancer() { if (this.applicationContext != null) { - Map enhancersMap = - this.applicationContext.getBeansOfType(AnnotationEnhancer.class, false, false); - if (enhancersMap.size() > 0) { - List enhancers = enhancersMap.values() - .stream() - .sorted(new OrderComparator()) - .collect(Collectors.toList()); + List enhancers = this.applicationContext + .getBeanProvider(AnnotationEnhancer.class, false) + .orderedStream() + .toList(); + if (!enhancers.isEmpty()) { this.enhancer = (attrs, element) -> { - Map newAttrs = attrs; for (AnnotationEnhancer enh : enhancers) { - newAttrs = enh.apply(newAttrs, element); + attrs = enh.apply(attrs, element); } return attrs; }; @@ -207,20 +207,15 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost public void afterSingletonsInstantiated() { this.registrar.setBeanFactory(this.beanFactory); - if (this.beanFactory instanceof ListableBeanFactory) { - Map instances = - ((ListableBeanFactory) this.beanFactory).getBeansOfType(PulsarListenerConfigurer.class); - for (PulsarListenerConfigurer configurer : instances.values()) { - configurer.configurePulsarListeners(this.registrar); - } - } + 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, "BeanFactory must be set to find endpoint registry by bean name"); this.endpointRegistry = this.beanFactory.getBean( - PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, + PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, PulsarListenerEndpointRegistry.class); } this.registrar.setEndpointRegistry(this.endpointRegistry); @@ -230,7 +225,7 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName); } - // Set the custom handler method factory once resolved by the configurer + // Set the custom handler method factory once resolved by the configurer - otherwise register default formatters MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory(); if (handlerMethodFactory != null) { this.messageHandlerMethodFactory.setHandlerMethodFactory(handlerMethodFactory); @@ -252,7 +247,6 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { if (!this.nonAnnotatedClasses.contains(bean.getClass())) { Class targetClass = AopUtils.getTargetClass(bean); - Collection classLevelListeners = findListenerAnnotations(targetClass); Map> annotatedMethods = MethodIntrospector.selectMethods(targetClass, (MethodIntrospector.MetadataLookup>) method -> { Set listenerMethods = findListenerAnnotations(method); @@ -360,9 +354,7 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup")); } resolvePulsarProperties(endpoint, pulsarListener.properties()); - if (StringUtils.hasText(pulsarListener.batch())) { - endpoint.setBatchListener(Boolean.parseBoolean(pulsarListener.batch())); - } + endpoint.setBatchListener(pulsarListener.batch()); endpoint.setBeanFactory(this.beanFactory); } @@ -588,26 +580,9 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost private void addFormatters(FormatterRegistry registry) { - for (Converter converter : getBeansOfType(Converter.class)) { - registry.addConverter(converter); - } - for (GenericConverter converter : getBeansOfType(GenericConverter.class)) { - registry.addConverter(converter); - } - for (Formatter formatter : getBeansOfType(Formatter.class)) { - registry.addFormatter(formatter); - } - } - - private Collection getBeansOfType(Class type) { - if (PulsarListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ListableBeanFactory) { - return ((ListableBeanFactory) PulsarListenerAnnotationBeanPostProcessor.this.beanFactory) - .getBeansOfType(type) - .values(); - } - else { - return Collections.emptySet(); - } + this.beanFactory.getBeanProvider(Converter.class).forEach(registry::addConverter); + this.beanFactory.getBeanProvider(GenericConverter.class).forEach(registry::addConverter); + this.beanFactory.getBeanProvider(Formatter.class).forEach(registry::addFormatter); } @Override @@ -802,4 +777,3 @@ public class PulsarListenerAnnotationBeanPostProcessor implements BeanPost } - diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java index 636f92f8..11075f15 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java @@ -16,14 +16,32 @@ package org.springframework.pulsar.annotation; +import org.springframework.pulsar.config.PulsarListenerContainerFactory; +import org.springframework.pulsar.config.PulsarListenerEndpoint; import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar; +import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; /** - * Allow custom configuration on PulsarListener endpoint registry. + * Optional interface to be implemented by Spring managed bean willing to + * customize how Pulsar listener endpoints are configured. Typically used + * to define the default {@link PulsarListenerContainerFactory} to use or + * for registering Pulsar endpoints in a programmatic fashion as + * opposed to the declarative approach of using the + * {@link PulsarListener} annotation. * * @author Soby Chacko + * @author Chris Bono + * + * @see PulsarListenerEndpointRegistrar */ public interface PulsarListenerConfigurer { + /** + * Callback allowing a {@link PulsarListenerEndpointRegistry} and specific + * {@link PulsarListenerEndpoint} instances to be registered against the + * given {@link PulsarListenerEndpointRegistrar}. The default + * {@link PulsarListenerContainerFactory} can also be customized. + * @param registrar the registrar to be configured + */ void configurePulsarListeners(PulsarListenerEndpointRegistrar registrar); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java index e9f3b0c4..8dc83547 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java @@ -23,9 +23,17 @@ import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** - * Container annotation for aggregating several {@link PulsarListener} annotations. + * Container annotation that aggregates several {@link PulsarListener} annotations. + *

+ * Can be used natively, declaring several nested {@link PulsarListener} annotations. + * Can also be used in conjunction with Java 8's support for repeatable annotations, + * where {@link PulsarListener} can simply be declared several times on the same method + * (or class), implicitly generating this container annotation. * * @author Soby Chacko + * @author Chris Bono + * + * @see PulsarListener */ @Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarListenerContainerFactory.java similarity index 94% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java rename to spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarListenerContainerFactory.java index 9067fdd2..ce9971a5 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarListenerContainerFactory.java @@ -32,8 +32,9 @@ import org.springframework.util.StringUtils; * @param message type in the listener. * * @author Soby Chacko + * @author Chris Bono */ -public class PulsarListenerContainerFactoryImpl extends AbstractPulsarListenerContainerFactory, T> { +public class DefaultPulsarListenerContainerFactory extends AbstractPulsarListenerContainerFactory, T> { @Override protected DefaultPulsarMessageListenerContainer createContainerInstance(PulsarListenerEndpoint endpoint) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java index 03357ee5..6fd56e99 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java @@ -16,17 +16,23 @@ package org.springframework.pulsar.config; +import org.apache.commons.logging.LogFactory; import org.apache.pulsar.client.api.PulsarClient; +import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.core.log.LogAccessor; /** - * {@link org.springframework.beans.factory.FactoryBean} implementation for the {@link PulsarClient}. + * {@link FactoryBean} implementation for the {@link PulsarClient}. * * @author Soby Chacko + * @author Chris Bono */ public class PulsarClientFactoryBean extends AbstractFactoryBean { + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); + private final PulsarClientConfiguration pulsarClientConfiguration; public PulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { @@ -48,7 +54,7 @@ public class PulsarClientFactoryBean extends AbstractFactoryBean { @Override protected void destroyInstance(PulsarClient instance) throws Exception { if (instance != null) { - this.logger.info("Closing the client: " + instance); + this.logger.info(() -> "Closing client " + instance); instance.close(); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java similarity index 66% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java rename to spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java index 9be83f4b..534c2823 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java @@ -17,29 +17,23 @@ package org.springframework.pulsar.config; /** - * Configuration constants for internal sharing across subpackages. + * Constants for bean names that need to be accessed in multiple places by name. * * @author Soby Chacko + * @author Chris Bono */ -public abstract class PulsarListenerConfigUtils { +public abstract class PulsarListenerBeanNames { /** - * The bean name of the internally managed Kafka listener annotation processor. + * The bean name of the internally managed Pulsar listener annotation processor. */ public static final String PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME = "org.springframework.pulsar.config.internalKafkaListenerAnnotationProcessor"; /** - * The bean name of the internally managed Kafka listener endpoint registry. + * The bean name of the internally managed Pulsar listener endpoint registry. */ public static final String PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME = "org.springframework.pulsar.config.internalKafkaListenerEndpointRegistry"; - /** - * The bean name of the internally managed Kafka consumer back off manager. - */ - public static final String PULSAR_CONSUMER_BACK_OFF_MANAGER_BEAN_NAME = - "org.springframework.pulsar.config.internalKafkaConsumerBackOffManager"; - } - diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index e5733485..b602bba1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -32,15 +32,16 @@ import org.springframework.core.log.LogAccessor; import org.springframework.util.CollectionUtils; /** - * Default implementation for {@link PulsarProducerFactory}. + * Default implementation of {@link PulsarProducerFactory}. * * @param producer type. * * @author Soby Chacko + * @author Chris Bono */ public class DefaultPulsarProducerFactory implements PulsarProducerFactory, DisposableBean { - protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); private final Map producerConfig = new HashMap<>(); @@ -57,25 +58,18 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public Producer createProducer(Schema schema) throws PulsarClientException { - - final ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); - - if (!CollectionUtils.isEmpty(this.producerConfig)) { - producerBuilder.loadConf(this.producerConfig); - } - this.producer = producerBuilder.create(); - return this.producer; + return createProducer(schema, null); } @Override public Producer createProducer(Schema schema, MessageRouter messageRouter) throws PulsarClientException { - final ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); - if (!CollectionUtils.isEmpty(this.producerConfig)) { producerBuilder.loadConf(this.producerConfig); } - producerBuilder.messageRouter(messageRouter); + if (messageRouter != null) { + producerBuilder.messageRouter(messageRouter); + } this.producer = producerBuilder.create(); return this.producer; } @@ -87,7 +81,7 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public void destroy() throws Exception { - this.logger.info("Closing the producer"); + this.logger.info("Closing producer"); this.producer.close(); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index 292a5e61..d43036b7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -24,17 +24,38 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; /** - * Pulsar producer factory interface. + * The strategy to produce a {@link Producer} instance(s). * - * @param producer payload type. + * @param producer payload type * * @author Soby Chacko + * @author Chris Bono */ public interface PulsarProducerFactory { + /** + * Create a producer. + * + * @param schema the schema of the messages to be sent + * @return the producer + * @throws PulsarClientException if any error occurs + */ Producer createProducer(Schema schema) throws PulsarClientException; + /** + * Create a producer. + * + * @param schema the schema of the messages to be sent + * @param messageRouter the optional message router to use + * @return the producer + * @throws PulsarClientException if any error occurs + */ Producer createProducer(Schema schema, MessageRouter messageRouter) throws PulsarClientException; + /** + * Return a map of configuration options to use when creating producers. + * + * @return the map of configuration options + */ Map getProducerConfig(); } diff --git a/src/api/overview.html b/src/api/overview.html index e063a069..3366f9a7 100644 --- a/src/api/overview.html +++ b/src/api/overview.html @@ -1,12 +1,12 @@ -This document is the API specification for Spring for Apache Kafka project +This document is the API specification for Spring for Apache Pulsar project

For further API reference and developer documentation, see the - Spring - for Apache Kafka reference documentation. + Spring + for Apache Pulsar reference documentation. That documentation contains more detailed, developer-targeted descriptions, with conceptual overviews, definitions of terms, workarounds, and working code examples. @@ -14,7 +14,7 @@ This document is the API specification for Spring for Apache Kafka project

If you are interested in commercial consultancy, and - support for Spring for Apache Kafka, please visit + support for Spring for Apache Pulsar, please visit https://spring.io/

diff --git a/src/dist/notice.txt b/src/dist/notice.txt index 94dbff2f..9d4fc59b 100644 --- a/src/dist/notice.txt +++ b/src/dist/notice.txt @@ -1,5 +1,5 @@ -Spring for Apache Kafka Framework ${version} -Copyright (c) 2016-${copyright} Pivotal, Inc. +Spring for Apache Pulsar Framework ${version} +Copyright (c) 2022-${copyright} VMware, Inc. This product is licensed to you under the Apache License, Version 2.0 (the "License"). You may not use this product except in compliance with