From e91e8e33d5c089ed700546ddf23a79101c45b0a6 Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Wed, 16 Nov 2022 21:32:23 +0100 Subject: [PATCH] Add ReactivePulsarListener annotation (#203) Fixes #8 --- .../PulsarAutoConfigurationTests.java | 4 +- .../pulsar/annotation/PulsarListener.java | 14 +- ...arListenerAnnotationBeanPostProcessor.java | 13 +- .../PulsarListenerConfigurationSelector.java | 8 +- .../ReactivePulsarBootstrapConfiguration.java | 60 ++ .../annotation/ReactivePulsarListener.java | 161 ++++ ...arListenerAnnotationBeanPostProcessor.java | 785 ++++++++++++++++++ .../annotation/ReactivePulsarListeners.java | 45 + ...bstractPulsarListenerContainerFactory.java | 2 +- .../AbstractPulsarListenerEndpoint.java | 4 - .../config/ListenerContainerFactory.java | 51 ++ .../pulsar/config/ListenerEndpoint.java | 106 +++ .../config/ListenerEndpointRegistry.java | 257 ++++++ .../config/PulsarListenerBeanNames.java | 10 + .../PulsarListenerContainerFactory.java | 9 +- .../pulsar/config/PulsarListenerEndpoint.java | 32 +- .../PulsarListenerEndpointRegistrar.java | 48 +- .../PulsarListenerEndpointRegistry.java | 217 +---- ...bstractReactivePulsarListenerEndpoint.java | 229 +++++ ...eactivePulsarListenerContainerFactory.java | 176 ++++ .../MethodReactivePulsarListenerEndpoint.java | 282 +++++++ ...eactivePulsarListenerContainerFactory.java | 31 + .../ReactivePulsarListenerEndpoint.java | 32 + ...ReactivePulsarListenerEndpointAdapter.java | 83 ++ ...eactivePulsarListenerEndpointRegistry.java | 46 + ...bstractPulsarMessageListenerContainer.java | 2 +- .../listener/ListenerContainerRegistry.java | 67 ++ .../listener/MessageListenerContainer.java | 39 + .../PulsarMessageListenerContainer.java | 15 +- ...PulsarMessagingMessageListenerAdapter.java | 24 +- ...eByOneMessagingMessageListenerAdapter.java | 67 ++ ...eamingMessagingMessageListenerAdapter.java | 59 ++ ...eactivePulsarMessageListenerContainer.java | 2 +- .../ReactivePulsarContainerProperties.java | 19 +- ...eactivePulsarMessageListenerContainer.java | 14 +- ...ntPulsarMessageListenerContainerTests.java | 2 +- .../pulsar/listener/PulsarListenerTests.java | 2 +- .../reactive/ReactivePulsarListenerTests.java | 613 ++++++++++++++ .../ObservationIntegrationTests.java | 2 +- .../pulsar/observation/ObservationTests.java | 2 +- 40 files changed, 3306 insertions(+), 328 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpointRegistry.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/ListenerContainerRegistry.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/MessageListenerContainer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 7f2f91af..d4928c74 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -48,7 +48,6 @@ import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; -import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.pulsar.observation.PulsarTemplateObservationConvention; @@ -156,8 +155,7 @@ class PulsarAutoConfigurationTests { @Test void customPulsarListenerContainerFactoryIsRespected() { - PulsarListenerContainerFactory> listenerContainerFactory = mock( - PulsarListenerContainerFactory.class); + PulsarListenerContainerFactory listenerContainerFactory = mock(PulsarListenerContainerFactory.class); this.contextRunner .withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class, () -> listenerContainerFactory) 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 399d2897..1feb85ed 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 @@ -34,11 +34,11 @@ import org.springframework.pulsar.listener.AckMode; * Annotation that marks a method to be the target of a Pulsar message listener on the * specified topics. * - * The {@link #containerFactory()} identifies the - * {@link org.springframework.pulsar.config.PulsarListenerContainerFactory} to use to - * build the Pulsar listener container. If not set, a default container factory - * is assumed to be available with a bean name of {@code pulsarListenerContainerFactory} - * unless an explicit default has been provided through configuration. + * The {@link #containerFactory()} identifies the {@link PulsarListenerContainerFactory} + * to use to build the Pulsar listener container. If not set, a default container + * factory is assumed to be available with a bean name of + * {@code pulsarListenerContainerFactory} unless an explicit default has been provided + * through configuration. * *

* Processing of {@code @PulsarListener} annotations is performed by registering a @@ -79,6 +79,10 @@ public @interface PulsarListener { */ SubscriptionType subscriptionType() default SubscriptionType.Exclusive; + /** + * Pulsar schema type for this listener. + * @return the {@code schemaType} for this listener + */ SchemaType schemaType() default SchemaType.NONE; /** 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 96190716..cfaf560c 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 @@ -149,7 +149,8 @@ public class PulsarListenerAnnotationBeanPostProcessor private String defaultContainerFactoryBeanName = DEFAULT_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME; - private final PulsarListenerEndpointRegistrar registrar = new PulsarListenerEndpointRegistrar(); + private final PulsarListenerEndpointRegistrar registrar = new PulsarListenerEndpointRegistrar( + PulsarListenerContainerFactory.class); private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory = new PulsarHandlerMethodFactoryAdapter(); @@ -289,26 +290,26 @@ public class PulsarListenerAnnotationBeanPostProcessor processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics, topicPattern); String containerFactory = resolve(PulsarListener.containerFactory()); - PulsarListenerContainerFactory listenerContainerFactory = resolveContainerFactory(PulsarListener, + PulsarListenerContainerFactory listenerContainerFactory = resolveContainerFactory(PulsarListener, containerFactory, beanName); this.registrar.registerEndpoint(endpoint, listenerContainerFactory); } @Nullable - private PulsarListenerContainerFactory resolveContainerFactory(PulsarListener PulsarListener, - Object factoryTarget, String beanName) { + private PulsarListenerContainerFactory resolveContainerFactory(PulsarListener PulsarListener, Object factoryTarget, + String beanName) { String containerFactory = PulsarListener.containerFactory(); if (!StringUtils.hasText(containerFactory)) { return null; } - PulsarListenerContainerFactory factory = null; + PulsarListenerContainerFactory factory = null; Object resolved = resolveExpression(containerFactory); if (resolved instanceof PulsarListenerContainerFactory) { - return (PulsarListenerContainerFactory) resolved; + return (PulsarListenerContainerFactory) resolved; } String containerFactoryBeanName = resolveExpressionAsString(containerFactory, "containerFactory"); if (StringUtils.hasText(containerFactoryBeanName)) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java index 11dd80fa..2eeac1de 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java @@ -21,8 +21,9 @@ import org.springframework.core.annotation.Order; import org.springframework.core.type.AnnotationMetadata; /** - * A {@link DeferredImportSelector} implementation with the lowest order to import a - * {@link PulsarBootstrapConfiguration} as late as possible. + * A {@link DeferredImportSelector} implementation with the lowest order to import + * {@link PulsarBootstrapConfiguration} and {@link ReactivePulsarBootstrapConfiguration} + * as late as possible. * * @author Soby Chacko * @@ -32,7 +33,8 @@ public class PulsarListenerConfigurationSelector implements DeferredImportSelect @Override public String[] selectImports(AnnotationMetadata importingClassMetadata) { - return new String[] { PulsarBootstrapConfiguration.class.getName() }; + return new String[] { PulsarBootstrapConfiguration.class.getName(), + ReactivePulsarBootstrapConfiguration.class.getName() }; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java new file mode 100644 index 00000000..f109f7d8 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java @@ -0,0 +1,60 @@ +/* + * Copyright 2022 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.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.PulsarListenerBeanNames; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; + +/** + * An {@link ImportBeanDefinitionRegistrar} class that registers a + * {@link ReactivePulsarListenerAnnotationBeanPostProcessor} bean capable of processing + * Spring's @{@link ReactivePulsarListener} annotation. Also register a default + * {@link ReactivePulsarListenerEndpointRegistry}. + * + *

+ * This configuration class is automatically imported when using the @{@link EnablePulsar} + * annotation. + * + * @author Christophe Bornet + * @see ReactivePulsarListenerAnnotationBeanPostProcessor + * @see ReactivePulsarListenerEndpointRegistry + * @see EnablePulsar + */ +public class ReactivePulsarBootstrapConfiguration implements ImportBeanDefinitionRegistrar { + + @Override + public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { + if (!registry.containsBeanDefinition( + PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) { + registry.registerBeanDefinition( + PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME, + new RootBeanDefinition(ReactivePulsarListenerAnnotationBeanPostProcessor.class)); + } + + if (!registry + .containsBeanDefinition(PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) { + registry.registerBeanDefinition( + PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, + new RootBeanDefinition(ReactivePulsarListenerEndpointRegistry.class)); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java new file mode 100644 index 00000000..bb386c33 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java @@ -0,0 +1,161 @@ +/* + * Copyright 2022 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 java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; + +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; + +/** + * Annotation that marks a method to be the target of a Pulsar message listener on the + * specified topics. + * + * The {@link #containerFactory()} identifies the + * {@link ReactivePulsarListenerContainerFactory} to use to build the Pulsar listener + * container. If not set, a default container factory is assumed to be available + * with a bean name of {@code pulsarListenerContainerFactory} unless an explicit default + * has been provided through configuration. + * + *

+ * Processing of {@code @ReactivePulsarListener} annotations is performed by registering a + * {@link ReactivePulsarListenerAnnotationBeanPostProcessor}. This can be done manually + * or, more conveniently, through {@link EnablePulsar} annotation. + *

+ * + * @author Christophe Bornet + */ +@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) +@Retention(RetentionPolicy.RUNTIME) +@MessageMapping +@Documented +public @interface ReactivePulsarListener { + + /** + * The unique identifier of the container for this listener. + *

+ * 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 ReactivePulsarListenerEndpointRegistry#getListenerContainer(String) + */ + String id() default ""; + + /** + * Pulsar subscription name associated with this listener. + * @return the {@code subscriptionName} for this Pulsar listener endpoint. + */ + String subscriptionName() default ""; + + /** + * Pulsar subscription type for this listener. + * @return the {@code subscriptionType} for this listener + */ + SubscriptionType subscriptionType() default SubscriptionType.Exclusive; + + /** + * Pulsar schema type for this listener. + * @return the {@code schemaType} for this listener + */ + SchemaType schemaType() default SchemaType.NONE; + + /** + * The bean name of the {@link ReactivePulsarListenerContainerFactory} 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 ""; + + /** + * Topics to listen to. + * @return a comma separated list of topics to listen from. + */ + String[] topics() default {}; + + /** + * Topic patten to listen to. + * @return topic pattern to listen to. + */ + String topicPattern() default ""; + + /** + * Set to true or false, to override the default setting in the container factory. May + * be a property placeholder or SpEL expression that evaluates to a {@link Boolean} or + * a {@link String}, in which case the {@link Boolean#parseBoolean(String)} is used to + * obtain the value. + *

+ * SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * @return true to auto start, false to not auto start. + */ + String autoStartup() default ""; + + /** + * Activate stream consumption. + * @return if true, the listener method shall take a + * {@link reactor.core.publisher.Flux} as input argument. + */ + boolean stream() default false; + + /** + * A pseudo bean name used in SpEL expressions within this annotation to reference the + * current bean within which this listener is defined. This allows access to + * properties and methods within the enclosing bean. Default '__listener'. + *

+ * @return the pseudo bean name. + */ + String beanRef() default "__listener"; + + /** + * Override the container factory's {@code concurrency} setting for this listener. May + * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in + * which case {@link Number#intValue()} is used to obtain the value. + *

+ * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported. + * @return the concurrency. + */ + String concurrency() default ""; + + /** + * The bean name or a 'SpEL' expression that resolves to a + * {@link org.apache.pulsar.client.api.DeadLetterPolicy} to use on the consumer to + * configure a dead letter policy for message redelivery. + * @return the bean name or empty string to not set any dead letter policy. + */ + String deadLetterPolicy() default ""; + + /** + * The bean name or a 'SpEL' expression that resolves to a + * {@link org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer} + * to use to configure the consumer. + * @return the bean name or empty string to not configure the consumer. + */ + String consumerCustomizer() default ""; + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java new file mode 100644 index 00000000..1d7ebef7 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java @@ -0,0 +1,785 @@ +/* + * Copyright 2022 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 java.io.IOException; +import java.io.StringReader; +import java.lang.reflect.AnnotatedElement; +import java.lang.reflect.Method; +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiFunction; + +import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.DeadLetterPolicy; + +import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; +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.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.ObjectFactory; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.beans.factory.config.BeanExpressionContext; +import org.springframework.beans.factory.config.BeanExpressionResolver; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.config.Scope; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.MethodIntrospector; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.core.convert.converter.ConditionalGenericConverter; +import org.springframework.core.convert.converter.Converter; +import org.springframework.core.convert.converter.GenericConverter; +import org.springframework.core.log.LogAccessor; +import org.springframework.format.Formatter; +import org.springframework.format.FormatterRegistry; +import org.springframework.format.support.DefaultFormattingConversionService; +import org.springframework.lang.Nullable; +import org.springframework.messaging.converter.GenericMessageConverter; +import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; +import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.pulsar.config.PulsarListenerBeanNames; +import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar; +import org.springframework.pulsar.config.reactive.MethodReactivePulsarListenerEndpoint; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpoint; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; +import org.springframework.validation.Validator; + +/** + * Bean post-processor that registers methods annotated with + * {@link ReactivePulsarListener} to be invoked by a Pulsar message listener container + * created under the covers by a {@link ReactivePulsarListenerContainerFactory} according + * to the parameters of the annotation. + * + *

+ * Annotated methods can use flexible arguments as defined by + * {@link ReactivePulsarListener}. + * + *

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

+ * Auto-detect any {@link PulsarListenerConfigurer} instances in the container, allowing + * for customization of the registry to be used, the default container factory or for + * fine-grained control over endpoints registration. See {@link EnablePulsar} Javadoc for + * complete usage details. + * + * @param the payload type. + * @author Christophe Bornet + * @see ReactivePulsarListener + * @see EnablePulsar + * @see PulsarListenerConfigurer + * @see PulsarListenerEndpointRegistrar + * @see ReactivePulsarListenerEndpointRegistry + * @see ReactivePulsarListenerEndpoint + * @see MethodReactivePulsarListenerEndpoint + */ +public class ReactivePulsarListenerAnnotationBeanPostProcessor + implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton { + + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); + + /** + * The bean name of the default {@link ReactivePulsarListenerContainerFactory}. + */ + public static final String DEFAULT_REACTIVE_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "reactivePulsarListenerContainerFactory"; + + private static final String THE_LEFT = "The ["; + + private static final String RESOLVED_TO_LEFT = "Resolved to ["; + + private static final String RIGHT_FOR_LEFT = "] for ["; + + private static final String GENERATED_ID_PREFIX = "org.springframework.Pulsar.ReactivePulsarListenerEndpointContainer#"; + + private ApplicationContext applicationContext; + + private BeanFactory beanFactory; + + private BeanExpressionResolver resolver; + + private BeanExpressionContext expressionContext; + + private ReactivePulsarListenerEndpointRegistry endpointRegistry; + + private String defaultContainerFactoryBeanName = DEFAULT_REACTIVE_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME; + + private final PulsarListenerEndpointRegistrar registrar = new PulsarListenerEndpointRegistrar( + ReactivePulsarListenerContainerFactory.class); + + private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory = new PulsarHandlerMethodFactoryAdapter(); + + private Charset charset = StandardCharsets.UTF_8; + + private final Set> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64)); + + private final ListenerScope listenerScope = new ListenerScope(); + + private AnnotationEnhancer enhancer; + + private final AtomicInteger counter = new AtomicInteger(); + + @Override + public int getOrder() { + return LOWEST_PRECEDENCE; + } + + public void setEndpointRegistry(ReactivePulsarListenerEndpointRegistry endpointRegistry) { + this.endpointRegistry = endpointRegistry; + } + + public void setDefaultContainerFactoryBeanName(String containerFactoryBeanName) { + this.defaultContainerFactoryBeanName = containerFactoryBeanName; + } + + public void setCharset(Charset charset) { + Assert.notNull(charset, "'charset' cannot be null"); + this.charset = charset; + } + + @Override + public void afterPropertiesSet() { + buildEnhancer(); + } + + private void buildEnhancer() { + if (this.applicationContext != null) { + List enhancers = this.applicationContext + .getBeanProvider(AnnotationEnhancer.class, false).orderedStream().toList(); + if (!enhancers.isEmpty()) { + this.enhancer = (attrs, element) -> { + for (AnnotationEnhancer enh : enhancers) { + attrs = enh.apply(attrs, element); + } + return attrs; + }; + } + } + } + + @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, + "BeanFactory must be set to find endpoint registry by bean name"); + this.endpointRegistry = this.beanFactory.getBean( + PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, + ReactivePulsarListenerEndpointRegistry.class); + } + this.registrar.setEndpointRegistry(this.endpointRegistry); + } + + if (this.defaultContainerFactoryBeanName != null) { + this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName); + } + + // 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); + } + else { + addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService); + } + + // Actually register all listeners + this.registrar.afterPropertiesSet(); + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @Override + public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { + if (!this.nonAnnotatedClasses.contains(bean.getClass())) { + Class targetClass = AopUtils.getTargetClass(bean); + Map> annotatedMethods = MethodIntrospector.selectMethods(targetClass, + (MethodIntrospector.MetadataLookup>) method -> { + Set listenerMethods = findListenerAnnotations(method); + return (!listenerMethods.isEmpty() ? listenerMethods : null); + }); + if (annotatedMethods.isEmpty()) { + this.nonAnnotatedClasses.add(bean.getClass()); + this.logger.trace(() -> "No @PulsarListener annotations found on bean type: " + bean.getClass()); + } + else { + // Non-empty set of methods + for (Map.Entry> entry : annotatedMethods.entrySet()) { + Method method = entry.getKey(); + for (ReactivePulsarListener listener : entry.getValue()) { + processReactivePulsarListener(listener, method, bean, beanName); + } + } + this.logger.debug(() -> annotatedMethods.size() + " @ReactivePulsarListener methods processed on bean '" + + beanName + "': " + annotatedMethods); + } + } + return bean; + } + + protected void processReactivePulsarListener(ReactivePulsarListener reactivePulsarListener, Method method, + Object bean, String beanName) { + Method methodToUse = checkProxy(method, bean); + MethodReactivePulsarListenerEndpoint endpoint = new MethodReactivePulsarListenerEndpoint<>(); + endpoint.setMethod(methodToUse); + + String beanRef = reactivePulsarListener.beanRef(); + this.listenerScope.addListener(beanRef, bean); + String[] topics = resolveTopics(reactivePulsarListener); + String topicPattern = getTopicPattern(reactivePulsarListener); + processListener(endpoint, reactivePulsarListener, bean, beanName, topics, topicPattern); + this.listenerScope.removeListener(beanRef); + } + + protected void processListener(MethodReactivePulsarListenerEndpoint endpoint, + ReactivePulsarListener ReactivePulsarListener, Object bean, String beanName, String[] topics, + String topicPattern) { + + processReactivePulsarListenerAnnotation(endpoint, ReactivePulsarListener, bean, topics, topicPattern); + + String containerFactory = resolve(ReactivePulsarListener.containerFactory()); + ReactivePulsarListenerContainerFactory listenerContainerFactory = resolveContainerFactory( + ReactivePulsarListener, containerFactory, beanName); + + this.registrar.registerEndpoint(endpoint, listenerContainerFactory); + } + + @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; + } + String containerFactoryBeanName = resolveExpressionAsString(containerFactory, "containerFactory"); + if (StringUtils.hasText(containerFactoryBeanName)) { + assertBeanFactory(); + try { + factory = this.beanFactory.getBean(containerFactoryBeanName, + ReactivePulsarListenerContainerFactory.class); + } + catch (NoSuchBeanDefinitionException ex) { + throw new BeanInitializationException(noBeanFoundMessage(factoryTarget, beanName, + containerFactoryBeanName, ReactivePulsarListenerContainerFactory.class), ex); + } + } + return factory; + } + + protected void assertBeanFactory() { + Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name"); + } + + protected String noBeanFoundMessage(Object target, String listenerBeanName, String requestedBeanName, + Class expectedClass) { + + return "Could not register Pulsar listener endpoint on [" + target + "] for bean " + listenerBeanName + ", no '" + + expectedClass.getSimpleName() + "' with id '" + requestedBeanName + + "' was found in the application context"; + } + + private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListenerEndpoint endpoint, + ReactivePulsarListener reactivePulsarListener, Object bean, String[] topics, String topicPattern) { + + endpoint.setBean(bean); + endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); + endpoint.setSubscriptionName(getEndpointSubscriptionName(reactivePulsarListener)); + endpoint.setId(getEndpointId(reactivePulsarListener)); + endpoint.setTopics(topics); + endpoint.setTopicPattern(topicPattern); + endpoint.setSubscriptionType(reactivePulsarListener.subscriptionType()); + endpoint.setSchemaType(reactivePulsarListener.schemaType()); + + String concurrency = reactivePulsarListener.concurrency(); + if (StringUtils.hasText(concurrency)) { + endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency")); + } + + 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); + } + + private void resolveDeadLetterPolicy(MethodReactivePulsarListenerEndpoint endpoint, + ReactivePulsarListener reactivePulsarListener) { + Object deadLetterPolicy = resolveExpression(reactivePulsarListener.deadLetterPolicy()); + if (deadLetterPolicy instanceof DeadLetterPolicy) { + endpoint.setDeadLetterPolicy((DeadLetterPolicy) deadLetterPolicy); + } + else { + String deadLetterPolicyBeanName = resolveExpressionAsString(reactivePulsarListener.deadLetterPolicy(), + "deadLetterPolicy"); + if (StringUtils.hasText(deadLetterPolicyBeanName)) { + endpoint.setDeadLetterPolicy( + this.beanFactory.getBean(deadLetterPolicyBeanName, DeadLetterPolicy.class)); + } + } + } + + @SuppressWarnings("unchecked") + private void resolveConsumerCustomizer(MethodReactivePulsarListenerEndpoint endpoint, + ReactivePulsarListener reactivePulsarListener) { + Object customizer = resolveExpression(reactivePulsarListener.consumerCustomizer()); + if (customizer instanceof ReactiveMessageConsumerBuilderCustomizer) { + endpoint.setConsumerCustomizer((ReactiveMessageConsumerBuilderCustomizer) customizer); + } + else { + String consumerCustomizerBeanName = resolveExpressionAsString(reactivePulsarListener.consumerCustomizer(), + "consumerCustomizer"); + if (StringUtils.hasText(consumerCustomizerBeanName)) { + endpoint.setConsumerCustomizer(this.beanFactory.getBean(consumerCustomizerBeanName, + ReactiveMessageConsumerBuilderCustomizer.class)); + } + } + } + + private Integer resolveExpressionAsInteger(String value, String attribute) { + Object resolved = resolveExpression(value); + Integer result = null; + if (resolved instanceof String) { + result = Integer.parseInt((String) resolved); + } + else if (resolved instanceof Number) { + result = ((Number) resolved).intValue(); + } + else if (resolved != null) { + throw new IllegalStateException( + THE_LEFT + attribute + "] must resolve to an Number or a String that can be parsed as an Integer. " + + RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]"); + } + return result; + } + + private Boolean resolveExpressionAsBoolean(String value, String attribute) { + Object resolved = resolveExpression(value); + Boolean result = null; + if (resolved instanceof Boolean) { + result = (Boolean) resolved; + } + else if (resolved instanceof String) { + result = Boolean.parseBoolean((String) resolved); + } + else if (resolved != null) { + throw new IllegalStateException( + THE_LEFT + attribute + "] must resolve to a Boolean or a String that can be parsed as a Boolean. " + + RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]"); + } + return result; + } + + private void loadProperty(Properties properties, String property, Object value) { + try { + properties.load(new StringReader((String) value)); + } + catch (IOException e) { + this.logger.error(e, () -> "Failed to load property " + property + ", continuing..."); + } + } + + private String getEndpointSubscriptionName(ReactivePulsarListener reactivePulsarListener) { + if (StringUtils.hasText(reactivePulsarListener.subscriptionName())) { + return resolveExpressionAsString(reactivePulsarListener.subscriptionName(), "subscriptionName"); + } + return GENERATED_ID_PREFIX + this.counter.getAndIncrement(); + } + + private String getEndpointId(ReactivePulsarListener reactivePulsarListener) { + if (StringUtils.hasText(reactivePulsarListener.id())) { + return resolveExpressionAsString(reactivePulsarListener.id(), "id"); + } + return GENERATED_ID_PREFIX + this.counter.getAndIncrement(); + } + + private String getTopicPattern(ReactivePulsarListener reactivePulsarListener) { + return resolveExpressionAsString(reactivePulsarListener.topicPattern(), "topicPattern"); + } + + private String resolveExpressionAsString(String value, String attribute) { + Object resolved = resolveExpression(value); + if (resolved instanceof String) { + return (String) resolved; + } + else if (resolved != null) { + throw new IllegalStateException(THE_LEFT + attribute + "] must resolve to a String. " + RESOLVED_TO_LEFT + + resolved.getClass() + RIGHT_FOR_LEFT + value + "]"); + } + return null; + } + + private String[] resolveTopics(ReactivePulsarListener ReactivePulsarListener) { + String[] topics = ReactivePulsarListener.topics(); + List result = new ArrayList<>(); + if (topics.length > 0) { + for (String topic1 : topics) { + Object topic = resolveExpression(topic1); + resolveAsString(topic, result); + } + } + return result.toArray(new String[0]); + } + + private Object resolveExpression(String value) { + return this.resolver.evaluate(resolve(value), this.expressionContext); + } + + private String resolve(String value) { + if (this.beanFactory != null && this.beanFactory instanceof ConfigurableBeanFactory) { + return ((ConfigurableBeanFactory) this.beanFactory).resolveEmbeddedValue(value); + } + return value; + } + + @SuppressWarnings("unchecked") + private void resolveAsString(Object resolvedValue, List result) { + if (resolvedValue instanceof String[]) { + for (Object object : (String[]) resolvedValue) { + resolveAsString(object, result); + } + } + else if (resolvedValue instanceof String) { + result.add((String) resolvedValue); + } + else if (resolvedValue instanceof Iterable) { + for (Object object : (Iterable) resolvedValue) { + resolveAsString(object, result); + } + } + else { + throw new IllegalArgumentException( + String.format("@ReactivePulsarListener can't resolve '%s' as a String", resolvedValue)); + } + } + + private Method checkProxy(Method methodArg, Object bean) { + Method method = methodArg; + if (AopUtils.isJdkDynamicProxy(bean)) { + try { + // Found a @ReactivePulsarListener method on the target class for this JDK + // proxy + // -> + // is it also present on the proxy itself? + method = bean.getClass().getMethod(method.getName(), method.getParameterTypes()); + Class[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces(); + for (Class iface : proxiedInterfaces) { + try { + method = iface.getMethod(method.getName(), method.getParameterTypes()); + break; + } + catch (@SuppressWarnings("unused") NoSuchMethodException noMethod) { + // NOSONAR + } + } + } + catch (SecurityException ex) { + ReflectionUtils.handleReflectionException(ex); + } + catch (NoSuchMethodException ex) { + throw new IllegalStateException(String.format( + "@ReactivePulsarListener method '%s' found on bean target class '%s', " + + "but not found in any interface(s) for bean JDK proxy. Either " + + "pull the method up to an interface or switch to subclass (CGLIB) " + + "proxies by setting proxy-target-class/proxyTargetClass " + "attribute to 'true'", + method.getName(), method.getDeclaringClass().getSimpleName()), ex); + } + } + return method; + } + + private Collection findListenerAnnotations(Class clazz) { + Set listeners = new HashSet<>(); + ReactivePulsarListener ann = AnnotatedElementUtils.findMergedAnnotation(clazz, ReactivePulsarListener.class); + if (ann != null) { + ann = enhance(clazz, ann); + listeners.add(ann); + } + ReactivePulsarListeners anns = AnnotationUtils.findAnnotation(clazz, ReactivePulsarListeners.class); + if (anns != null) { + listeners.addAll(Arrays.stream(anns.value()).map(anno -> enhance(clazz, anno)).toList()); + } + return listeners; + } + + private Set findListenerAnnotations(Method method) { + Set listeners = new HashSet<>(); + ReactivePulsarListener ann = AnnotatedElementUtils.findMergedAnnotation(method, ReactivePulsarListener.class); + if (ann != null) { + ann = enhance(method, ann); + listeners.add(ann); + } + ReactivePulsarListeners anns = AnnotationUtils.findAnnotation(method, ReactivePulsarListeners.class); + if (anns != null) { + listeners.addAll(Arrays.stream(anns.value()).map(anno -> enhance(method, anno)).toList()); + } + return listeners; + } + + private ReactivePulsarListener enhance(AnnotatedElement element, ReactivePulsarListener ann) { + if (this.enhancer == null) { + return ann; + } + return AnnotationUtils.synthesizeAnnotation( + this.enhancer.apply(AnnotationUtils.getAnnotationAttributes(ann), element), + ReactivePulsarListener.class, null); + } + + private void addFormatters(FormatterRegistry registry) { + 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 + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + if (applicationContext instanceof ConfigurableApplicationContext) { + setBeanFactory(((ConfigurableApplicationContext) applicationContext).getBeanFactory()); + } + else { + setBeanFactory(applicationContext); + } + } + + public void setBeanFactory(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + if (beanFactory instanceof ConfigurableListableBeanFactory) { + this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver(); + this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory, + this.listenerScope); + } + } + + private class PulsarHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory { + + private final DefaultFormattingConversionService defaultFormattingConversionService = new DefaultFormattingConversionService(); + + private MessageHandlerMethodFactory handlerMethodFactory; + + public void setHandlerMethodFactory(MessageHandlerMethodFactory pulsarHandlerMethodFactory1) { + this.handlerMethodFactory = pulsarHandlerMethodFactory1; + } + + @Override + public InvocableHandlerMethod createInvocableHandlerMethod(Object bean, Method method) { + return getHandlerMethodFactory().createInvocableHandlerMethod(bean, method); + } + + private MessageHandlerMethodFactory getHandlerMethodFactory() { + if (this.handlerMethodFactory == null) { + this.handlerMethodFactory = createDefaultMessageHandlerMethodFactory(); + } + return this.handlerMethodFactory; + } + + private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() { + DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory(); + Validator validator = ReactivePulsarListenerAnnotationBeanPostProcessor.this.registrar.getValidator(); + if (validator != null) { + defaultFactory.setValidator(validator); + } + defaultFactory.setBeanFactory(ReactivePulsarListenerAnnotationBeanPostProcessor.this.beanFactory); + this.defaultFormattingConversionService.addConverter( + new BytesToStringConverter(ReactivePulsarListenerAnnotationBeanPostProcessor.this.charset)); + this.defaultFormattingConversionService.addConverter(new BytesToNumberConverter()); + defaultFactory.setConversionService(this.defaultFormattingConversionService); + GenericMessageConverter messageConverter = new GenericMessageConverter( + this.defaultFormattingConversionService); + defaultFactory.setMessageConverter(messageConverter); + + List customArgumentsResolver = new ArrayList<>( + ReactivePulsarListenerAnnotationBeanPostProcessor.this.registrar + .getCustomMethodArgumentResolvers()); + // Has to be at the end - look at PayloadMethodArgumentResolver documentation + // customArgumentsResolver.add(new + // PulsarNullAwarePayloadArgumentResolver(messageConverter, validator)); + defaultFactory.setCustomArgumentResolvers(customArgumentsResolver); + + defaultFactory.afterPropertiesSet(); + + return defaultFactory; + } + + } + + private static class BytesToStringConverter implements Converter { + + private final Charset charset; + + BytesToStringConverter(Charset charset) { + this.charset = charset; + } + + @Override + public String convert(byte[] source) { + return new String(source, this.charset); + } + + } + + private final class BytesToNumberConverter implements ConditionalGenericConverter { + + BytesToNumberConverter() { + } + + @Override + @Nullable + public Set getConvertibleTypes() { + HashSet pairs = new HashSet<>(); + pairs.add(new ConvertiblePair(byte[].class, long.class)); + pairs.add(new ConvertiblePair(byte[].class, int.class)); + pairs.add(new ConvertiblePair(byte[].class, short.class)); + pairs.add(new ConvertiblePair(byte[].class, byte.class)); + pairs.add(new ConvertiblePair(byte[].class, Long.class)); + pairs.add(new ConvertiblePair(byte[].class, Integer.class)); + pairs.add(new ConvertiblePair(byte[].class, Short.class)); + pairs.add(new ConvertiblePair(byte[].class, Byte.class)); + return pairs; + } + + @Override + @Nullable + public Object convert(@Nullable Object source, TypeDescriptor sourceType, TypeDescriptor targetType) { + byte[] bytes = (byte[]) source; + if (targetType.getType().equals(long.class) || targetType.getType().equals(Long.class)) { + Assert.state(bytes.length >= 8, "At least 8 bytes needed to convert a byte[] to a long"); // NOSONAR + return ByteBuffer.wrap(bytes).getLong(); + } + else if (targetType.getType().equals(int.class) || targetType.getType().equals(Integer.class)) { + Assert.state(bytes.length >= 4, "At least 4 bytes needed to convert a byte[] to an integer"); // NOSONAR + return ByteBuffer.wrap(bytes).getInt(); + } + else if (targetType.getType().equals(short.class) || targetType.getType().equals(Short.class)) { + Assert.state(bytes.length >= 2, "At least 2 bytes needed to convert a byte[] to a short"); + return ByteBuffer.wrap(bytes).getShort(); + } + else if (targetType.getType().equals(byte.class) || targetType.getType().equals(Byte.class)) { + Assert.state(bytes.length >= 1, "At least 1 byte needed to convert a byte[] to a byte"); + return ByteBuffer.wrap(bytes).get(); + } + return null; + } + + @Override + public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) { + if (sourceType.getType().equals(byte[].class)) { + Class target = targetType.getType(); + return target.equals(long.class) || target.equals(int.class) || target.equals(short.class) // NOSONAR + || target.equals(byte.class) || target.equals(Long.class) || target.equals(Integer.class) + || target.equals(Short.class) || target.equals(Byte.class); + } + return false; + } + + } + + static class ListenerScope implements Scope { + + private final Map listeners = new HashMap<>(); + + ListenerScope() { + } + + public void addListener(String key, Object bean) { + this.listeners.put(key, bean); + } + + public void removeListener(String key) { + this.listeners.remove(key); + } + + @Override + public Object get(String name, ObjectFactory objectFactory) { + return this.listeners.get(name); + } + + @Override + public Object remove(String name) { + return null; + } + + @Override + public void registerDestructionCallback(String name, Runnable callback) { + } + + @Override + public Object resolveContextualObject(String key) { + return this.listeners.get(key); + } + + @Override + public String getConversationId() { + return null; + } + + } + + public interface AnnotationEnhancer extends BiFunction, AnnotatedElement, Map> { + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java new file mode 100644 index 00000000..0afdef60 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java @@ -0,0 +1,45 @@ +/* + * Copyright 2022 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 java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Container annotation that aggregates several {@link ReactivePulsarListener} + * annotations. + *

+ * Can be used natively, declaring several nested {@link ReactivePulsarListener} + * annotations. Can also be used in conjunction with Java 8's support for repeatable + * annotations, where {@link ReactivePulsarListener} can simply be declared several times + * on the same method (or class), implicitly generating this container annotation. + * + * @author Christophe Bornet + * + * @see ReactivePulsarListener + */ +@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) +@Retention(RetentionPolicy.RUNTIME) +@Documented +public @interface ReactivePulsarListeners { + + ReactivePulsarListener[] value(); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index 038423d2..ff643be6 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -43,7 +43,7 @@ import io.micrometer.observation.ObservationRegistry; * @author Chris Bono */ public abstract class AbstractPulsarListenerContainerFactory, T> - implements PulsarListenerContainerFactory, ApplicationEventPublisherAware, ApplicationContextAware { + implements PulsarListenerContainerFactory, ApplicationEventPublisherAware, ApplicationContextAware { protected final LogAccessor logger = new LogAccessor(this.getClass()); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java index 23055084..a85552f3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java @@ -22,7 +22,6 @@ import java.util.Collection; import java.util.Collections; import java.util.Properties; -import org.apache.commons.logging.LogFactory; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; @@ -34,7 +33,6 @@ import org.springframework.beans.factory.config.BeanExpressionContext; import org.springframework.beans.factory.config.BeanExpressionResolver; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.expression.BeanFactoryResolver; -import org.springframework.core.log.LogAccessor; import org.springframework.expression.BeanResolver; import org.springframework.lang.Nullable; import org.springframework.pulsar.listener.AckMode; @@ -54,8 +52,6 @@ import org.springframework.util.StringUtils; public abstract class AbstractPulsarListenerEndpoint implements PulsarListenerEndpoint, BeanFactoryAware, InitializingBean { - private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); - private String subscriptionName; private SubscriptionType subscriptionType; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java new file mode 100644 index 00000000..deb8249d --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java @@ -0,0 +1,51 @@ +/* + * Copyright 2022 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.PulsarListener; +import org.springframework.pulsar.annotation.ReactivePulsarListener; +import org.springframework.pulsar.listener.MessageListenerContainer; + +/** + * Factory for Pulsar message listener containers. + * + * @param message listener container type. + * @param listener endpoint type. + * @author Soby Chacko + * @author Christophe Bornet + */ +public interface ListenerContainerFactory> { + + /** + * Create a {@link MessageListenerContainer} for the given {@link ListenerEndpoint}. + * Containers created using this method are added to the listener endpoint registry. + * @param endpoint the endpoint to configure + * @return the created container + */ + C createListenerContainer(E endpoint); + + /** + * Create and configure a container without a listener; used to create containers that + * are not used for {@link PulsarListener} and {@link ReactivePulsarListener} + * annotations. Containers created using this method are not added to the listener + * endpoint registry. + * @param topics the topics. + * @return the container. + */ + C createContainer(String... topics); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java new file mode 100644 index 00000000..73ee48dc --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java @@ -0,0 +1,106 @@ +/* + * Copyright 2022 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 java.util.Collection; + +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; + +import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.MessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; + +/** + * Model for a Pulsar listener endpoint. Can be used against a + * {@link org.springframework.pulsar.annotation.PulsarListenerConfigurer} to register + * endpoints programmatically. + * + * @param Message listener container type. + * @author Christophe Bornet + */ +public interface ListenerEndpoint { + + /** + * Return the id of this endpoint. + * @return the id of this endpoint. The id can be further qualified when the endpoint + * is resolved against its actual listener container. + * @see ListenerContainerFactory#createListenerContainer + */ + @Nullable + String getId(); + + /** + * Return the subscription name for this endpoint's container. + * @return the subscription name. + */ + @Nullable + String getSubscriptionName(); + + /** + * Return the subscription type for this endpoint's container. + * @return the subscription type. + */ + @Nullable + SubscriptionType getSubscriptionType(); + + /** + * Return the topics for this endpoint's container. + * @return the topics. + */ + Collection getTopics(); + + /** + * Return the topic pattern for this endpoint's container. + * @return the topic pattern. + */ + String getTopicPattern(); + + /** + * Return the autoStartup for this endpoint's container. + * @return the autoStartup. + */ + @Nullable + Boolean getAutoStartup(); + + /** + * Return the schema type for this endpoint's container. + * @return the schema type. + */ + SchemaType getSchemaType(); + + /** + * Return the concurrency for this endpoint's container. + * @return the concurrency. + */ + @Nullable + Integer getConcurrency(); + + /** + * Setup the specified message listener container with the model defined by this + * endpoint. + *

+ * This endpoint must provide the requested missing option(s) of the specified + * container to make it usable. Usually, this is about setting the {@code queues} and + * the {@code messageListener} to use but an implementation may override any default + * setting that was already set. + * @param listenerContainer the listener container to configure + * @param messageConverter the message converter - can be null + */ + void setupListenerContainer(C listenerContainer, @Nullable MessageConverter messageConverter); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpointRegistry.java new file mode 100644 index 00000000..bbd399f7 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpointRegistry.java @@ -0,0 +1,257 @@ +/* + * Copyright 2022 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 java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationListener; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.SmartLifecycle; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.ListenerContainerRegistry; +import org.springframework.pulsar.listener.MessageListenerContainer; +import org.springframework.util.Assert; + +/** + * Creates the necessary container instances for the registered + * {@linkplain ListenerEndpoint endpoints}. Also manages the lifecycle of the listener + * containers, in particular within the lifecycle of the application context. + * + *

+ * Contrary to containers created manually, listener containers managed by registry are + * not beans in the application context and are not candidates for autowiring. Use + * {@link #getListenerContainers()} if you need to access this registry's listener + * containers for management purposes. If you need to access to a specific message + * listener container, use {@link #getListenerContainer(String)} with the id of the + * endpoint. + * + * @param listener container type. + * @param listener endpoint type. + * @author Soby Chacko + * @author Christophe Bornet + */ +public class ListenerEndpointRegistry> + implements ListenerContainerRegistry, DisposableBean, SmartLifecycle, ApplicationContextAware, + ApplicationListener { + + private final Class type; + + private final Map listenerContainers = new ConcurrentHashMap<>(); + + private ConfigurableApplicationContext applicationContext; + + private int phase = C.DEFAULT_PHASE; + + private boolean contextRefreshed; + + private volatile boolean running; + + @SuppressWarnings("unchecked") + protected ListenerEndpointRegistry(Class type) { + this.type = (Class) type; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + if (applicationContext instanceof ConfigurableApplicationContext) { + this.applicationContext = (ConfigurableApplicationContext) applicationContext; + } + } + + @Override + @Nullable + public C getListenerContainer(String id) { + Assert.hasText(id, "Container identifier must not be empty"); + return this.listenerContainers.get(id); + } + + @Override + public Set getListenerContainerIds() { + return Collections.unmodifiableSet(this.listenerContainers.keySet()); + } + + @Override + public Collection getListenerContainers() { + return Collections.unmodifiableCollection(this.listenerContainers.values()); + } + + @Override + public Collection getAllListenerContainers() { + List containers = new ArrayList<>(getListenerContainers()); + containers.addAll(this.applicationContext.getBeansOfType(this.type, true, false).values()); + return containers; + } + + public void registerListenerContainer(E endpoint, ListenerContainerFactory factory) { + registerListenerContainer(endpoint, factory, false); + } + + public void registerListenerContainer(E endpoint, ListenerContainerFactory factory, + boolean startImmediately) { + Assert.notNull(endpoint, "Endpoint must not be null"); + Assert.notNull(factory, "Factory must not be null"); + + String subscriptionName = endpoint.getSubscriptionName(); + String id = endpoint.getId(); + + Assert.hasText(subscriptionName, "Endpoint id must not be empty"); + + synchronized (this.listenerContainers) { + Assert.state(!this.listenerContainers.containsKey(id), + "Another endpoint is already registered with id '" + subscriptionName + "'"); + C container = createListenerContainer(endpoint, factory); + this.listenerContainers.put(id, container); + } + } + + protected C createListenerContainer(E endpoint, ListenerContainerFactory factory) { + + C listenerContainer = factory.createListenerContainer(endpoint); + + if (listenerContainer instanceof InitializingBean) { + try { + ((InitializingBean) listenerContainer).afterPropertiesSet(); + } + catch (Exception ex) { + throw new BeanInitializationException("Failed to initialize message listener container", ex); + } + } + + int containerPhase = listenerContainer.getPhase(); + if (listenerContainer.isAutoStartup() && containerPhase != C.DEFAULT_PHASE) { // a + // custom + // phase + // value + if (this.phase != C.DEFAULT_PHASE && this.phase != containerPhase) { + throw new IllegalStateException("Encountered phase mismatch between container " + + "factory definitions: " + this.phase + " vs " + containerPhase); + } + this.phase = listenerContainer.getPhase(); + } + + return listenerContainer; + } + + @Override + public void destroy() throws Exception { + for (C listenerContainer : getListenerContainers()) { + listenerContainer.destroy(); + } + } + + // Delegating implementation of SmartLifecycle + + @Override + public int getPhase() { + return this.phase; + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public void start() { + for (C listenerContainer : getListenerContainers()) { + startIfNecessary(listenerContainer); + } + this.running = true; + } + + @Override + public void stop() { + this.running = false; + for (C listenerContainer : getListenerContainers()) { + listenerContainer.stop(); + } + } + + @Override + public void stop(Runnable callback) { + this.running = false; + Collection listenerContainersToStop = getListenerContainers(); + if (listenerContainersToStop.size() > 0) { + AggregatingCallback aggregatingCallback = new AggregatingCallback(listenerContainersToStop.size(), + callback); + for (C listenerContainer : listenerContainersToStop) { + if (listenerContainer.isRunning()) { + listenerContainer.stop(aggregatingCallback); + } + else { + aggregatingCallback.run(); + } + } + } + else { + callback.run(); + } + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getApplicationContext().equals(this.applicationContext)) { + this.contextRefreshed = true; + } + } + + private void startIfNecessary(C listenerContainer) { + if (this.contextRefreshed || listenerContainer.isAutoStartup()) { + listenerContainer.start(); + } + } + + private static final class AggregatingCallback implements Runnable { + + private final AtomicInteger count; + + private final Runnable finishCallback; + + private AggregatingCallback(int count, Runnable finishCallback) { + this.count = new AtomicInteger(count); + this.finishCallback = finishCallback; + } + + @Override + public void run() { + if (this.count.decrementAndGet() <= 0) { + this.finishCallback.run(); + } + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java index 08d2be55..f0fbefd9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java @@ -34,4 +34,14 @@ public abstract class PulsarListenerBeanNames { */ public static final String PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME = "org.springframework.pulsar.config.internalPulsarListenerEndpointRegistry"; + /** + * The bean name of the internally managed Pulsar listener annotation processor. + */ + public static final String REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME = "org.springframework.pulsar.config.internalReactivePulsarListenerAnnotationProcessor"; + + /** + * The bean name of the internally managed Pulsar listener endpoint registry. + */ + public static final String REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME = "org.springframework.pulsar.config.internalReactivePulsarListenerEndpointRegistry"; + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java index 730af85f..329c105d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java @@ -21,13 +21,10 @@ import org.springframework.pulsar.listener.PulsarMessageListenerContainer; /** * Factory for Pulsar message listener containers. * - * @param message listener container type. * @author Soby Chacko + * @author Christophe Bornet */ -public interface PulsarListenerContainerFactory { - - C createListenerContainer(PulsarListenerEndpoint endpoint); - - C createContainer(String... topics); +public interface PulsarListenerContainerFactory + extends ListenerContainerFactory { } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java index a4676300..9d735a84 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java @@ -16,16 +16,10 @@ package org.springframework.pulsar.config; -import java.util.Collection; import java.util.Properties; -import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.common.schema.SchemaType; - -import org.springframework.lang.Nullable; import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; -import org.springframework.pulsar.support.MessageConverter; /** * Model for a Pulsar listener endpoint. Can be used against a @@ -35,36 +29,12 @@ import org.springframework.pulsar.support.MessageConverter; * @author Soby Chacko * @author Alexander Preuß */ -public interface PulsarListenerEndpoint { - - @Nullable - String getId(); - - @Nullable - String getSubscriptionName(); - - @Nullable - SubscriptionType getSubscriptionType(); - - Collection getTopics(); - - String getTopicPattern(); - - @Nullable - Boolean getAutoStartup(); - - void setupListenerContainer(PulsarMessageListenerContainer listenerContainer, - @Nullable MessageConverter messageConverter); +public interface PulsarListenerEndpoint extends ListenerEndpoint { boolean isBatchListener(); - SchemaType getSchemaType(); - Properties getConsumerProperties(); - @Nullable - Integer getConcurrency(); - AckMode getAckMode(); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java index 35a2aac1..e5e841ad 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java @@ -31,16 +31,20 @@ import org.springframework.util.Assert; import org.springframework.validation.Validator; /** - * Helper bean for registering {@link PulsarListenerEndpoint} with a - * {@link PulsarListenerEndpointRegistry}. + * Helper bean for registering {@link ListenerEndpoint} with a + * {@link ListenerEndpointRegistry}. * * @author Soby Chacko + * @author Christophe Bornet */ +@SuppressWarnings({ "rawtypes", "unchecked" }) public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, InitializingBean { + private final Class type; + private final List endpointDescriptors = new ArrayList<>(); - private PulsarListenerEndpointRegistry endpointRegistry; + private ListenerEndpointRegistry endpointRegistry; private List customMethodArgumentResolvers = new ArrayList<>(); @@ -48,7 +52,7 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia private MessageHandlerMethodFactory messageHandlerMethodFactory; - private PulsarListenerContainerFactory containerFactory; + private ListenerContainerFactory containerFactory; private String containerFactoryBeanName; @@ -56,12 +60,16 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia private boolean startImmediately; - public void setEndpointRegistry(PulsarListenerEndpointRegistry endpointRegistry) { + public PulsarListenerEndpointRegistrar(Class type) { + this.type = type; + } + + public void setEndpointRegistry(ListenerEndpointRegistry endpointRegistry) { this.endpointRegistry = endpointRegistry; } @Nullable - public PulsarListenerEndpointRegistry getEndpointRegistry() { + public ListenerEndpointRegistry getEndpointRegistry() { return this.endpointRegistry; } @@ -83,7 +91,7 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia return this.messageHandlerMethodFactory; } - public void setContainerFactory(PulsarListenerContainerFactory containerFactory) { + public void setContainerFactory(ListenerContainerFactory containerFactory) { this.containerFactory = containerFactory; } @@ -115,14 +123,14 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia protected void registerAllEndpoints() { synchronized (this.endpointDescriptors) { for (PulsarListenerEndpointDescriptor descriptor : this.endpointDescriptors) { - this.endpointRegistry.registerListenerContainer(descriptor.endpoint, - resolveContainerFactory(descriptor)); + ListenerContainerFactory factory = resolveContainerFactory(descriptor); + this.endpointRegistry.registerListenerContainer(descriptor.endpoint, factory); } this.startImmediately = true; // trigger immediate startup } } - private PulsarListenerContainerFactory resolveContainerFactory(PulsarListenerEndpointDescriptor descriptor) { + private ListenerContainerFactory resolveContainerFactory(PulsarListenerEndpointDescriptor descriptor) { if (descriptor.containerFactory != null) { return descriptor.containerFactory; } @@ -131,19 +139,17 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia } else if (this.containerFactoryBeanName != null) { Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name"); - this.containerFactory = this.beanFactory.getBean(this.containerFactoryBeanName, - PulsarListenerContainerFactory.class); + this.containerFactory = this.beanFactory.getBean(this.containerFactoryBeanName, this.type); return this.containerFactory; // Consider changing this if live change of the - // factory is required + // factory is required } else { - throw new IllegalStateException( - "Could not resolve the " + PulsarListenerContainerFactory.class.getSimpleName() + " to use for [" - + descriptor.endpoint + "] no factory was given and no default is set."); + throw new IllegalStateException("Could not resolve the " + ListenerContainerFactory.class.getSimpleName() + + " to use for [" + descriptor.endpoint + "] no factory was given and no default is set."); } } - public void registerEndpoint(PulsarListenerEndpoint endpoint, @Nullable PulsarListenerContainerFactory factory) { + public void registerEndpoint(ListenerEndpoint endpoint, @Nullable ListenerContainerFactory factory) { Assert.notNull(endpoint, "Endpoint must be set"); Assert.hasText(endpoint.getSubscriptionName(), "Endpoint id must be set"); // Factory may be null, we defer the resolution right before actually creating the @@ -162,12 +168,12 @@ public class PulsarListenerEndpointRegistrar implements BeanFactoryAware, Initia private static final class PulsarListenerEndpointDescriptor { - private final PulsarListenerEndpoint endpoint; + private final ListenerEndpoint endpoint; - private final PulsarListenerContainerFactory containerFactory; + private final ListenerContainerFactory containerFactory; - private PulsarListenerEndpointDescriptor(PulsarListenerEndpoint endpoint, - @Nullable PulsarListenerContainerFactory containerFactory) { + private PulsarListenerEndpointDescriptor(ListenerEndpoint endpoint, + @Nullable ListenerContainerFactory containerFactory) { this.endpoint = endpoint; this.containerFactory = containerFactory; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java index 543a09b4..4fc32146 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java @@ -16,30 +16,7 @@ package org.springframework.pulsar.config; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicInteger; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanInitializationException; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ApplicationListener; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.SmartLifecycle; -import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.lang.Nullable; -import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer; -import org.springframework.pulsar.listener.PulsarListenerContainerRegistry; import org.springframework.pulsar.listener.PulsarMessageListenerContainer; -import org.springframework.util.Assert; /** * Creates the necessary {@link PulsarMessageListenerContainer} instances for the @@ -55,197 +32,13 @@ import org.springframework.util.Assert; * id of the endpoint. * * @author Soby Chacko + * @author Christophe Bornet */ -public class PulsarListenerEndpointRegistry implements PulsarListenerContainerRegistry, DisposableBean, SmartLifecycle, - ApplicationContextAware, ApplicationListener { - - private final Map listenerContainers = new ConcurrentHashMap<>(); - - private ConfigurableApplicationContext applicationContext; - - private int phase = AbstractPulsarMessageListenerContainer.DEFAULT_PHASE; - - private boolean contextRefreshed; - - private volatile boolean running; - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - if (applicationContext instanceof ConfigurableApplicationContext) { - this.applicationContext = (ConfigurableApplicationContext) applicationContext; - } - } - - @Override - @Nullable - public PulsarMessageListenerContainer getListenerContainer(String id) { - Assert.hasText(id, "Container identifier must not be empty"); - return this.listenerContainers.get(id); - } - - @Override - public Set getListenerContainerIds() { - return Collections.unmodifiableSet(this.listenerContainers.keySet()); - } - - @Override - public Collection getListenerContainers() { - return Collections.unmodifiableCollection(this.listenerContainers.values()); - } - - @Override - public Collection getAllListenerContainers() { - List containers = new ArrayList<>(); - containers.addAll(getListenerContainers()); - containers.addAll( - this.applicationContext.getBeansOfType(PulsarMessageListenerContainer.class, true, false).values()); - return containers; - } - - public void registerListenerContainer(PulsarListenerEndpoint endpoint, PulsarListenerContainerFactory factory) { - registerListenerContainer(endpoint, factory, false); - } - - public void registerListenerContainer(PulsarListenerEndpoint endpoint, PulsarListenerContainerFactory factory, - boolean startImmediately) { - Assert.notNull(endpoint, "Endpoint must not be null"); - Assert.notNull(factory, "Factory must not be null"); - - String subscriptionName = endpoint.getSubscriptionName(); - String id = endpoint.getId(); - - Assert.hasText(subscriptionName, "Endpoint id must not be empty"); - - synchronized (this.listenerContainers) { - Assert.state(!this.listenerContainers.containsKey(id), - "Another endpoint is already registered with id '" + subscriptionName + "'"); - PulsarMessageListenerContainer container = createListenerContainer(endpoint, factory); - this.listenerContainers.put(id, container); - ConfigurableApplicationContext appContext = this.applicationContext; - } - } - - protected PulsarMessageListenerContainer createListenerContainer(PulsarListenerEndpoint endpoint, - PulsarListenerContainerFactory factory) { - PulsarMessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint); - - if (listenerContainer instanceof InitializingBean) { - try { - ((InitializingBean) listenerContainer).afterPropertiesSet(); - } - catch (Exception ex) { - throw new BeanInitializationException("Failed to initialize message listener container", ex); - } - } - - int containerPhase = listenerContainer.getPhase(); - if (listenerContainer.isAutoStartup() - && containerPhase != AbstractPulsarMessageListenerContainer.DEFAULT_PHASE) { // a - // custom - // phase - // value - if (this.phase != AbstractPulsarMessageListenerContainer.DEFAULT_PHASE && this.phase != containerPhase) { - throw new IllegalStateException("Encountered phase mismatch between container " - + "factory definitions: " + this.phase + " vs " + containerPhase); - } - this.phase = listenerContainer.getPhase(); - } - - return listenerContainer; - } - - @Override - public void destroy() { - for (PulsarMessageListenerContainer listenerContainer : getListenerContainers()) { - listenerContainer.destroy(); - } - } - - // Delegating implementation of SmartLifecycle - - @Override - public int getPhase() { - return this.phase; - } - - @Override - public boolean isAutoStartup() { - return true; - } - - @Override - public void start() { - for (PulsarMessageListenerContainer listenerContainer : getListenerContainers()) { - startIfNecessary(listenerContainer); - } - this.running = true; - } - - @Override - public void stop() { - this.running = false; - for (PulsarMessageListenerContainer listenerContainer : getListenerContainers()) { - listenerContainer.stop(); - } - } - - @Override - public void stop(Runnable callback) { - this.running = false; - Collection listenerContainersToStop = getListenerContainers(); - if (listenerContainersToStop.size() > 0) { - AggregatingCallback aggregatingCallback = new AggregatingCallback(listenerContainersToStop.size(), - callback); - for (PulsarMessageListenerContainer listenerContainer : listenerContainersToStop) { - if (listenerContainer.isRunning()) { - listenerContainer.stop(aggregatingCallback); - } - else { - aggregatingCallback.run(); - } - } - } - else { - callback.run(); - } - } - - @Override - public boolean isRunning() { - return this.running; - } - - @Override - public void onApplicationEvent(ContextRefreshedEvent event) { - if (event.getApplicationContext().equals(this.applicationContext)) { - this.contextRefreshed = true; - } - } - - private void startIfNecessary(PulsarMessageListenerContainer listenerContainer) { - if (this.contextRefreshed || listenerContainer.isAutoStartup()) { - listenerContainer.start(); - } - } - - private static final class AggregatingCallback implements Runnable { - - private final AtomicInteger count; - - private final Runnable finishCallback; - - private AggregatingCallback(int count, Runnable finishCallback) { - this.count = new AtomicInteger(count); - this.finishCallback = finishCallback; - } - - @Override - public void run() { - if (this.count.decrementAndGet() <= 0) { - this.finishCallback.run(); - } - } +public class PulsarListenerEndpointRegistry + extends ListenerEndpointRegistry { + public PulsarListenerEndpointRegistry() { + super(PulsarMessageListenerContainer.class); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java new file mode 100644 index 00000000..c3687b28 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java @@ -0,0 +1,229 @@ +/* + * Copyright 2022 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.reactive; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.BeanExpressionContext; +import org.springframework.beans.factory.config.BeanExpressionResolver; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.context.expression.BeanFactoryResolver; +import org.springframework.expression.BeanResolver; +import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Base implementation for {@link ReactivePulsarListenerEndpoint}. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public abstract class AbstractReactivePulsarListenerEndpoint + implements ReactivePulsarListenerEndpoint, BeanFactoryAware, InitializingBean { + + private String subscriptionName; + + private SubscriptionType subscriptionType; + + private SchemaType schemaType; + + private String id; + + private Collection topics = new ArrayList<>(); + + private String topicPattern; + + private BeanFactory beanFactory; + + private BeanExpressionResolver resolver; + + private BeanExpressionContext expressionContext; + + private BeanResolver beanResolver; + + private Boolean autoStartup; + + private Boolean fluxListener; + + private Integer concurrency; + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + if (beanFactory instanceof ConfigurableListableBeanFactory) { + this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver(); + this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory, null); + } + this.beanResolver = new BeanFactoryResolver(beanFactory); + } + + @Nullable + protected BeanFactory getBeanFactory() { + return this.beanFactory; + } + + @Override + public void afterPropertiesSet() { + boolean topicsEmpty = getTopics().isEmpty(); + if (!topicsEmpty && !StringUtils.hasText(getTopicPattern())) { + throw new IllegalStateException("Topics or topicPattern must be provided but not both for " + this); + } + } + + @Nullable + protected BeanExpressionResolver getResolver() { + return this.resolver; + } + + @Nullable + protected BeanExpressionContext getBeanExpressionContext() { + return this.expressionContext; + } + + @Nullable + protected BeanResolver getBeanResolver() { + return this.beanResolver; + } + + public void setSubscriptionName(String subscriptionName) { + + this.subscriptionName = subscriptionName; + } + + @Nullable + @Override + public String getSubscriptionName() { + return this.subscriptionName; + } + + public void setId(String id) { + this.id = id; + } + + @Override + public String getId() { + return this.id; + } + + public void setTopics(String... topics) { + Assert.notNull(topics, "'topics' must not be null"); + this.topics = Arrays.asList(topics); + } + + @Override + public List getTopics() { + return new ArrayList<>(this.topics); + } + + public void setTopicPattern(String topicPattern) { + Assert.notNull(topicPattern, "'topicPattern' must not be null"); + this.topicPattern = topicPattern; + } + + @Override + public String getTopicPattern() { + return this.topicPattern; + } + + @Override + @Nullable + public Boolean getAutoStartup() { + return this.autoStartup; + } + + public void setAutoStartup(Boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public void setupListenerContainer(ReactivePulsarMessageListenerContainer listenerContainer, + @Nullable MessageConverter messageConverter) { + + setupMessageListener(listenerContainer, messageConverter); + } + + @SuppressWarnings("unchecked") + private void setupMessageListener(ReactivePulsarMessageListenerContainer container, + @Nullable MessageConverter messageConverter) { + + PulsarMessagingMessageListenerAdapter adapter = createMessageHandler(container, messageConverter); + Assert.state(adapter != null, () -> "Endpoint [" + this + "] must provide a non null message handler"); + container.setupMessageHandler((ReactivePulsarMessageHandler) adapter); + } + + protected abstract PulsarMessagingMessageListenerAdapter createMessageHandler( + ReactivePulsarMessageListenerContainer container, @Nullable MessageConverter messageConverter); + + @Nullable + public Boolean getFluxListener() { + return this.fluxListener; + } + + public void setFluxListener(boolean fluxListener) { + this.fluxListener = fluxListener; + } + + public boolean isFluxListener() { + return this.fluxListener != null && this.fluxListener; + } + + public SubscriptionType getSubscriptionType() { + return this.subscriptionType; + } + + public void setSubscriptionType(SubscriptionType subscriptionType) { + this.subscriptionType = subscriptionType; + } + + public SchemaType getSchemaType() { + return this.schemaType; + } + + public void setSchemaType(SchemaType schemaType) { + this.schemaType = schemaType; + } + + @Override + @Nullable + public Integer getConcurrency() { + return this.concurrency; + } + + /** + * Set the concurrency for this endpoint's container. + * @param concurrency the concurrency. + */ + public void setConcurrency(Integer concurrency) { + this.concurrency = concurrency; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java new file mode 100644 index 00000000..f72da8a3 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java @@ -0,0 +1,176 @@ +/* + * Copyright 2022 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.reactive; + +import java.util.Arrays; +import java.util.List; + +import org.apache.pulsar.client.api.Schema; + +import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.listener.reactive.DefaultReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; +import org.springframework.pulsar.support.JavaUtils; +import org.springframework.pulsar.support.MessageConverter; +import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; + +/** + * Concrete implementation for {@link ReactivePulsarListenerContainerFactory}. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public class DefaultReactivePulsarListenerContainerFactory implements ReactivePulsarListenerContainerFactory { + + protected final LogAccessor logger = new LogAccessor(this.getClass()); + + private final ReactivePulsarConsumerFactory consumerFactory; + + private final ReactivePulsarContainerProperties containerProperties; + + private Boolean autoStartup; + + private MessageConverter messageConverter; + + private Boolean fluxListener; + + public DefaultReactivePulsarListenerContainerFactory(ReactivePulsarConsumerFactory consumerFactory, + ReactivePulsarContainerProperties containerProperties) { + this.consumerFactory = consumerFactory; + this.containerProperties = containerProperties; + } + + protected ReactivePulsarConsumerFactory getConsumerFactory() { + return this.consumerFactory; + } + + public ReactivePulsarContainerProperties getContainerProperties() { + return this.containerProperties; + } + + public void setAutoStartup(Boolean autoStartup) { + this.autoStartup = autoStartup; + } + + /** + * Set the message converter to use if dynamic argument type matching is needed. + * @param messageConverter the converter. + */ + public void setMessageConverter(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + public void setFluxListener(Boolean fluxListener) { + this.fluxListener = fluxListener; + } + + @SuppressWarnings("unchecked") + public DefaultReactivePulsarMessageListenerContainer createContainerInstance( + ReactivePulsarListenerEndpoint endpoint) { + + ReactivePulsarContainerProperties properties = new ReactivePulsarContainerProperties<>(); + + if (!CollectionUtils.isEmpty(endpoint.getTopics())) { + properties.setTopics(endpoint.getTopics()); + } + + if (StringUtils.hasText(endpoint.getTopicPattern())) { + properties.setTopicsPattern(endpoint.getTopicPattern()); + } + + if (StringUtils.hasText(endpoint.getSubscriptionName())) { + properties.setSubscriptionName(endpoint.getSubscriptionName()); + } + + if (endpoint.getSubscriptionType() != null) { + properties.setSubscriptionType(endpoint.getSubscriptionType()); + } + else { + properties.setSubscriptionType(this.containerProperties.getSubscriptionType()); + } + + if (endpoint.getSchemaType() != null) { + properties.setSchemaType(endpoint.getSchemaType()); + } + else { + properties.setSchemaType(this.containerProperties.getSchemaType()); + } + + if (properties.getSchema() == null) { + properties.setSchema((Schema) Schema.BYTES); + } + + if (endpoint.getConcurrency() != null) { + properties.setConcurrency(endpoint.getConcurrency()); + } + else { + properties.setConcurrency(this.containerProperties.getConcurrency()); + } + + return new DefaultReactivePulsarMessageListenerContainer<>(this.getConsumerFactory(), properties); + } + + @Override + public DefaultReactivePulsarMessageListenerContainer createListenerContainer( + ReactivePulsarListenerEndpoint endpoint) { + DefaultReactivePulsarMessageListenerContainer instance = createContainerInstance(endpoint); + if (endpoint instanceof AbstractReactivePulsarListenerEndpoint) { + configureEndpoint((AbstractReactivePulsarListenerEndpoint) endpoint); + } + + endpoint.setupListenerContainer(instance, this.messageConverter); + initializeContainer(instance, endpoint); + return instance; + } + + private void configureEndpoint(AbstractReactivePulsarListenerEndpoint aplEndpoint) { + if (aplEndpoint.getFluxListener() == null) { + JavaUtils.INSTANCE.acceptIfNotNull(this.fluxListener, aplEndpoint::setFluxListener); + } + } + + @Override + public DefaultReactivePulsarMessageListenerContainer createContainer(String... topics) { + ReactivePulsarListenerEndpoint endpoint = new ReactivePulsarListenerEndpointAdapter<>() { + + @Override + public List getTopics() { + return Arrays.asList(topics); + } + + }; + DefaultReactivePulsarMessageListenerContainer container = createContainerInstance(endpoint); + initializeContainer(container, endpoint); + return container; + } + + @SuppressWarnings("unchecked") + private void initializeContainer(DefaultReactivePulsarMessageListenerContainer instance, + ReactivePulsarListenerEndpoint endpoint) { + Boolean autoStart = endpoint.getAutoStartup(); + if (autoStart != null) { + instance.setAutoStartup(autoStart); + } + else if (this.autoStartup != null) { + instance.setAutoStartup(this.autoStartup); + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java new file mode 100644 index 00000000..8d50fd06 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java @@ -0,0 +1,282 @@ +/* + * Copyright 2022 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.reactive; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; + +import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.DeadLetterPolicy; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Messages; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.schema.AvroSchema; +import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.impl.schema.ProtobufSchema; +import org.apache.pulsar.common.schema.KeyValueEncodingType; +import org.apache.pulsar.common.schema.SchemaType; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; +import org.springframework.core.log.LogAccessor; +import org.springframework.expression.BeanResolver; +import org.springframework.lang.Nullable; +import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.pulsar.core.SchemaUtils; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.listener.Acknowledgement; +import org.springframework.pulsar.listener.adapter.HandlerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarReactiveOneByOneMessagingMessageListenerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarReactiveStreamingMessagingMessageListenerAdapter; +import org.springframework.pulsar.listener.reactive.DefaultReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; +import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; +import org.springframework.util.Assert; + +import com.google.protobuf.GeneratedMessageV3; +import reactor.core.publisher.Flux; + +/** + * A {@link ReactivePulsarListenerEndpoint} providing the method to invoke to process an + * incoming message for this endpoint. + * + * @param Message payload type + * @author Christophe Bornet + */ +public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePulsarListenerEndpoint { + + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); + + private Object bean; + + private Method method; + + private MessageHandlerMethodFactory messageHandlerMethodFactory; + + private SmartMessageConverter messagingConverter; + + private ReactiveMessageConsumerBuilderCustomizer consumerCustomizer; + + private DeadLetterPolicy deadLetterPolicy; + + public void setBean(Object bean) { + this.bean = bean; + } + + public Object getBean() { + return this.bean; + } + + /** + * Set the method to invoke to process a message managed by this endpoint. + * @param method the target method for the {@link #bean}. + */ + public void setMethod(Method method) { + this.method = method; + } + + public Method getMethod() { + return this.method; + } + + public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory messageHandlerMethodFactory) { + this.messageHandlerMethodFactory = messageHandlerMethodFactory; + } + + @Override + @SuppressWarnings("unchecked") + protected PulsarMessagingMessageListenerAdapter createMessageHandler( + ReactivePulsarMessageListenerContainer container, @Nullable MessageConverter messageConverter) { + Assert.state(this.messageHandlerMethodFactory != null, + "Could not create message listener - MessageHandlerMethodFactory not set"); + PulsarMessagingMessageListenerAdapter messageListener = createMessageListenerInstance(messageConverter); + final HandlerAdapter handlerMethod = configureListenerAdapter(messageListener); + messageListener.setHandlerMethod(handlerMethod); + + // Since we have access to the handler method here, check if we can type infer the + // Schema used. + + // TODO: filter out the payload type by excluding Consumer, Message, Messages etc. + + final MethodParameter[] methodParameters = handlerMethod.getInvokerHandlerMethod().getMethodParameters(); + MethodParameter messageParameter = null; + final Optional parameter = Arrays.stream(methodParameters) + .filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class) + || !methodParameter1.getParameterType().equals(Acknowledgement.class) + || !methodParameter1.hasParameterAnnotation(Header.class)) + .findFirst(); + final long count = Arrays.stream(methodParameters) + .filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class) + && !methodParameter1.getParameterType().equals(Acknowledgement.class) + && !methodParameter1.hasParameterAnnotation(Header.class)) + .count(); + Assert.isTrue(count == 1, "More than 1 expected payload types found"); + if (parameter.isPresent()) { + messageParameter = parameter.get(); + } + + final DefaultReactivePulsarMessageListenerContainer containerInstance = (DefaultReactivePulsarMessageListenerContainer) container; + final ReactivePulsarContainerProperties pulsarContainerProperties = containerInstance + .getContainerProperties(); + final SchemaType schemaType = pulsarContainerProperties.getSchemaType(); + if (schemaType != SchemaType.NONE) { + switch (schemaType) { + case STRING -> pulsarContainerProperties.setSchema((Schema) Schema.STRING); + case BYTES -> pulsarContainerProperties.setSchema((Schema) Schema.BYTES); + case INT8 -> pulsarContainerProperties.setSchema((Schema) Schema.INT8); + case INT16 -> pulsarContainerProperties.setSchema((Schema) Schema.INT16); + case INT32 -> pulsarContainerProperties.setSchema((Schema) Schema.INT32); + case INT64 -> pulsarContainerProperties.setSchema((Schema) Schema.INT64); + case BOOLEAN -> pulsarContainerProperties.setSchema((Schema) Schema.BOOL); + case DATE -> pulsarContainerProperties.setSchema((Schema) Schema.DATE); + case DOUBLE -> pulsarContainerProperties.setSchema((Schema) Schema.DOUBLE); + case FLOAT -> pulsarContainerProperties.setSchema((Schema) Schema.FLOAT); + case INSTANT -> pulsarContainerProperties.setSchema((Schema) Schema.INSTANT); + case LOCAL_DATE -> pulsarContainerProperties.setSchema((Schema) Schema.LOCAL_DATE); + case LOCAL_DATE_TIME -> pulsarContainerProperties.setSchema((Schema) Schema.LOCAL_DATE_TIME); + case LOCAL_TIME -> pulsarContainerProperties.setSchema((Schema) Schema.LOCAL_TIME); + case JSON -> { + Schema messageSchema = getMessageSchema(messageParameter, JSONSchema::of); + pulsarContainerProperties.setSchema((Schema) messageSchema); + } + case AVRO -> { + Schema messageSchema = getMessageSchema(messageParameter, AvroSchema::of); + pulsarContainerProperties.setSchema((Schema) messageSchema); + } + case PROTOBUF -> { + @SuppressWarnings("unchecked") + Schema messageSchema = getMessageSchema(messageParameter, + (c -> ProtobufSchema.of((Class) c))); + pulsarContainerProperties.setSchema((Schema) messageSchema); + } + case KEY_VALUE -> { + Schema messageSchema = getMessageKeyValueSchema(messageParameter); + pulsarContainerProperties.setSchema((Schema) messageSchema); + } + } + } + else { + if (messageParameter != null) { + Schema messageSchema = getMessageSchema(messageParameter, + (messageClass) -> SchemaUtils.getSchema(messageClass, false)); + if (messageSchema != null) { + pulsarContainerProperties.setSchema((Schema) messageSchema); + } + } + } + final SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType(); + pulsarContainerProperties.setSchemaType(type); + + ReactiveMessageConsumerBuilderCustomizer customizer1 = b -> b.deadLetterPolicy(this.deadLetterPolicy); + container.setConsumerCustomizer(b -> { + if (this.consumerCustomizer != null) { + this.consumerCustomizer.customize(b); + } + customizer1.customize(b); + }); + + return messageListener; + } + + private Schema getMessageSchema(MethodParameter messageParameter, Function, Schema> schemaFactory) { + ResolvableType messageType = resolvableType(messageParameter); + final Class messageClass = messageType.getRawClass(); + return schemaFactory.apply(messageClass); + } + + private Schema getMessageKeyValueSchema(MethodParameter messageParameter) { + ResolvableType messageType = resolvableType(messageParameter); + Class keyClass = messageType.resolveGeneric(0); + Class valueClass = messageType.resolveGeneric(1); + Schema> keySchema = SchemaUtils.getSchema(keyClass); + Schema> valueSchema = SchemaUtils.getSchema(valueClass); + return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); + } + + private ResolvableType resolvableType(MethodParameter methodParameter) { + ResolvableType resolvableType = ResolvableType.forMethodParameter(methodParameter); + final Class rawClass = resolvableType.getRawClass(); + if (rawClass != null && isContainerType(rawClass)) { + resolvableType = resolvableType.getGeneric(0); + } + if (Message.class.isAssignableFrom(resolvableType.getRawClass()) + || org.springframework.messaging.Message.class.isAssignableFrom(resolvableType.getRawClass())) { + resolvableType = resolvableType.getGeneric(0); + } + return resolvableType; + } + + private boolean isContainerType(Class rawClass) { + return rawClass.isAssignableFrom(Flux.class) || rawClass.isAssignableFrom(List.class) + || rawClass.isAssignableFrom(Message.class) || rawClass.isAssignableFrom(Messages.class) + || rawClass.isAssignableFrom(org.springframework.messaging.Message.class); + } + + protected HandlerAdapter configureListenerAdapter(PulsarMessagingMessageListenerAdapter messageListener) { + InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory + .createInvocableHandlerMethod(getBean(), getMethod()); + return new HandlerAdapter(invocableHandlerMethod); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + protected PulsarMessagingMessageListenerAdapter createMessageListenerInstance( + @Nullable MessageConverter messageConverter) { + + PulsarMessagingMessageListenerAdapter listener; + if (isFluxListener()) { + listener = new PulsarReactiveStreamingMessagingMessageListenerAdapter(this.bean, this.method); + } + else { + listener = new PulsarReactiveOneByOneMessagingMessageListenerAdapter(this.bean, this.method); + } + + if (messageConverter instanceof PulsarRecordMessageConverter) { + listener.setMessageConverter((PulsarRecordMessageConverter) messageConverter); + } + if (this.messagingConverter != null) { + listener.setMessagingConverter(this.messagingConverter); + } + BeanResolver resolver = getBeanResolver(); + if (resolver != null) { + listener.setBeanResolver(resolver); + } + return listener; + } + + public void setMessagingConverter(SmartMessageConverter messagingConverter) { + this.messagingConverter = messagingConverter; + } + + public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) { + this.deadLetterPolicy = deadLetterPolicy; + } + + public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer consumerCustomizer) { + this.consumerCustomizer = consumerCustomizer; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java new file mode 100644 index 00000000..cfec5a99 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java @@ -0,0 +1,31 @@ +/* + * Copyright 2022 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.reactive; + +import org.springframework.pulsar.config.ListenerContainerFactory; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; + +/** + * Factory for Pulsar reactive message listener containers. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public interface ReactivePulsarListenerContainerFactory + extends ListenerContainerFactory, ReactivePulsarListenerEndpoint> { + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java new file mode 100644 index 00000000..7d45d859 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java @@ -0,0 +1,32 @@ +/* + * Copyright 2022 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.reactive; + +import org.springframework.pulsar.config.ListenerEndpoint; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; + +/** + * Model for a Pulsar reactive listener endpoint. Can be used against a + * {@link org.springframework.pulsar.annotation.PulsarListenerConfigurer} to register + * endpoints programmatically. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public interface ReactivePulsarListenerEndpoint extends ListenerEndpoint> { + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java new file mode 100644 index 00000000..01bd83c9 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java @@ -0,0 +1,83 @@ +/* + * Copyright 2022 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.reactive; + +import java.util.Collections; +import java.util.List; + +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; + +import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.support.MessageConverter; + +/** + * Adapter to avoid having to implement all methods. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public class ReactivePulsarListenerEndpointAdapter implements ReactivePulsarListenerEndpoint { + + @Override + public String getId() { + return null; + } + + @Override + public String getSubscriptionName() { + return null; + } + + @Override + public SubscriptionType getSubscriptionType() { + return SubscriptionType.Exclusive; + } + + @Override + public List getTopics() { + return Collections.emptyList(); + } + + @Override + public String getTopicPattern() { + return null; + } + + @Override + public Boolean getAutoStartup() { + return null; + } + + @Override + public void setupListenerContainer(ReactivePulsarMessageListenerContainer listenerContainer, + MessageConverter messageConverter) { + } + + @Override + public SchemaType getSchemaType() { + return null; + } + + @Nullable + @Override + public Integer getConcurrency() { + return null; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java new file mode 100644 index 00000000..70a7f0ee --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java @@ -0,0 +1,46 @@ +/* + * Copyright 2022 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.reactive; + +import org.springframework.pulsar.config.ListenerEndpointRegistry; +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; + +/** + * Creates the necessary {@link ReactivePulsarMessageListenerContainer} instances for the + * registered {@linkplain ReactivePulsarListenerEndpoint endpoints}. Also manages the + * lifecycle of the listener containers, in particular within the lifecycle of the + * application context. + * + *

+ * Contrary to {@link ReactivePulsarMessageListenerContainer}s created manually, listener + * containers managed by registry are not beans in the application context and are not + * candidates for autowiring. Use {@link #getListenerContainers()} if you need to access + * this registry's listener containers for management purposes. If you need to access to a + * specific message listener container, use {@link #getListenerContainer(String)} with the + * id of the endpoint. + * + * @param Message payload type. + * @author Christophe Bornet + */ +public class ReactivePulsarListenerEndpointRegistry + extends ListenerEndpointRegistry, ReactivePulsarListenerEndpoint> { + + public ReactivePulsarListenerEndpointRegistry() { + super(ReactivePulsarMessageListenerContainer.class); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 0c9fb38b..bdc992db 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -39,7 +39,7 @@ import io.micrometer.observation.ObservationRegistry; * @author Soby Chacko * @author Alexander Preuß */ -public abstract class AbstractPulsarMessageListenerContainer implements PulsarMessageListenerContainer, +public non-sealed abstract class AbstractPulsarMessageListenerContainer implements PulsarMessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware { protected final LogAccessor logger = new LogAccessor(this.getClass()); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ListenerContainerRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ListenerContainerRegistry.java new file mode 100644 index 00000000..5bc0825f --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ListenerContainerRegistry.java @@ -0,0 +1,67 @@ +/* + * Copyright 2022 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 java.util.Collection; +import java.util.Set; + +import org.springframework.lang.Nullable; +import org.springframework.pulsar.config.ListenerEndpoint; + +/** + * A registry for containers. + * + * @author Christophe Bornet + */ +public interface ListenerContainerRegistry { + + /** + * Return the listener container with the specified id or {@code null} if no such + * container exists. + * @param id the id of the container + * @return the container or {@code null} if no container with that id exists + * @see ListenerEndpoint#getId() + * @see #getListenerContainerIds() + */ + @Nullable + MessageListenerContainer getListenerContainer(String id); + + /** + * Return the ids of the managed listener container instance(s). + * @return the ids. + * @see #getListenerContainer(String) + */ + Set getListenerContainerIds(); + + /** + * Return the managed listener container instance(s). + * @return the managed listener container instance(s). + * @see #getAllListenerContainers() + */ + Collection getListenerContainers(); + + /** + * Return all listener container instances including those managed by this registry + * and those declared as beans in the application context. Prototype-scoped containers + * will be included. Lazy beans that have not yet been created will not be initialized + * by a call to this method. + * @return the listener container instance(s). + * @see #getListenerContainers() + */ + Collection getAllListenerContainers(); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/MessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/MessageListenerContainer.java new file mode 100644 index 00000000..5249ad35 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/MessageListenerContainer.java @@ -0,0 +1,39 @@ +/* + * Copyright 2022 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 org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; + +/** + * Internal abstraction used by the framework representing a message listener container. + * Not meant to be implemented externally. + * + * @author Christophe Bornet + */ +public interface MessageListenerContainer extends SmartLifecycle, DisposableBean { + + @Override + default void destroy() { + stop(); + } + + default void setAutoStartup(boolean autoStartup) { + // empty + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java index b4ba3a1e..be79c49f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java @@ -19,28 +19,17 @@ package org.springframework.pulsar.listener; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.RedeliveryBackoff; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.context.SmartLifecycle; - /** * Internal abstraction used by the framework representing a message listener container. * Not meant to be implemented externally. * * @author Soby Chacko */ -public interface PulsarMessageListenerContainer extends SmartLifecycle, DisposableBean { +public sealed interface PulsarMessageListenerContainer + extends MessageListenerContainer permits AbstractPulsarMessageListenerContainer { void setupMessageListener(Object messageListener); - @Override - default void destroy() { - stop(); - } - - default void setAutoStartup(boolean autoStartup) { - // empty - } - default PulsarContainerProperties getContainerProperties() { throw new UnsupportedOperationException("This container doesn't support retrieving its properties"); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java index 1c505a9d..9ac19aa1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java @@ -44,12 +44,15 @@ import org.springframework.pulsar.support.converter.PulsarMessagingMessageConver import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; import org.springframework.util.Assert; +import reactor.core.publisher.Flux; + /** * An abstract {@link org.apache.pulsar.client.api.MessageListener} adapter providing the * necessary infrastructure to extract the payload from a Pulsar message. * * @param payload type. * @author Soby Chacko + * @author Christophe Bornet */ public abstract class PulsarMessagingMessageListenerAdapter { @@ -73,6 +76,8 @@ public abstract class PulsarMessagingMessageListenerAdapter { private boolean isSpringMessageList; + private boolean isSpringMessageFlux; + private boolean isSpringMessage; private boolean isConsumerRecords; @@ -130,6 +135,10 @@ public abstract class PulsarMessagingMessageListenerAdapter { return this.isSpringMessageList; } + protected boolean isSpringMessageFlux() { + return this.isSpringMessageFlux; + } + protected org.springframework.messaging.Message toMessagingMessage(Message record, Consumer consumer) { return getMessageConverter().toMessage(record, consumer, getType()); } @@ -167,7 +176,8 @@ public abstract class PulsarMessagingMessageListenerAdapter { else if (parameterIsType(parameterType, Message.class)) { pulsarMessageFound = true; } - else if (parameterIsType(parameterType, List.class) || parameterIsType(parameterType, Messages.class)) { + else if (parameterIsType(parameterType, List.class) || parameterIsType(parameterType, Messages.class) + || parameterIsType(parameterType, Flux.class)) { collectionFound = true; } } @@ -226,6 +236,18 @@ public abstract class PulsarMessagingMessageListenerAdapter { this.simpleExtraction = true; } } + else if (parameterizedType.getRawType().equals(Flux.class) + && parameterizedType.getActualTypeArguments().length == 1) { + + Type paramType = parameterizedType.getActualTypeArguments()[0]; + boolean messageHasGeneric = paramType instanceof ParameterizedType && ((ParameterizedType) paramType) + .getRawType().equals(org.springframework.messaging.Message.class); + this.isSpringMessageFlux = paramType.equals(org.springframework.messaging.Message.class) + || messageHasGeneric; + if (messageHasGeneric) { + genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0]; + } + } else { this.isConsumerRecords = parameterizedType.getRawType().equals(Messages.class); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java new file mode 100644 index 00000000..ceddcc54 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java @@ -0,0 +1,67 @@ +/* + * Copyright 2022 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.adapter; + +import java.lang.reflect.Method; + +import org.apache.pulsar.client.api.Message; +import org.reactivestreams.Publisher; + +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; +import org.springframework.pulsar.listener.reactive.ReactivePulsarOneByOneMessageHandler; + +import reactor.core.publisher.Mono; + +/** + * A {@link ReactivePulsarMessageHandler MessageListener} adapter that invokes a + * configurable {@link HandlerAdapter}; used when the factory is configured for the + * listener to receive individual messages. + * + * @param payload type. + * @author Christophe Bornet + */ +public class PulsarReactiveOneByOneMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter + implements ReactivePulsarOneByOneMessageHandler { + + public PulsarReactiveOneByOneMessagingMessageListenerAdapter(Object bean, Method method) { + super(bean, method); + } + + @Override + @SuppressWarnings("unchecked") + public Publisher received(Message record) { + org.springframework.messaging.Message message = null; + Object theRecord = record; + if (isHeaderFound() || isSpringMessage()) { + message = toMessagingMessage(record, null); + } + else if (isSimpleExtraction()) { + theRecord = record.getValue(); + } + + if (logger.isDebugEnabled()) { + this.logger.debug("Processing [" + message + "]"); + } + try { + return (Mono) invokeHandler(theRecord, message, null, null); + } + catch (Exception e) { + return Mono.error(e); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java new file mode 100644 index 00000000..7cc99acd --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java @@ -0,0 +1,59 @@ +/* + * Copyright 2022 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.adapter; + +import java.lang.reflect.Method; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.reactive.client.api.MessageResult; + +import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; +import org.springframework.pulsar.listener.reactive.ReactivePulsarStreamingHandler; + +import reactor.core.publisher.Flux; + +/** + * A {@link ReactivePulsarMessageHandler MessageListener} adapter that invokes a + * configurable {@link HandlerAdapter}; used when the factory is configured for the + * listener to receive a flux of messages. + * + * @param payload type. + * @author Christophe Bornet + */ +public class PulsarReactiveStreamingMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter + implements ReactivePulsarStreamingHandler { + + public PulsarReactiveStreamingMessagingMessageListenerAdapter(Object bean, Method method) { + super(bean, method); + } + + @Override + @SuppressWarnings("unchecked") + public Flux> received(Flux> records) { + Flux theRecords = records; + if (isSpringMessageFlux()) { + theRecords = records.map(record -> toMessagingMessage(record, null)); + } + try { + return (Flux>) invokeHandler(theRecords, null, null, null); + } + catch (Exception e) { + return Flux.error(e); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java index aa6c8516..5415e9e6 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java @@ -151,7 +151,7 @@ public non-sealed class DefaultReactivePulsarMessageListenerContainer builder.subscriptionName(containerProperties.getSubscriptionName()); } if (!CollectionUtils.isEmpty(containerProperties.getTopics())) { - builder.topicNames(containerProperties.getTopics()); + builder.topicNames(new ArrayList<>(containerProperties.getTopics())); } if (containerProperties.getTopicsPattern() != null) { builder.topicsPattern(containerProperties.getTopicsPattern()); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java index ae3063cd..84125a81 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java @@ -17,11 +17,12 @@ package org.springframework.pulsar.listener.reactive; import java.time.Duration; -import java.util.List; +import java.util.Collection; import java.util.regex.Pattern; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; /** * Contains runtime properties for a reactive listener container. @@ -31,7 +32,7 @@ import org.apache.pulsar.client.api.SubscriptionType; */ public class ReactivePulsarContainerProperties { - private List topics; + private Collection topics; private Pattern topicsPattern; @@ -41,6 +42,8 @@ public class ReactivePulsarContainerProperties { private Schema schema; + private SchemaType schemaType; + private ReactivePulsarMessageHandler messageHandler; private Duration handlingTimeout = Duration.ofMinutes(2); @@ -73,11 +76,19 @@ public class ReactivePulsarContainerProperties { this.schema = schema; } - public List getTopics() { + public SchemaType getSchemaType() { + return this.schemaType; + } + + public void setSchemaType(SchemaType schemaType) { + this.schemaType = schemaType; + } + + public Collection getTopics() { return this.topics; } - public void setTopics(List topics) { + public void setTopics(Collection topics) { this.topics = topics; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java index 50452f59..48bbc469 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java @@ -16,9 +16,8 @@ package org.springframework.pulsar.listener.reactive; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.context.SmartLifecycle; import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.listener.MessageListenerContainer; /** * Internal abstraction used by the framework representing a reactive message listener @@ -28,19 +27,10 @@ import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCu * @author Christophe Bornet */ public sealed interface ReactivePulsarMessageListenerContainer - extends SmartLifecycle, DisposableBean permits DefaultReactivePulsarMessageListenerContainer { + extends MessageListenerContainer permits DefaultReactivePulsarMessageListenerContainer { void setupMessageHandler(ReactivePulsarMessageHandler messageListener); - @Override - default void destroy() { - stop(); - } - - default void setAutoStartup(boolean autoStartup) { - // empty - } - default ReactivePulsarContainerProperties getContainerProperties() { throw new UnsupportedOperationException("This container doesn't support retrieving its properties"); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index 924beb29..9cd09006 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java @@ -64,7 +64,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { PulsarListenerEndpoint pulsarListenerEndpoint = mock(PulsarListenerEndpoint.class); when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1); - ConcurrentPulsarMessageListenerContainer concurrentContainer = containerFactory + AbstractPulsarMessageListenerContainer concurrentContainer = containerFactory .createListenerContainer(pulsarListenerEndpoint); PulsarContainerProperties pulsarContainerProperties = concurrentContainer.getContainerProperties(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 2b29ecec..9a66794d 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -127,7 +127,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory) { final ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( pulsarConsumerFactory, new PulsarContainerProperties(), null); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java new file mode 100644 index 00000000..72534ab1 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java @@ -0,0 +1,613 @@ +/* + * Copyright 2022 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.reactive; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.DeadLetterPolicy; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.impl.schema.AvroSchema; +import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.impl.schema.ProtobufSchema; +import org.apache.pulsar.common.schema.KeyValue; +import org.apache.pulsar.common.schema.KeyValueEncodingType; +import org.apache.pulsar.common.schema.SchemaType; +import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory; +import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.pulsar.annotation.EnablePulsar; +import org.springframework.pulsar.annotation.ReactivePulsarListener; +import org.springframework.pulsar.config.PulsarClientConfiguration; +import org.springframework.pulsar.config.PulsarClientFactoryBean; +import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.core.PulsarProducerFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTestContainerSupport; +import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.listener.Proto; +import org.springframework.pulsar.support.PulsarHeaders; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Tests for {@link ReactivePulsarListener} annotation. + * + * @author Christophe Bornet + */ +@SpringJUnitConfig +@DirtiesContext +public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { + + @Autowired + PulsarTemplate pulsarTemplate; + + @Autowired + private PulsarClient pulsarClient; + + @Configuration(proxyBeanMethods = false) + @EnablePulsar + public static class TopLevelConfig { + + @Bean + public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { + return new DefaultPulsarProducerFactory<>(pulsarClient, new HashMap<>()); + } + + @Bean + public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { + return new PulsarClientFactoryBean(pulsarClientConfiguration); + } + + @Bean + public PulsarClientConfiguration pulsarClientConfiguration() { + return new PulsarClientConfiguration(Map.of("serviceUrl", PulsarTestContainerSupport.getPulsarBrokerUrl())); + } + + @Bean + public ReactivePulsarClient pulsarReactivePulsarClient(PulsarClient pulsarClient) { + return AdaptedReactivePulsarClientFactory.create(pulsarClient); + } + + @Bean + public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + return new PulsarTemplate<>(pulsarProducerFactory); + } + + @Bean + public ReactivePulsarConsumerFactory pulsarConsumerFactory(ReactivePulsarClient pulsarClient) { + return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, new MutableReactiveMessageConsumerSpec()); + } + + @Bean + ReactivePulsarListenerContainerFactory reactivePulsarListenerContainerFactory( + ReactivePulsarConsumerFactory pulsarConsumerFactory) { + return new DefaultReactivePulsarListenerContainerFactory<>(pulsarConsumerFactory, + new ReactivePulsarContainerProperties<>()); + } + + @Bean + PulsarAdministration pulsarAdministration() { + return new PulsarAdministration( + PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + } + + @Bean + PulsarTopic partitionedTopic() { + return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); + } + + } + + @Nested + @ContextConfiguration(classes = PulsarListenerBasicTestCases.TestPulsarListenersForBasicScenario.class) + class PulsarListenerBasicTestCases { + + static CountDownLatch latch1 = new CountDownLatch(1); + static CountDownLatch latch2 = new CountDownLatch(1); + static CountDownLatch latch3 = new CountDownLatch(3); + + @Autowired + ReactivePulsarListenerEndpointRegistry registry; + + @Test + void testPulsarListener() throws Exception { + ReactivePulsarContainerProperties pulsarContainerProperties = registry.getListenerContainer("id-1") + .getContainerProperties(); + assertThat(pulsarContainerProperties.getTopics()).containsExactly("topic-1"); + assertThat(pulsarContainerProperties.getSubscriptionName()).isEqualTo("subscription-1"); + pulsarTemplate.send("topic-1", "hello foo"); + assertThat(latch1.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testPulsarListenerWithConsumerCustomizer() throws Exception { + pulsarTemplate.send("topic-2", "hello foo"); + assertThat(latch2.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testPulsarListenerWithTopicsPattern() throws Exception { + ReactivePulsarContainerProperties containerProperties = registry.getListenerContainer("id-3") + .getContainerProperties(); + assertThat(containerProperties.getTopicsPattern().toString()) + .isEqualTo("persistent://public/default/pattern.*"); + + pulsarTemplate.send("persistent://public/default/pattern-1", "hello baz"); + pulsarTemplate.send("persistent://public/default/pattern-2", "hello baz"); + pulsarTemplate.send("persistent://public/default/pattern-3", "hello baz"); + + assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class TestPulsarListenersForBasicScenario { + + @ReactivePulsarListener(id = "id-1", topics = "topic-1", subscriptionName = "subscription-1", + consumerCustomizer = "consumerCustomizer") + Mono listen1(String message) { + latch1.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(consumerCustomizer = "listen2Customizer") + Mono listen2(String message) { + latch2.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer listen2Customizer() { + return b -> b.topicNames(List.of("topic-2")) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + @ReactivePulsarListener(id = "id-3", topicPattern = "persistent://public/default/pattern.*", + subscriptionName = "subscription-3", consumerCustomizer = "consumerCustomizer") + Mono listen3(String message) { + latch3.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + return b -> b.topicsPatternAutoDiscoveryPeriod(Duration.ofSeconds(2)) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + } + + @Nested + @ContextConfiguration(classes = PulsarListenerStreamingTestCases.TestPulsarListenersForStreaming.class) + class PulsarListenerStreamingTestCases { + + static CountDownLatch latch1 = new CountDownLatch(10); + static CountDownLatch latch2 = new CountDownLatch(10); + + @Test + void testPulsarListenerStreaming() throws Exception { + for (int i = 0; i < 10; i++) { + pulsarTemplate.send("topic-4", "hello foo"); + } + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testPulsarListenerStreamingSpringMessage() throws Exception { + for (int i = 0; i < 10; i++) { + pulsarTemplate.send("topic-5", "hello foo"); + } + assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class TestPulsarListenersForStreaming { + + @ReactivePulsarListener(topics = "topic-4", subscriptionName = "subscription-4", stream = true, + consumerCustomizer = "consumerCustomizer") + Flux> listen4(Flux> messages) { + return messages.doOnNext(m -> latch1.countDown()).map(m -> MessageResult.acknowledge(m.getMessageId())); + } + + @ReactivePulsarListener(topics = "topic-5", subscriptionName = "subscription-5", stream = true, + consumerCustomizer = "consumerCustomizer") + Flux> listen5(Flux> messages) { + return messages.doOnNext(m -> latch2.countDown()).map(m -> { + Object mId = m.getHeaders().get(PulsarHeaders.MESSAGE_ID); + if (mId instanceof MessageId) { + return (MessageId) mId; + } + else { + throw new RuntimeException("Missing message Id"); + } + }).map(MessageResult::acknowledge); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + } + + @Nested + @ContextConfiguration(classes = DeadLetterPolicyTest.DeadLetterPolicyConfig.class) + class DeadLetterPolicyTest { + + private static CountDownLatch latch = new CountDownLatch(2); + + private static CountDownLatch dlqLatch = new CountDownLatch(1); + + @Test + void pulsarListenerWithDeadLetterPolicy() throws Exception { + pulsarTemplate.send("dlpt-topic-1", "hello"); + assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class DeadLetterPolicyConfig { + + @ReactivePulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", + topics = "dlpt-topic-1", deadLetterPolicy = "deadLetterPolicy", + consumerCustomizer = "consumerCustomizer", subscriptionType = SubscriptionType.Shared) + Mono listen(String msg) { + latch.countDown(); + return Mono.error(new RuntimeException("fail " + msg)); + } + + @ReactivePulsarListener(id = "dlqListener", topics = "dlpt-dlq-topic", + consumerCustomizer = "consumerCustomizer") + Mono listenDlq(String msg) { + dlqLatch.countDown(); + return Mono.empty(); + } + + @Bean + DeadLetterPolicy deadLetterPolicy() { + return DeadLetterPolicy.builder().maxRedeliverCount(1).deadLetterTopic("dlpt-dlq-topic").build(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerCustomizer() { + return b -> b.negativeAckRedeliveryDelay(Duration.ofSeconds(1)) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + } + + @Nested + @ContextConfiguration(classes = SchemaTestCases.SchemaTestConfig.class) + class SchemaTestCases { + + static CountDownLatch jsonLatch = new CountDownLatch(3); + static CountDownLatch avroLatch = new CountDownLatch(3); + static CountDownLatch keyvalueLatch = new CountDownLatch(3); + static CountDownLatch protobufLatch = new CountDownLatch(3); + + @Test + void jsonSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(JSONSchema.of(User.class)); + + for (int i = 0; i < 3; i++) { + template.send("json-topic", new User("Jason", i)); + } + assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void avroSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(AvroSchema.of(User.class)); + + for (int i = 0; i < 3; i++) { + template.send("avro-topic", new User("Avi", i)); + } + assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void keyvalueSchema() throws Exception { + PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, Collections.emptyMap()); + PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); + Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, + KeyValueEncodingType.INLINE); + template.setSchema(kvSchema); + + for (int i = 0; i < 3; i++) { + template.send("keyvalue-topic", new KeyValue<>("Kevin", i)); + } + assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void protobufSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(ProtobufSchema.of(Proto.Person.class)); + + for (int i = 0; i < 3; i++) { + template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build()); + } + assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class SchemaTestConfig { + + @ReactivePulsarListener(id = "jsonListener", topics = "json-topic", schemaType = SchemaType.JSON, + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenJson(User message) { + jsonLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "avroListener", topics = "avro-topic", schemaType = SchemaType.AVRO, + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenAvro(User message) { + avroLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "keyvalueListener", topics = "keyvalue-topic", + schemaType = SchemaType.KEY_VALUE, consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenKeyvalue(KeyValue message) { + keyvalueLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "protobufListener", topics = "protobuf-topic", + schemaType = SchemaType.PROTOBUF, consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenProtobuf(Proto.Person message) { + protobufLatch.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + static class User { + + private String name; + + private int age; + + User() { + + } + + User(String name, int age) { + this.name = name; + this.age = age; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public int getAge() { + return age; + } + + public void setAge(int age) { + this.age = age; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + User user = (User) o; + return age == user.age && Objects.equals(name, user.name); + } + + @Override + public int hashCode() { + return Objects.hash(name, age); + } + + @Override + public String toString() { + return "User{" + "name='" + name + '\'' + ", age=" + age + '}'; + } + + } + + } + + @Nested + @ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class) + class PulsarHeadersTest { + + static CountDownLatch simpleListenerLatch = new CountDownLatch(1); + static CountDownLatch pulsarMessageListenerLatch = new CountDownLatch(1); + static CountDownLatch springMessagingMessageListenerLatch = new CountDownLatch(1); + + static AtomicReference capturedData = new AtomicReference<>(); + static AtomicReference messageId = new AtomicReference<>(); + static AtomicReference topicName = new AtomicReference<>(); + static AtomicReference fooValue = new AtomicReference<>(); + static AtomicReference rawData = new AtomicReference<>(); + + @Test + void simpleListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-simple-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "simpleListenerWithHeaders")) + .withTopic("simpleListenerWithHeaders").send(); + assertThat(simpleListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData.get()).isEqualTo("hello-simple-listener"); + assertThat(PulsarHeadersTest.messageId.get()).isEqualTo(messageId); + assertThat(topicName.get()).isEqualTo("persistent://public/default/simpleListenerWithHeaders"); + assertThat(fooValue.get()).isEqualTo("simpleListenerWithHeaders"); + assertThat(rawData.get()).isEqualTo("hello-simple-listener".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void pulsarMessageListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-message-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "pulsarMessageListenerWithHeaders")) + .withTopic("pulsarMessageListenerWithHeaders").send(); + assertThat(pulsarMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData.get()).isEqualTo("hello-pulsar-message-listener"); + assertThat(PulsarHeadersTest.messageId.get()).isEqualTo(messageId); + assertThat(topicName.get()).isEqualTo("persistent://public/default/pulsarMessageListenerWithHeaders"); + assertThat(fooValue.get()).isEqualTo("pulsarMessageListenerWithHeaders"); + assertThat(rawData.get()).isEqualTo("hello-pulsar-message-listener".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void springMessagingMessageListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-spring-messaging-message-listener") + .withMessageCustomizer(messageBuilder -> messageBuilder.property("foo", + "springMessagingMessageListenerWithHeaders")) + .withTopic("springMessagingMessageListenerWithHeaders").send(); + assertThat(springMessagingMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData.get()).isEqualTo("hello-spring-messaging-message-listener"); + assertThat(PulsarHeadersTest.messageId.get()).isEqualTo(messageId); + assertThat(topicName.get()) + .isEqualTo("persistent://public/default/springMessagingMessageListenerWithHeaders"); + assertThat(fooValue.get()).isEqualTo("springMessagingMessageListenerWithHeaders"); + assertThat(rawData.get()) + .isEqualTo("hello-spring-messaging-message-listener".getBytes(StandardCharsets.UTF_8)); + } + + @EnablePulsar + @Configuration + static class PulsarListenerWithHeadersConfig { + + @ReactivePulsarListener(subscriptionName = "simple-listener-with-headers-sub", + topics = "simpleListenerWithHeaders", consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono simpleListenerWithHeaders(String data, @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData, + @Header("foo") String foo) { + capturedData.set(data); + PulsarHeadersTest.messageId.set(messageId); + PulsarHeadersTest.topicName.set(topicName); + fooValue.set(foo); + PulsarHeadersTest.rawData.set(rawData); + simpleListenerLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub", + topics = "pulsarMessageListenerWithHeaders", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono pulsarMessageListenerWithHeaders(Message data, + @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData, + @Header("foo") String foo) { + capturedData.set(data.getValue()); + PulsarHeadersTest.messageId.set(messageId); + PulsarHeadersTest.topicName.set(topicName); + fooValue.set(foo); + PulsarHeadersTest.rawData.set(rawData); + pulsarMessageListenerLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub", + topics = "springMessagingMessageListenerWithHeaders", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono springMessagingMessageListenerWithHeaders(org.springframework.messaging.Message data, + @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.RAW_DATA) byte[] rawData, @Header(PulsarHeaders.TOPIC_NAME) String topicName, + @Header("foo") String foo) { + capturedData.set(data.getPayload()); + PulsarHeadersTest.messageId.set(messageId); + PulsarHeadersTest.topicName.set(topicName); + fooValue.set(foo); + PulsarHeadersTest.rawData.set(rawData); + springMessagingMessageListenerLatch.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index 10a462dd..3f034fe4 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java @@ -155,7 +155,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul } @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory, ObservationRegistry observationRegistry) { return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, new PulsarContainerProperties(), observationRegistry); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 43bc46d5..9f5ee4d0 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java @@ -202,7 +202,7 @@ public class ObservationTests implements PulsarTestContainerSupport { } @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory, ObservationRegistry observationRegistry) { PulsarContainerProperties containerProperties = new PulsarContainerProperties(); containerProperties.setObservationConvention(new DefaultPulsarListenerObservationConvention() {