PulsarReader Infrastructure

 - Introducing PulsarReader annotation
 - Annotation bean post processor for PulsarReader
 - PulsarReader container and registry
 - PulsarReader endpoint and registry
 - PulsarReaderConfigurer
 - PulsarReader based listener adapter
 - PulsarReader annotation driven Boot auto configuration

Resolves #353

Author: sobychacko

settings.gradle

@@ -35,6 +35,7 @@ include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-app'
 include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-function'
 include 'spring-pulsar-sample-apps:sample-reactive'
 include 'spring-pulsar-sample-apps:sample-pulsar-binder'
+include 'spring-pulsar-sample-apps:sample-pulsar-reader'
 include 'spring-pulsar-docs'
 include 'spring-pulsar-spring-cloud-stream-binder'
 include 'spring-pulsar-test'

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarBootstrapConfiguration.java

@@ -20,7 +20,7 @@
 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.PulsarAnnotationSupportBeanNames;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
 
 /**
@@ -43,16 +43,16 @@ public class ReactivePulsarBootstrapConfiguration implements ImportBeanDefinitio
 	@Override
 	public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
 		if (!registry.containsBeanDefinition(
-				PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
+				PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
 			registry.registerBeanDefinition(
-					PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
+					PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
 					new RootBeanDefinition(ReactivePulsarListenerAnnotationBeanPostProcessor.class));
 		}
 
-		if (!registry
-				.containsBeanDefinition(PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
+		if (!registry.containsBeanDefinition(
+				PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
 			registry.registerBeanDefinition(
-					PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
+					PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
 					new RootBeanDefinition(ReactivePulsarListenerEndpointRegistry.class));
 		}
 	}

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java

@@ -75,7 +75,7 @@
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.pulsar.annotation.PulsarListenerConfigurer;
-import org.springframework.pulsar.config.PulsarListenerBeanNames;
+import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar;
 import org.springframework.pulsar.reactive.config.MethodReactivePulsarListenerEndpoint;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
@@ -211,7 +211,7 @@ public void afterSingletonsInstantiated() {
 				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,
+						PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
 						ReactivePulsarListenerEndpointRegistry.class);
 			}
 			this.registrar.setEndpointRegistry(this.endpointRegistry);

spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle

@@ -0,0 +1,27 @@
+plugins {
+	id 'org.springframework.pulsar.spring-module'
+	id 'org.springframework.boot' version '3.0.2'
+}
+
+description = 'Spring Pulsar Sample Application (Send and Receive)'
+
+dependencies {
+	implementation project(':spring-pulsar-spring-boot-starter')
+	implementation 'com.google.code.findbugs:jsr305'
+}
+
+bootRun {
+	jvmArgs = [
+			"--add-opens", "java.base/java.lang=ALL-UNNAMED",
+			"--add-opens", "java.base/java.util=ALL-UNNAMED",
+			"--add-opens", "java.base/sun.net=ALL-UNNAMED"
+	]
+}
+
+project.afterEvaluate {
+	project.tasks.publishArtifacts.enabled(false)
+	project.tasks.artifactoryPublish.enabled(false)
+	project.tasks.publishToOssrh.enabled(false)
+	project.tasks.publishMavenJavaPublicationToOssrhRepository.enabled(false)
+	project.tasks.publishAllPublicationsToOssrhRepository.enabled(false)
+}

spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/SpringPulsarReaderBootApp.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2023 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 reader.app;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.pulsar.annotation.PulsarReader;
+import org.springframework.pulsar.core.PulsarTemplate;
+
+@SpringBootApplication
+public class SpringPulsarReaderBootApp {
+
+	private static final Logger logger = LoggerFactory.getLogger(SpringPulsarReaderBootApp.class);
+
+	public static void main(String[] args) {
+		SpringApplication.run(SpringPulsarReaderBootApp.class, args);
+	}
+
+	/*
+	 * Basic publisher using PulsarTemplate<String> and a PulsarReader.
+	 */
+	@Bean
+	ApplicationRunner runner1(PulsarTemplate<String> pulsarTemplate) {
+
+		String topic1 = "pulsar-reader-demo-topic";
+
+		return args -> {
+			for (int i = 0; i < 10; i++) {
+				pulsarTemplate.send(topic1, "This is message " + (i + 1));
+			}
+		};
+	}
+
+	@PulsarReader(id = "my-id", subscriptionName = "pulsar-reader-demo-subscription",
+			topics = "pulsar-reader-demo-topic")
+	void read(String message) {
+		logger.info(message);
+	}
+
+}

