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

Minor cleanup and refactoring (#24)

Browse Source
This commit is contained in:
Chris Bono
2022-07-11 13:06:14 -05:00
committed by GitHub
parent fa606af64d
commit ecd0a30c83
15 changed files with 156 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java
View File

@@ -22,8 +22,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.annotation.EnablePulsar;
import org.springframework.pulsar.config.PulsarListenerConfigUtils;
import org.springframework.pulsar.config.PulsarListenerContainerFactoryImpl;
import org.springframework.pulsar.config.DefaultPulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerBeanNames;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -31,6 +31,7 @@ import org.springframework.pulsar.listener.PulsarContainerProperties;
* Configuration for Pulsar annotation-driven support.
*
* @author Soby Chacko
* @author Chris Bono
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(EnablePulsar.class)
@@ -44,9 +45,9 @@ public class PulsarAnnotationDrivenConfiguration {
@Bean
@ConditionalOnMissingBean(name = "pulsarListenerContainerFactory")
PulsarListenerContainerFactoryImpl<?, ?> pulsarListenerContainerFactory(
DefaultPulsarListenerContainerFactory<?, ?> pulsarListenerContainerFactory(
ObjectProvider<PulsarConsumerFactory<Object>> pulsarConsumerFactory) {
PulsarListenerContainerFactoryImpl<Object, Object> factory = new PulsarListenerContainerFactoryImpl<>();
DefaultPulsarListenerContainerFactory<Object, Object> factory = new DefaultPulsarListenerContainerFactory<>();
final PulsarConsumerFactory<Object> pulsarConsumerFactory1 = pulsarConsumerFactory.getIfAvailable();
factory.setPulsarConsumerFactory(pulsarConsumerFactory1);
@@ -68,7 +69,7 @@ public class PulsarAnnotationDrivenConfiguration {
@Configuration(proxyBeanMethods = false)
@EnablePulsar
@ConditionalOnMissingBean(name = PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
@ConditionalOnMissingBean(name = PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
static class EnableKafkaConfiguration {
}

5
spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java
View File

@@ -33,7 +33,10 @@ import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.PulsarTemplate;
/**
* Tests for {@link PulsarListener}.
*
* @author Soby Chacko
* @author Chris Bono
*/
class PulsarListenerTests extends AbstractContainerBaseTests {
@@ -86,7 +89,7 @@ class PulsarListenerTests extends AbstractContainerBaseTests {
@Import(PulsarAutoConfiguration.class)
public static class BatchListenerConfig {
@PulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive", batch = "true")
@PulsarListener(subscriptionName = "test-exclusive-sub-2", topics = "hello-pulsar-exclusive", batch = true)
public void listen(List<String> foo) {
foo.forEach(t -> latch2.countDown());
}

5
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/EnablePulsar.java
View File

@@ -25,10 +25,10 @@ import java.lang.annotation.Target;
import org.springframework.context.annotation.Import;
/**
* Enable Pulsar listener annotated endpoints that are created under the covers by a
* {@link org.springframework.pulsar.config.AbstractPulsarListenerContainerFactory}.
* Enables detection of {@link PulsarListener} annotations on any Spring-managed bean in the container.
*
* @author Soby Chacko
* @author Chris Bono
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@@ -36,4 +36,3 @@ import org.springframework.context.annotation.Import;
@Import(PulsarListenerConfigurationSelector.class)
public @interface EnablePulsar {
}

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

@@ -20,7 +20,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.pulsar.config.PulsarListenerConfigUtils;
import org.springframework.pulsar.config.PulsarListenerBeanNames;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
/**
@@ -32,6 +32,7 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
* annotation.
*
* @author Soby Chacko
* @author Chris Bono
*
* @see PulsarListenerAnnotationBeanPostProcessor
* @see PulsarListenerEndpointRegistry
@@ -41,15 +42,13 @@ public class PulsarBootstrapConfiguration implements ImportBeanDefinitionRegistr
@Override
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
if (!registry.containsBeanDefinition(
PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
registry.registerBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
if (!registry.containsBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
registry.registerBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
new RootBeanDefinition(PulsarListenerAnnotationBeanPostProcessor.class));
}
if (!registry.containsBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
registry.registerBeanDefinition(PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
if (!registry.containsBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
registry.registerBeanDefinition(PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
new RootBeanDefinition(PulsarListenerEndpointRegistry.class));
}
}

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

@@ -25,6 +25,8 @@ import java.lang.annotation.Target;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
/**
* Annotation that marks a method to be the target of a Pulsar message listener on the
@@ -43,6 +45,7 @@ import org.springframework.messaging.handler.annotation.MessageMapping;
* </p>
*
* @author Soby Chacko
* @author Chris Bono
*/
@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@@ -55,7 +58,7 @@ public @interface PulsarListener {
* <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 org.springframework.pulsar.config.PulsarListenerEndpointRegistry#getListenerContainer(String)
* @see PulsarListenerEndpointRegistry#getListenerContainer(String)
*/
String id() default "";
@@ -74,8 +77,13 @@ public @interface PulsarListener {
SchemaType schemaType() default SchemaType.NONE;
/**
* Specific container factory to use on this listener.
* @return {@code containerFactory} to use on this Pulsar listener.
* The bean name of the {@link PulsarListenerContainerFactory}
* to use to create the message listener container responsible to serve this endpoint.
* <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 "";
@@ -108,7 +116,7 @@ public @interface PulsarListener {
*
* @return whether this listener is in batch mode or not.
*/
String batch() default "";
boolean batch() default false;
/**
* A pseudo bean name used in SpEL expressions within this annotation to reference
@@ -120,5 +128,25 @@ public @interface PulsarListener {
*/
String beanRef() default "__listener";
/**
* Pulsar consumer properties; they will supersede any properties with the same name
* defined in the consumer factory (if the consumer factory supports property overrides).
* <p>
* <b>Supported Syntax</b>
* <p>The supported syntax for key-value pairs is the same as the
* syntax defined for entries in a Java
* {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
* <ul>
* <li>{@code key=value}</li>
* <li>{@code key:value}</li>
* <li>{@code key value}</li>
* </ul>
* {@code group.id} and {@code client.id} are ignored.
* <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
* SpEL expressions must resolve to a {@link String}, a @{link String[]} or a
* {@code Collection<String>} where each member of the array or collection is a
* property name + value with the above formats.
* @return the properties.
*/
String[] properties() default {};
}

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

@@ -47,7 +47,6 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -61,7 +60,6 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.OrderComparator;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationUtils;
@@ -80,8 +78,9 @@ import org.springframework.messaging.handler.annotation.support.MessageHandlerMe
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.pulsar.config.MethodPulsarListenerEndpoint;
import org.springframework.pulsar.config.PulsarListenerConfigUtils;
import org.springframework.pulsar.config.PulsarListenerBeanNames;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpoint;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.util.Assert;
@@ -92,12 +91,12 @@ import org.springframework.validation.Validator;
/**
* Bean post-processor that registers methods annotated with {@link PulsarListener}
* to be invoked by a Pulsar message listener container created under the covers
* by a {@link org.springframework.pulsar.config.PulsarListenerContainerFactory}
* by a {@link PulsarListenerContainerFactory}
* according to the parameters of the annotation.
*
* <p>Annotated methods can use flexible arguments as defined by {@link PulsarListener}.
*
* <p>This post-processor is automatically registered by Spring's {@link EnablePulsar}
* <p>This post-processor is automatically registered by the {@link EnablePulsar}
* annotation.
*
* <p>Auto-detect any {@link PulsarListenerConfigurer} instances in the container,
@@ -109,13 +108,14 @@ import org.springframework.validation.Validator;
* @param <V> the value type.
*
* @author Soby Chacko
* @author Chris Bono
*
* @see PulsarListener
* @see EnablePulsar
* @see PulsarListenerConfigurer
* @see PulsarListenerEndpointRegistrar
* @see PulsarListenerEndpointRegistry
* @see org.springframework.pulsar.config.PulsarListenerEndpoint
* @see PulsarListenerEndpoint
* @see MethodPulsarListenerEndpoint
*/
public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
@@ -136,17 +136,20 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
private static final String GENERATED_ID_PREFIX = "org.springframework.Pulsar.PulsarListenerEndpointContainer#";
private ApplicationContext applicationContext;
private BeanFactory beanFactory;
private BeanExpressionResolver resolver;
private BeanExpressionContext expressionContext;
private PulsarListenerEndpointRegistry endpointRegistry;
private String defaultContainerFactoryBeanName = DEFAULT_PULSAR_LISTENER_CONTAINER_FACTORY_BEAN_NAME;
private final PulsarListenerEndpointRegistrar registrar = new PulsarListenerEndpointRegistrar();
private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory =
new PulsarHandlerMethodFactoryAdapter();
private final PulsarHandlerMethodFactoryAdapter messageHandlerMethodFactory = new PulsarHandlerMethodFactoryAdapter();
private Charset charset = StandardCharsets.UTF_8;
@@ -179,23 +182,20 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
buildEnhancer();
}
private void buildEnhancer() {
if (this.applicationContext != null) {
Map<String, AnnotationEnhancer> enhancersMap =
this.applicationContext.getBeansOfType(AnnotationEnhancer.class, false, false);
if (enhancersMap.size() > 0) {
List<AnnotationEnhancer> enhancers = enhancersMap.values()
.stream()
.sorted(new OrderComparator())
.collect(Collectors.toList());
List<AnnotationEnhancer> enhancers = this.applicationContext
.getBeanProvider(AnnotationEnhancer.class, false)
.orderedStream()
.toList();
if (!enhancers.isEmpty()) {
this.enhancer = (attrs, element) -> {
Map<String, Object> newAttrs = attrs;
for (AnnotationEnhancer enh : enhancers) {
newAttrs = enh.apply(newAttrs, element);
attrs = enh.apply(attrs, element);
}
return attrs;
};
@@ -207,20 +207,15 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
public void afterSingletonsInstantiated() {
this.registrar.setBeanFactory(this.beanFactory);
if (this.beanFactory instanceof ListableBeanFactory) {
Map<String, PulsarListenerConfigurer> instances =
((ListableBeanFactory) this.beanFactory).getBeansOfType(PulsarListenerConfigurer.class);
for (PulsarListenerConfigurer configurer : instances.values()) {
configurer.configurePulsarListeners(this.registrar);
}
}
this.beanFactory.getBeanProvider(PulsarListenerConfigurer.class)
.forEach(c -> c.configurePulsarListeners(this.registrar));
if (this.registrar.getEndpointRegistry() == null) {
if (this.endpointRegistry == null) {
Assert.state(this.beanFactory != null,
"BeanFactory must be set to find endpoint registry by bean name");
this.endpointRegistry = this.beanFactory.getBean(
PulsarListenerConfigUtils.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
PulsarListenerBeanNames.PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
PulsarListenerEndpointRegistry.class);
}
this.registrar.setEndpointRegistry(this.endpointRegistry);
@@ -230,7 +225,7 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);
}
// Set the custom handler method factory once resolved by the configurer
// Set the custom handler method factory once resolved by the configurer - otherwise register default formatters
MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();
if (handlerMethodFactory != null) {
this.messageHandlerMethodFactory.setHandlerMethodFactory(handlerMethodFactory);
@@ -252,7 +247,6 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
Class<?> targetClass = AopUtils.getTargetClass(bean);
Collection<PulsarListener> classLevelListeners = findListenerAnnotations(targetClass);
Map<Method, Set<PulsarListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
(MethodIntrospector.MetadataLookup<Set<PulsarListener>>) method -> {
Set<PulsarListener> listenerMethods = findListenerAnnotations(method);
@@ -360,9 +354,7 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
}
resolvePulsarProperties(endpoint, pulsarListener.properties());
if (StringUtils.hasText(pulsarListener.batch())) {
endpoint.setBatchListener(Boolean.parseBoolean(pulsarListener.batch()));
}
endpoint.setBatchListener(pulsarListener.batch());
endpoint.setBeanFactory(this.beanFactory);
}
@@ -588,26 +580,9 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
private void addFormatters(FormatterRegistry registry) {
for (Converter<?, ?> converter : getBeansOfType(Converter.class)) {
registry.addConverter(converter);
}
for (GenericConverter converter : getBeansOfType(GenericConverter.class)) {
registry.addConverter(converter);
}
for (Formatter<?> formatter : getBeansOfType(Formatter.class)) {
registry.addFormatter(formatter);
}
}
private <T> Collection<T> getBeansOfType(Class<T> type) {
if (PulsarListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ListableBeanFactory) {
return ((ListableBeanFactory) PulsarListenerAnnotationBeanPostProcessor.this.beanFactory)
.getBeansOfType(type)
.values();
}
else {
return Collections.emptySet();
}
this.beanFactory.getBeanProvider(Converter.class).forEach(registry::addConverter);
this.beanFactory.getBeanProvider(GenericConverter.class).forEach(registry::addConverter);
this.beanFactory.getBeanProvider(Formatter.class).forEach(registry::addFormatter);
}
@Override
@@ -802,4 +777,3 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V> implements BeanPost
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurer.java
View File

@@ -16,14 +16,32 @@
package org.springframework.pulsar.annotation;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpoint;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
/**
* Allow custom configuration on PulsarListener endpoint registry.
* Optional interface to be implemented by Spring managed bean willing to
* customize how Pulsar listener endpoints are configured. Typically used
* to define the default {@link PulsarListenerContainerFactory} to use or
* for registering Pulsar endpoints in a <em>programmatic</em> fashion as
* opposed to the <em>declarative</em> approach of using the
* {@link PulsarListener} annotation.
*
* @author Soby Chacko
* @author Chris Bono
*
* @see PulsarListenerEndpointRegistrar
*/
public interface PulsarListenerConfigurer {
/**
* Callback allowing a {@link PulsarListenerEndpointRegistry} and specific
* {@link PulsarListenerEndpoint} instances to be registered against the
* given {@link PulsarListenerEndpointRegistrar}. The default
* {@link PulsarListenerContainerFactory} can also be customized.
* @param registrar the registrar to be configured
*/
void configurePulsarListeners(PulsarListenerEndpointRegistrar registrar);
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListeners.java
View File

@@ -23,9 +23,17 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Container annotation for aggregating several {@link PulsarListener} annotations.
* Container annotation that aggregates several {@link PulsarListener} annotations.
* <p>
* Can be used natively, declaring several nested {@link PulsarListener} annotations.
* Can also be used in conjunction with Java 8's support for repeatable annotations,
* where {@link PulsarListener} can simply be declared several times on the same method
* (or class), implicitly generating this container annotation.
*
* @author Soby Chacko
* @author Chris Bono
*
* @see PulsarListener
*/
@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)

3
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerContainerFactoryImpl.java → spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarListenerContainerFactory.java
View File

@@ -32,8 +32,9 @@ import org.springframework.util.StringUtils;
* @param <T> message type in the listener.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class PulsarListenerContainerFactoryImpl<C, T> extends AbstractPulsarListenerContainerFactory<DefaultPulsarMessageListenerContainer<T>, T> {
public class DefaultPulsarListenerContainerFactory<C, T> extends AbstractPulsarListenerContainerFactory<DefaultPulsarMessageListenerContainer<T>, T> {
@Override
protected DefaultPulsarMessageListenerContainer<T> createContainerInstance(PulsarListenerEndpoint endpoint) {

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

@@ -16,17 +16,23 @@
package org.springframework.pulsar.config;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.core.log.LogAccessor;
/**
* {@link org.springframework.beans.factory.FactoryBean} implementation for the {@link PulsarClient}.
* {@link FactoryBean} implementation for the {@link PulsarClient}.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class PulsarClientFactoryBean extends AbstractFactoryBean<PulsarClient> {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final PulsarClientConfiguration pulsarClientConfiguration;
public PulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
@@ -48,7 +54,7 @@ public class PulsarClientFactoryBean extends AbstractFactoryBean<PulsarClient> {
@Override
protected void destroyInstance(PulsarClient instance) throws Exception {
if (instance != null) {
this.logger.info("Closing the client: " + instance);
this.logger.info(() -> "Closing client " + instance);
instance.close();
}
}

16
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerConfigUtils.java → spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java
View File

@@ -17,29 +17,23 @@
package org.springframework.pulsar.config;
/**
* Configuration constants for internal sharing across subpackages.
* Constants for bean names that need to be accessed in multiple places by name.
*
* @author Soby Chacko
* @author Chris Bono
*/
public abstract class PulsarListenerConfigUtils {
public abstract class PulsarListenerBeanNames {
/**
* The bean name of the internally managed Kafka listener annotation processor.
* The bean name of the internally managed Pulsar listener annotation processor.
*/
public static final String PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME =
"org.springframework.pulsar.config.internalKafkaListenerAnnotationProcessor";
/**
* The bean name of the internally managed Kafka listener endpoint registry.
* The bean name of the internally managed Pulsar listener endpoint registry.
*/
public static final String PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME =
"org.springframework.pulsar.config.internalKafkaListenerEndpointRegistry";
/**
* The bean name of the internally managed Kafka consumer back off manager.
*/
public static final String PULSAR_CONSUMER_BACK_OFF_MANAGER_BEAN_NAME =
"org.springframework.pulsar.config.internalKafkaConsumerBackOffManager";
}

22
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -32,15 +32,16 @@ import org.springframework.core.log.LogAccessor;
import org.springframework.util.CollectionUtils;
/**
* Default implementation for {@link PulsarProducerFactory}.
* Default implementation of {@link PulsarProducerFactory}.
*
* @param <T> producer type.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>, DisposableBean {
protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final Map<String, Object> producerConfig = new HashMap<>();
@@ -57,25 +58,18 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
@Override
public Producer<T> createProducer(Schema<T> schema) throws PulsarClientException {
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
if (!CollectionUtils.isEmpty(this.producerConfig)) {
producerBuilder.loadConf(this.producerConfig);
}
this.producer = producerBuilder.create();
return this.producer;
return createProducer(schema, null);
}
@Override
public Producer<T> createProducer(Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
if (!CollectionUtils.isEmpty(this.producerConfig)) {
producerBuilder.loadConf(this.producerConfig);
}
producerBuilder.messageRouter(messageRouter);
if (messageRouter != null) {
producerBuilder.messageRouter(messageRouter);
}
this.producer = producerBuilder.create();
return this.producer;
}
@@ -87,7 +81,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
@Override
public void destroy() throws Exception {
this.logger.info("Closing the producer");
this.logger.info("Closing producer");
this.producer.close();
}
}

25
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
View File

@@ -24,17 +24,38 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
/**
* Pulsar producer factory interface.
* The strategy to produce a {@link Producer} instance(s).
*
* @param <T> producer payload type.
* @param <T> producer payload type
*
* @author Soby Chacko
* @author Chris Bono
*/
public interface PulsarProducerFactory<T> {
/**
* Create a producer.
*
* @param schema the schema of the messages to be sent
* @return the producer
* @throws PulsarClientException if any error occurs
*/
Producer<T> createProducer(Schema<T> schema) throws PulsarClientException;
/**
* Create a producer.
*
* @param schema the schema of the messages to be sent
* @param messageRouter the optional message router to use
* @return the producer
* @throws PulsarClientException if any error occurs
*/
Producer<T> createProducer(Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException;
/**
* Return a map of configuration options to use when creating producers.
*
* @return the map of configuration options
*/
Map<String, Object> getProducerConfig();
}

8
src/api/overview.html
View File

@@ -1,12 +1,12 @@
<html>
<body>
This document is the API specification for Spring for Apache Kafka project
This document is the API specification for Spring for Apache Pulsar project
<hr>
<div id="overviewBody">
<p>
For further API reference and developer documentation, see the
<a href="https://docs.spring.io/spring-kafka/reference" target="_top">Spring
for Apache Kafka reference documentation</a>.
<a href="https://docs.spring.io/spring-pulsar/reference" target="_top">Spring
for Apache Pulsar reference documentation</a>.
That documentation contains more detailed, developer-targeted
descriptions, with conceptual overviews, definitions of terms,
workarounds, and working code examples.
@@ -14,7 +14,7 @@ This document is the API specification for Spring for Apache Kafka project
<p>
If you are interested in commercial consultancy, and
support for Spring for Apache Kafka, please visit <a href="https://spring.io/" target="_top">
support for Spring for Apache Pulsar, please visit <a href="https://spring.io/" target="_top">
https://spring.io/</a>
</p>
</div>

4
src/dist/notice.txt vendored
View File

@@ -1,5 +1,5 @@
Spring for Apache Kafka Framework ${version}
Copyright (c) 2016-${copyright} Pivotal, Inc.
Spring for Apache Pulsar Framework ${version}
Copyright (c) 2022-${copyright} VMware, Inc.
This product is licensed to you under the Apache License, Version 2.0
(the "License"). You may not use this product except in compliance with