Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add ReactivePulsarListener annotation (#203)

Browse Source 
Fixes #8
This commit is contained in:
Christophe Bornet
2022-11-16 21:32:23 +01:00
committed by GitHub
parent 9e1b1df1e8
commit e91e8e33d5
40 changed files with 3306 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -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<DefaultPulsarMessageListenerContainer<String>> listenerContainerFactory = mock(
PulsarListenerContainerFactory.class);
PulsarListenerContainerFactory listenerContainerFactory = mock(PulsarListenerContainerFactory.class);
this.contextRunner
.withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class,
() -> listenerContainerFactory)

14
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java
View File

@@ -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 <em>default</em> 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 <em>default</em> container
* factory is assumed to be available with a bean name of
* {@code pulsarListenerContainerFactory} unless an explicit default has been provided
* through configuration.
*
* <p>
* 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;
/**

13
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -149,7 +149,8 @@ public class PulsarListenerAnnotationBeanPostProcessor<V>
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<V>
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)) {

8
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java
View File

@@ -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() };
}
}

60
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java Normal file
View File

@@ -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}.
*
* <p>
* 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));
}
}
}

161
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java Normal file
View File

@@ -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 <em>default</em> container factory is assumed to be available
* with a bean name of {@code pulsarListenerContainerFactory} unless an explicit default
* has been provided through configuration.
*
* <p>
* Processing of {@code @ReactivePulsarListener} annotations is performed by registering a
* {@link ReactivePulsarListenerAnnotationBeanPostProcessor}. This can be done manually
* or, more conveniently, through {@link EnablePulsar} annotation.
* </p>
*
* @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.
* <p>
* If none is specified an auto-generated id is used.
* <p>
* 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.
* <p>
* 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.
* <p>
* 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'.
* <p>
* @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.
* <p>
* 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 "";
}

785
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java Normal file
View File

@@ -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.
*
* <p>
* Annotated methods can use flexible arguments as defined by
* {@link ReactivePulsarListener}.
*
* <p>
* This post-processor is automatically registered by the {@link EnablePulsar} annotation.
*
* <p>
* 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 <V> the payload type.
* @author Christophe Bornet
* @see ReactivePulsarListener
* @see EnablePulsar
* @see PulsarListenerConfigurer
* @see PulsarListenerEndpointRegistrar
* @see ReactivePulsarListenerEndpointRegistry
* @see ReactivePulsarListenerEndpoint
* @see MethodReactivePulsarListenerEndpoint
*/
public class ReactivePulsarListenerAnnotationBeanPostProcessor<V>
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<Class<?>> 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<AnnotationEnhancer> 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<Method, Set<ReactivePulsarListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
(MethodIntrospector.MetadataLookup<Set<ReactivePulsarListener>>) method -> {
Set<ReactivePulsarListener> 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<Method, Set<ReactivePulsarListener>> 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<V> 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<String> 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<String> 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<Object>) 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<ReactivePulsarListener> findListenerAnnotations(Class<?> clazz) {
Set<ReactivePulsarListener> 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<ReactivePulsarListener> findListenerAnnotations(Method method) {
Set<ReactivePulsarListener> 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<HandlerMethodArgumentResolver> 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<byte[], String> {
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<ConvertiblePair> getConvertibleTypes() {
HashSet<ConvertiblePair> 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<String, Object> 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<Map<String, Object>, AnnotatedElement, Map<String, Object>> {
}
}

45
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java Normal file
View File

@@ -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.
* <p>
* 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();
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java
View File

@@ -43,7 +43,7 @@ import io.micrometer.observation.ObservationRegistry;
* @author Chris Bono
*/
public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractPulsarMessageListenerContainer<T>, T>
implements PulsarListenerContainerFactory<C>, ApplicationEventPublisherAware, ApplicationContextAware {
implements PulsarListenerContainerFactory, ApplicationEventPublisherAware, ApplicationContextAware {
protected final LogAccessor logger = new LogAccessor(this.getClass());

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java
View File

@@ -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<K>
implements PulsarListenerEndpoint, BeanFactoryAware, InitializingBean {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
private String subscriptionName;
private SubscriptionType subscriptionType;

51
spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java Normal file
View File

@@ -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 <C> message listener container type.
* @param <E> listener endpoint type.
* @author Soby Chacko
* @author Christophe Bornet
*/
public interface ListenerContainerFactory<C extends MessageListenerContainer, E extends ListenerEndpoint<C>> {
/**
* 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);
}

106
spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java Normal file
View File

@@ -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 <C> Message listener container type.
* @author Christophe Bornet
*/
public interface ListenerEndpoint<C extends MessageListenerContainer> {
/**
* 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<String> 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.
* <p>
* 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);
}

257
spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpointRegistry.java Normal file
View File

@@ -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.
*
* <p>
* 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 <C> listener container type.
* @param <E> listener endpoint type.
* @author Soby Chacko
* @author Christophe Bornet
*/
public class ListenerEndpointRegistry<C extends MessageListenerContainer, E extends ListenerEndpoint<C>>
implements ListenerContainerRegistry, DisposableBean, SmartLifecycle, ApplicationContextAware,
ApplicationListener<ContextRefreshedEvent> {
private final Class<? extends C> type;
private final Map<String, C> 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<? extends C>) 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<String> getListenerContainerIds() {
return Collections.unmodifiableSet(this.listenerContainers.keySet());
}
@Override
public Collection<C> getListenerContainers() {
return Collections.unmodifiableCollection(this.listenerContainers.values());
}
@Override
public Collection<C> getAllListenerContainers() {
List<C> containers = new ArrayList<>(getListenerContainers());
containers.addAll(this.applicationContext.getBeansOfType(this.type, true, false).values());
return containers;
}
public void registerListenerContainer(E endpoint, ListenerContainerFactory<? extends C, E> factory) {
registerListenerContainer(endpoint, factory, false);
}
public void registerListenerContainer(E endpoint, ListenerContainerFactory<? extends C, E> 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<? extends C, E> 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<C> 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();
}
}
}
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java
View File

@@ -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";
}

9
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactory.java
View File

@@ -21,13 +21,10 @@ import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
/**
* Factory for Pulsar message listener containers.
*
* @param <C> message listener container type.
* @author Soby Chacko
* @author Christophe Bornet
*/
public interface PulsarListenerContainerFactory<C extends PulsarMessageListenerContainer> {
C createListenerContainer(PulsarListenerEndpoint endpoint);
C createContainer(String... topics);
public interface PulsarListenerContainerFactory
extends ListenerContainerFactory<PulsarMessageListenerContainer, PulsarListenerEndpoint> {
}

32
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java
View File

@@ -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<String> getTopics();
String getTopicPattern();
@Nullable
Boolean getAutoStartup();
void setupListenerContainer(PulsarMessageListenerContainer listenerContainer,
@Nullable MessageConverter messageConverter);
public interface PulsarListenerEndpoint extends ListenerEndpoint<PulsarMessageListenerContainer> {
boolean isBatchListener();
SchemaType getSchemaType();
Properties getConsumerProperties();
@Nullable
Integer getConcurrency();
AckMode getAckMode();
}

48
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistrar.java
View File

@@ -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<? extends ListenerContainerFactory> type;
private final List<PulsarListenerEndpointDescriptor> endpointDescriptors = new ArrayList<>();
private PulsarListenerEndpointRegistry endpointRegistry;
private ListenerEndpointRegistry endpointRegistry;
private List<HandlerMethodArgumentResolver> 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<? extends ListenerContainerFactory> 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;

217
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistry.java
View File

@@ -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<ContextRefreshedEvent> {
private final Map<String, PulsarMessageListenerContainer> 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<String> getListenerContainerIds() {
return Collections.unmodifiableSet(this.listenerContainers.keySet());
}
@Override
public Collection<PulsarMessageListenerContainer> getListenerContainers() {
return Collections.unmodifiableCollection(this.listenerContainers.values());
}
@Override
public Collection<PulsarMessageListenerContainer> getAllListenerContainers() {
List<PulsarMessageListenerContainer> 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<PulsarMessageListenerContainer> 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<PulsarMessageListenerContainer, PulsarListenerEndpoint> {
public PulsarListenerEndpointRegistry() {
super(PulsarMessageListenerContainer.class);
}
}

229
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java Normal file
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
*/
public abstract class AbstractReactivePulsarListenerEndpoint<T>
implements ReactivePulsarListenerEndpoint<T>, BeanFactoryAware, InitializingBean {
private String subscriptionName;
private SubscriptionType subscriptionType;
private SchemaType schemaType;
private String id;
private Collection<String> 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<String> 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<T> listenerContainer,
@Nullable MessageConverter messageConverter) {
setupMessageListener(listenerContainer, messageConverter);
}
@SuppressWarnings("unchecked")
private void setupMessageListener(ReactivePulsarMessageListenerContainer<T> container,
@Nullable MessageConverter messageConverter) {
PulsarMessagingMessageListenerAdapter<T> adapter = createMessageHandler(container, messageConverter);
Assert.state(adapter != null, () -> "Endpoint [" + this + "] must provide a non null message handler");
container.setupMessageHandler((ReactivePulsarMessageHandler) adapter);
}
protected abstract PulsarMessagingMessageListenerAdapter<T> createMessageHandler(
ReactivePulsarMessageListenerContainer<T> 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;
}
}

176
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java Normal file
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
*/
public class DefaultReactivePulsarListenerContainerFactory<T> implements ReactivePulsarListenerContainerFactory<T> {
protected final LogAccessor logger = new LogAccessor(this.getClass());
private final ReactivePulsarConsumerFactory<T> consumerFactory;
private final ReactivePulsarContainerProperties<T> containerProperties;
private Boolean autoStartup;
private MessageConverter messageConverter;
private Boolean fluxListener;
public DefaultReactivePulsarListenerContainerFactory(ReactivePulsarConsumerFactory<T> consumerFactory,
ReactivePulsarContainerProperties<T> containerProperties) {
this.consumerFactory = consumerFactory;
this.containerProperties = containerProperties;
}
protected ReactivePulsarConsumerFactory<T> getConsumerFactory() {
return this.consumerFactory;
}
public ReactivePulsarContainerProperties<T> 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<T> createContainerInstance(
ReactivePulsarListenerEndpoint<T> endpoint) {
ReactivePulsarContainerProperties<T> 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<T>) 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<T> createListenerContainer(
ReactivePulsarListenerEndpoint<T> endpoint) {
DefaultReactivePulsarMessageListenerContainer<T> instance = createContainerInstance(endpoint);
if (endpoint instanceof AbstractReactivePulsarListenerEndpoint) {
configureEndpoint((AbstractReactivePulsarListenerEndpoint<T>) endpoint);
}
endpoint.setupListenerContainer(instance, this.messageConverter);
initializeContainer(instance, endpoint);
return instance;
}
private void configureEndpoint(AbstractReactivePulsarListenerEndpoint<T> aplEndpoint) {
if (aplEndpoint.getFluxListener() == null) {
JavaUtils.INSTANCE.acceptIfNotNull(this.fluxListener, aplEndpoint::setFluxListener);
}
}
@Override
public DefaultReactivePulsarMessageListenerContainer<T> createContainer(String... topics) {
ReactivePulsarListenerEndpoint<T> endpoint = new ReactivePulsarListenerEndpointAdapter<>() {
@Override
public List<String> getTopics() {
return Arrays.asList(topics);
}
};
DefaultReactivePulsarMessageListenerContainer<T> container = createContainerInstance(endpoint);
initializeContainer(container, endpoint);
return container;
}
@SuppressWarnings("unchecked")
private void initializeContainer(DefaultReactivePulsarMessageListenerContainer<T> instance,
ReactivePulsarListenerEndpoint<T> endpoint) {
Boolean autoStart = endpoint.getAutoStartup();
if (autoStart != null) {
instance.setAutoStartup(autoStart);
}
else if (this.autoStartup != null) {
instance.setAutoStartup(this.autoStartup);
}
}
}

282
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java Normal file
View File

@@ -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 <V> Message payload type
* @author Christophe Bornet
*/
public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePulsarListenerEndpoint<V> {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
private Object bean;
private Method method;
private MessageHandlerMethodFactory messageHandlerMethodFactory;
private SmartMessageConverter messagingConverter;
private ReactiveMessageConsumerBuilderCustomizer<V> 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<V> createMessageHandler(
ReactivePulsarMessageListenerContainer<V> container, @Nullable MessageConverter messageConverter) {
Assert.state(this.messageHandlerMethodFactory != null,
"Could not create message listener - MessageHandlerMethodFactory not set");
PulsarMessagingMessageListenerAdapter<V> 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<MethodParameter> 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<? extends GeneratedMessageV3>) 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<V> 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<Class<?>, 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<? extends Class<?>> keySchema = SchemaUtils.getSchema(keyClass);
Schema<? extends Class<?>> 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<V> messageListener) {
InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory
.createInvocableHandlerMethod(getBean(), getMethod());
return new HandlerAdapter(invocableHandlerMethod);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
protected PulsarMessagingMessageListenerAdapter<V> createMessageListenerInstance(
@Nullable MessageConverter messageConverter) {
PulsarMessagingMessageListenerAdapter<V> listener;
if (isFluxListener()) {
listener = new PulsarReactiveStreamingMessagingMessageListenerAdapter<V>(this.bean, this.method);
}
else {
listener = new PulsarReactiveOneByOneMessagingMessageListenerAdapter<V>(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<V> consumerCustomizer) {
this.consumerCustomizer = consumerCustomizer;
}
}

31
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java Normal file
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
*/
public interface ReactivePulsarListenerContainerFactory<T>
extends ListenerContainerFactory<ReactivePulsarMessageListenerContainer<T>, ReactivePulsarListenerEndpoint<T>> {
}

32
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java Normal file
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
*/
public interface ReactivePulsarListenerEndpoint<T> extends ListenerEndpoint<ReactivePulsarMessageListenerContainer<T>> {
}

83
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java Normal file
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
*/
public class ReactivePulsarListenerEndpointAdapter<T> implements ReactivePulsarListenerEndpoint<T> {
@Override
public String getId() {
return null;
}
@Override
public String getSubscriptionName() {
return null;
}
@Override
public SubscriptionType getSubscriptionType() {
return SubscriptionType.Exclusive;
}
@Override
public List<String> getTopics() {
return Collections.emptyList();
}
@Override
public String getTopicPattern() {
return null;
}
@Override
public Boolean getAutoStartup() {
return null;
}
@Override
public void setupListenerContainer(ReactivePulsarMessageListenerContainer<T> listenerContainer,
MessageConverter messageConverter) {
}
@Override
public SchemaType getSchemaType() {
return null;
}
@Nullable
@Override
public Integer getConcurrency() {
return null;
}
}

46
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java Normal file
View File

@@ -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.
*
* <p>
* 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 <T> Message payload type.
* @author Christophe Bornet
*/
public class ReactivePulsarListenerEndpointRegistry<T>
extends ListenerEndpointRegistry<ReactivePulsarMessageListenerContainer<T>, ReactivePulsarListenerEndpoint<T>> {
public ReactivePulsarListenerEndpointRegistry() {
super(ReactivePulsarMessageListenerContainer.class);
}
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java
View File

@@ -39,7 +39,7 @@ import io.micrometer.observation.ObservationRegistry;
* @author Soby Chacko
* @author Alexander Preuß
*/
public abstract class AbstractPulsarMessageListenerContainer<T> implements PulsarMessageListenerContainer,
public non-sealed abstract class AbstractPulsarMessageListenerContainer<T> implements PulsarMessageListenerContainer,
BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware {
protected final LogAccessor logger = new LogAccessor(this.getClass());

67
spring-pulsar/src/main/java/org/springframework/pulsar/listener/ListenerContainerRegistry.java Normal file
View File

@@ -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<String> getListenerContainerIds();
/**
* Return the managed listener container instance(s).
* @return the managed listener container instance(s).
* @see #getAllListenerContainers()
*/
Collection<? extends MessageListenerContainer> 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<? extends MessageListenerContainer> getAllListenerContainers();
}

39
spring-pulsar/src/main/java/org/springframework/pulsar/listener/MessageListenerContainer.java Normal file
View File

@@ -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
}
}

15
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java
View File

@@ -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");
}

24
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java
View File

@@ -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 <V> payload type.
* @author Soby Chacko
* @author Christophe Bornet
*/
public abstract class PulsarMessagingMessageListenerAdapter<V> {
@@ -73,6 +76,8 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
private boolean isSpringMessageList;
private boolean isSpringMessageFlux;
private boolean isSpringMessage;
private boolean isConsumerRecords;
@@ -130,6 +135,10 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
return this.isSpringMessageList;
}
protected boolean isSpringMessageFlux() {
return this.isSpringMessageFlux;
}
protected org.springframework.messaging.Message<?> toMessagingMessage(Message<V> record, Consumer<V> consumer) {
return getMessageConverter().toMessage(record, consumer, getType());
}
@@ -167,7 +176,8 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
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<V> {
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);
}

67
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java Normal file
View File

@@ -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 <V> payload type.
* @author Christophe Bornet
*/
public class PulsarReactiveOneByOneMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements ReactivePulsarOneByOneMessageHandler<V> {
public PulsarReactiveOneByOneMessagingMessageListenerAdapter(Object bean, Method method) {
super(bean, method);
}
@Override
@SuppressWarnings("unchecked")
public Publisher<Void> received(Message<V> 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<Void>) invokeHandler(theRecord, message, null, null);
}
catch (Exception e) {
return Mono.error(e);
}
}
}

59
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java Normal file
View File

@@ -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 <V> payload type.
* @author Christophe Bornet
*/
public class PulsarReactiveStreamingMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements ReactivePulsarStreamingHandler<V> {
public PulsarReactiveStreamingMessagingMessageListenerAdapter(Object bean, Method method) {
super(bean, method);
}
@Override
@SuppressWarnings("unchecked")
public Flux<MessageResult<Void>> received(Flux<Message<V>> records) {
Flux<?> theRecords = records;
if (isSpringMessageFlux()) {
theRecords = records.map(record -> toMessagingMessage(record, null));
}
try {
return (Flux<MessageResult<Void>>) invokeHandler(theRecords, null, null, null);
}
catch (Exception e) {
return Flux.error(e);
}
}
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java
View File

@@ -151,7 +151,7 @@ public non-sealed class DefaultReactivePulsarMessageListenerContainer<T>
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());

19
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java
View File

@@ -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<T> {
private List<String> topics;
private Collection<String> topics;
private Pattern topicsPattern;
@@ -41,6 +42,8 @@ public class ReactivePulsarContainerProperties<T> {
private Schema<T> schema;
private SchemaType schemaType;
private ReactivePulsarMessageHandler messageHandler;
private Duration handlingTimeout = Duration.ofMinutes(2);
@@ -73,11 +76,19 @@ public class ReactivePulsarContainerProperties<T> {
this.schema = schema;
}
public List<String> getTopics() {
public SchemaType getSchemaType() {
return this.schemaType;
}
public void setSchemaType(SchemaType schemaType) {
this.schemaType = schemaType;
}
public Collection<String> getTopics() {
return this.topics;
}
public void setTopics(List<String> topics) {
public void setTopics(Collection<String> topics) {
this.topics = topics;
}

14
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java
View File

@@ -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<T>
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<T> getContainerProperties() {
throw new UnsupportedOperationException("This container doesn't support retrieving its properties");
}

2
spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java
View File

@@ -64,7 +64,7 @@ public class ConcurrentPulsarMessageListenerContainerTests {
PulsarListenerEndpoint pulsarListenerEndpoint = mock(PulsarListenerEndpoint.class);
when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1);
ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = containerFactory
AbstractPulsarMessageListenerContainer<String> concurrentContainer = containerFactory
.createListenerContainer(pulsarListenerEndpoint);
PulsarContainerProperties pulsarContainerProperties = concurrentContainer.getContainerProperties();

2
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -127,7 +127,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
}
@Bean
PulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
PulsarListenerContainerFactory pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> pulsarConsumerFactory) {
final ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
pulsarConsumerFactory, new PulsarContainerProperties(), null);

613
spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java Normal file
View File

@@ -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<String> pulsarTemplate;
@Autowired
private PulsarClient pulsarClient;
@Configuration(proxyBeanMethods = false)
@EnablePulsar
public static class TopLevelConfig {
@Bean
public PulsarProducerFactory<String> 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<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
return new PulsarTemplate<>(pulsarProducerFactory);
}
@Bean
public ReactivePulsarConsumerFactory<String> pulsarConsumerFactory(ReactivePulsarClient pulsarClient) {
return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, new MutableReactiveMessageConsumerSpec());
}
@Bean
ReactivePulsarListenerContainerFactory<String> reactivePulsarListenerContainerFactory(
ReactivePulsarConsumerFactory<String> 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<String> registry;
@Test
void testPulsarListener() throws Exception {
ReactivePulsarContainerProperties<String> 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<String> 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<Void> listen1(String message) {
latch1.countDown();
return Mono.empty();
}
@ReactivePulsarListener(consumerCustomizer = "listen2Customizer")
Mono<Void> listen2(String message) {
latch2.countDown();
return Mono.empty();
}
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> 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<Void> listen3(String message) {
latch3.countDown();
return Mono.empty();
}
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> 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<MessageResult<Void>> listen4(Flux<Message<String>> 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<MessageResult<Void>> listen5(Flux<org.springframework.messaging.Message<String>> 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<String> 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<Void> listen(String msg) {
latch.countDown();
return Mono.error(new RuntimeException("fail " + msg));
}
@ReactivePulsarListener(id = "dlqListener", topics = "dlpt-dlq-topic",
consumerCustomizer = "consumerCustomizer")
Mono<Void> listenDlq(String msg) {
dlqLatch.countDown();
return Mono.empty();
}
@Bean
DeadLetterPolicy deadLetterPolicy() {
return DeadLetterPolicy.builder().maxRedeliverCount(1).deadLetterTopic("dlpt-dlq-topic").build();
}
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> 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<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap());
PulsarTemplate<User> 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<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap());
PulsarTemplate<User> 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<KeyValue<String, Integer>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
pulsarClient, Collections.emptyMap());
PulsarTemplate<KeyValue<String, Integer>> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<KeyValue<String, Integer>> 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<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap());
PulsarTemplate<Proto.Person> 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<Void> listenJson(User message) {
jsonLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "avroListener", topics = "avro-topic", schemaType = SchemaType.AVRO,
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenAvro(User message) {
avroLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "keyvalueListener", topics = "keyvalue-topic",
schemaType = SchemaType.KEY_VALUE, consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenKeyvalue(KeyValue<String, Integer> message) {
keyvalueLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "protobufListener", topics = "protobuf-topic",
schemaType = SchemaType.PROTOBUF, consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> 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<String> capturedData = new AtomicReference<>();
static AtomicReference<MessageId> messageId = new AtomicReference<>();
static AtomicReference<String> topicName = new AtomicReference<>();
static AtomicReference<String> fooValue = new AtomicReference<>();
static AtomicReference<byte[]> 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<Void> 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<Void> pulsarMessageListenerWithHeaders(Message<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.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<Void> springMessagingMessageListenerWithHeaders(org.springframework.messaging.Message<String> 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);
}
}
}
}

2
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java
View File

@@ -155,7 +155,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
}
@Bean
PulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
PulsarListenerContainerFactory pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> pulsarConsumerFactory, ObservationRegistry observationRegistry) {
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
new PulsarContainerProperties(), observationRegistry);

2
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java
View File

@@ -202,7 +202,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
}
@Bean
PulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
PulsarListenerContainerFactory pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> pulsarConsumerFactory, ObservationRegistry observationRegistry) {
PulsarContainerProperties containerProperties = new PulsarContainerProperties();
containerProperties.setObservationConvention(new DefaultPulsarListenerObservationConvention() {