spring-pulsar-sample-apps/sample-pulsar-reader/src/main/java/reader/app/package-info.java

@@ -0,0 +1,9 @@
+/**
+ * Package containing sample app for the framework.
+ */
+@NonNullApi
+@NonNullFields
+package reader.app;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;

spring-pulsar-sample-apps/sample-pulsar-reader/src/main/resources/application.yml

@@ -0,0 +1 @@
+

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

@@ -26,12 +26,15 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
-import org.springframework.pulsar.config.PulsarListenerBeanNames;
+import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
+import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.PulsarReaderFactory;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 import org.springframework.pulsar.observation.PulsarListenerObservationConvention;
+import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
 import org.springframework.util.unit.DataSize;
 
 import io.micrometer.observation.ObservationRegistry;
@@ -80,9 +83,25 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 						? observationRegistryProvider.getIfUnique() : null);
 	}
 
+	@Bean
+	@ConditionalOnMissingBean(name = "pulsarReaderContainerFactory")
+	DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(
+			ObjectProvider<PulsarReaderFactory<Object>> readerFactoryProvider, SchemaResolver schemaResolver) {
+
+		PulsarReaderContainerProperties containerProperties = new PulsarReaderContainerProperties();
+		containerProperties.setSchemaResolver(schemaResolver);
+
+		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		PulsarProperties.Reader readerProperties = this.pulsarProperties.getReader();
+		map.from(readerProperties::getTopicNames).to(containerProperties::setTopics);
+
+		return new DefaultPulsarReaderContainerFactory<>(readerFactoryProvider.getIfAvailable(), containerProperties);
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	@EnablePulsar
-	@ConditionalOnMissingBean(name = PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
+	@ConditionalOnMissingBean(name = { PulsarAnnotationSupportBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
+			PulsarAnnotationSupportBeanNames.PULSAR_READER_ANNOTATION_PROCESSOR_BEAN_NAME })
 	static class EnablePulsarConfiguration {
 
 	}

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java

@@ -1216,7 +1216,7 @@ public static class Reader {
 		/**
 		 * Topic names.
 		 */
-		private String[] topicNames;
+		private List<String> topicNames;
 
 		/**
 		 * Size of a consumer's receiver queue.
@@ -1249,11 +1249,11 @@ public static class Reader {
 		 */
 		private Boolean resetIncludeHead;
 
-		public String[] getTopicNames() {
+		public List<String> getTopicNames() {
 			return this.topicNames;
 		}
 
-		public void setTopicNames(String[] topicNames) {
+		public void setTopicNames(List<String> topicNames) {
 			this.topicNames = topicNames;
 		}
 

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java

@@ -22,7 +22,7 @@
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.pulsar.config.PulsarListenerBeanNames;
+import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
@@ -68,7 +68,8 @@ DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainer
 
 	@Configuration(proxyBeanMethods = false)
 	@EnableReactivePulsar
-	@ConditionalOnMissingBean(name = PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
+	@ConditionalOnMissingBean(
+			name = PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
 	static class EnableReactivePulsarConfiguration {
 
 	}

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

@@ -20,6 +20,7 @@
 import static org.assertj.core.api.Assertions.entry;
 import static org.mockito.Mockito.mock;
 
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.api.PulsarClient;
@@ -496,8 +497,7 @@ void readerFactoryCanBeConfigured() {
 					"spring.pulsar.reader.subscription-role-prefix=test-prefix",
 					"spring.pulsar.reader.read-compacted=true", "spring.pulsar.reader.reset-include-head=true")
 					.run((context -> assertThat(context).hasNotFailed().getBean(PulsarReaderFactory.class)
-							.extracting("readerConfig")
-							.hasFieldOrPropertyWithValue("topicNames", new String[] { "foo" })
+							.extracting("readerConfig").hasFieldOrPropertyWithValue("topicNames", List.of("foo"))
 							.hasFieldOrPropertyWithValue("receiverQueueSize", 200)
 							.hasFieldOrPropertyWithValue("readerName", "test-reader")
 							.hasFieldOrPropertyWithValue("subscriptionName", "test-subscription")

spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java

@@ -529,7 +529,7 @@ void readerProperties() {
 
 			assertThat(readerProps)
 					.hasEntrySatisfying("topicNames",
-							topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.array(String[].class))
+							topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.list(String.class))
 									.containsExactly("my-topic"))
 					.containsEntry("receiverQueueSize", 100).containsEntry("readerName", "my-reader")
 					.containsEntry("subscriptionName", "my-subscription")