Pulsar Reader auto config (imperative)

Resolves #328

Author: sobychacko

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

@@ -35,11 +35,13 @@
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarReaderFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
@@ -157,4 +159,10 @@ public PulsarFunctionAdministration pulsarFunctionAdministration(PulsarAdministr
 				this.properties.getFunction().getPropagateStopFailures());
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	public PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
+		return new DefaultPulsarReaderFactory<>(pulsarClient, this.properties.buildReaderProperties());
+	}
+
 }

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

@@ -2185,7 +2185,7 @@ public Map<String, Object> buildProperties() {
 
 			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 
-			map.from(this::getTopicNames).to(properties.in("topicName"));
+			map.from(this::getTopicNames).to(properties.in("topicNames"));
 			map.from(this::getReceiverQueueSize).to(properties.in("receiverQueueSize"));
 			map.from(this::getReaderName).to(properties.in("readerName"));
 			map.from(this::getSubscriptionName).to(properties.in("subscriptionName"));

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

@@ -50,11 +50,13 @@
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarReaderFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
@@ -490,6 +492,35 @@ private void assertHasProducerFactoryOfType(Class<?> producerFactoryType,
 
 	}
 
+	@Nested
+	class ReaderFactoryAutoConfigurationTests {
+
+		@Test
+		void readerFactoryIsAutoConfiguredByDefault() {
+			contextRunner.run((context) -> assertThat(context).hasNotFailed().hasSingleBean(PulsarReaderFactory.class)
+					.getBean(PulsarReaderFactory.class).isExactlyInstanceOf(DefaultPulsarReaderFactory.class));
+		}
+
+		@Test
+		void readerFactoryCanBeConfigured() {
+			contextRunner.withPropertyValues("spring.pulsar.reader.topic-names=foo",
+					"spring.pulsar.reader.receiver-queue-size=200", "spring.pulsar.reader.reader-name=test-reader",
+					"spring.pulsar.reader.subscription-name=test-subscription",
+					"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" })
+							.hasFieldOrPropertyWithValue("receiverQueueSize", 200)
+							.hasFieldOrPropertyWithValue("readerName", "test-reader")
+							.hasFieldOrPropertyWithValue("subscriptionName", "test-subscription")
+							.hasFieldOrPropertyWithValue("subscriptionRolePrefix", "test-prefix")
+							.hasFieldOrPropertyWithValue("readCompacted", true)
+							.hasFieldOrPropertyWithValue("resetIncludeHead", true)));
+		}
+
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	static class InterceptorTestConfiguration {
 

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

@@ -528,7 +528,7 @@ void readerProperties() {
 					new ReaderConfigurationData<>(), ReaderConfigurationData.class));
 
 			assertThat(readerProps)
-					.hasEntrySatisfying("topicName",
+					.hasEntrySatisfying("topicNames",
 							topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.array(String[].class))
 									.containsExactly("my-topic"))
 					.containsEntry("receiverQueueSize", 100).containsEntry("readerName", "my-reader")