Add boot property support to Pulsar Reader (imperative) (#327)

buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java

@@ -35,6 +35,7 @@ import org.gradle.api.tasks.TaskAction;
  * @author Phillip Webb
  * @author Chris Bono
  * @author Alexander Preuß
+ * @author Soby Chacko
  */
 public class DocumentConfigurationProperties extends DefaultTask {
 
@@ -73,6 +74,9 @@ public class DocumentConfigurationProperties extends DefaultTask {
 			c.accept("spring.pulsar.consumer");
 			c.accept("spring.pulsar.listener");
 		});
+		snippets.add("application-properties.pulsar-reader", "Pulsar Reader Properties", (c) -> {
+			c.accept("spring.pulsar.reader");
+		});
 		snippets.add("application-properties.pulsar-defaults", "Pulsar Defaults Properties", (c) -> c.accept("spring.pulsar.defaults"));
 		snippets.add("application-properties.pulsar-function", "Pulsar Function Properties", (c) -> c.accept("spring.pulsar.function"));
 		snippets.add("application-properties.pulsar-administration", "Pulsar Administration Properties", (c) -> c.accept("spring.pulsar.administration"));

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

@@ -79,6 +79,8 @@ public class PulsarProperties {
 
 	private final Admin admin = new Admin();
 
+	private final Reader reader = new Reader();
+
 	private final Defaults defaults = new Defaults();
 
 	public Consumer getConsumer() {
@@ -109,6 +111,10 @@ public class PulsarProperties {
 		return this.admin;
 	}
 
+	public Reader getReader() {
+		return this.reader;
+	}
+
 	public Defaults getDefaults() {
 		return this.defaults;
 	}
@@ -129,6 +135,10 @@ public class PulsarProperties {
 		return new HashMap<>(this.admin.buildProperties());
 	}
 
+	public Map<String, Object> buildReaderProperties() {
+		return new HashMap<>(this.reader.buildProperties());
+	}
+
 	public static class Consumer {
 
 		/**
@@ -2075,6 +2085,119 @@ public class PulsarProperties {
 
 	}
 
+	public static class Reader {
+
+		/**
+		 * Topic names.
+		 */
+		private String[] topicNames;
+
+		/**
+		 * Size of a consumer's receiver queue.
+		 */
+		private Integer receiverQueueSize;
+
+		/**
+		 * Reader name.
+		 */
+		private String readerName;
+
+		/**
+		 * Subscription name.
+		 */
+		private String subscriptionName;
+
+		/**
+		 * Prefix of subscription role.
+		 */
+		private String subscriptionRolePrefix;
+
+		/**
+		 * Whether to read messages from a compacted topic rather than a full message
+		 * backlog of a topic.
+		 */
+		private Boolean readCompacted;
+
+		/**
+		 * Whether the first message to be returned is the one specified by messageId.
+		 */
+		private Boolean resetIncludeHead;
+
+		public String[] getTopicNames() {
+			return this.topicNames;
+		}
+
+		public void setTopicNames(String[] topicNames) {
+			this.topicNames = topicNames;
+		}
+
+		public Integer getReceiverQueueSize() {
+			return this.receiverQueueSize;
+		}
+
+		public void setReceiverQueueSize(Integer receiverQueueSize) {
+			this.receiverQueueSize = receiverQueueSize;
+		}
+
+		public String getReaderName() {
+			return this.readerName;
+		}
+
+		public void setReaderName(String readerName) {
+			this.readerName = readerName;
+		}
+
+		public String getSubscriptionName() {
+			return this.subscriptionName;
+		}
+
+		public void setSubscriptionName(String subscriptionName) {
+			this.subscriptionName = subscriptionName;
+		}
+
+		public String getSubscriptionRolePrefix() {
+			return this.subscriptionRolePrefix;
+		}
+
+		public void setSubscriptionRolePrefix(String subscriptionRolePrefix) {
+			this.subscriptionRolePrefix = subscriptionRolePrefix;
+		}
+
+		public Boolean getReadCompacted() {
+			return this.readCompacted;
+		}
+
+		public void setReadCompacted(Boolean readCompacted) {
+			this.readCompacted = readCompacted;
+		}
+
+		public Boolean getResetIncludeHead() {
+			return this.resetIncludeHead;
+		}
+
+		public void setResetIncludeHead(Boolean resetIncludeHead) {
+			this.resetIncludeHead = resetIncludeHead;
+		}
+
+		public Map<String, Object> buildProperties() {
+
+			PulsarProperties.Properties properties = new Properties();
+
+			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+
+			map.from(this::getTopicNames).to(properties.in("topicName"));
+			map.from(this::getReceiverQueueSize).to(properties.in("receiverQueueSize"));
+			map.from(this::getReaderName).to(properties.in("readerName"));
+			map.from(this::getSubscriptionName).to(properties.in("subscriptionName"));
+			map.from(this::getSubscriptionRolePrefix).to(properties.in("subscriptionRolePrefix"));
+			map.from(this::getReadCompacted).to(properties.in("readCompacted"));
+			map.from(this::getResetIncludeHead).to(properties.in("resetIncludeHead"));
+
+			return properties;
+		}
+
+	}
+
 	public static class Defaults {
 
 		/**

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

@@ -41,6 +41,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
@@ -56,6 +57,7 @@ import org.springframework.pulsar.autoconfigure.PulsarProperties.TypeMapping;
  *
  * @author Chris Bono
  * @author Christophe Bornet
+ * @author Soby Chacko
  */
 public class PulsarPropertiesTests {
 
@@ -503,4 +505,38 @@ public class PulsarPropertiesTests {
 
 	}
 
+	@Nested
+	class ReaderPropertiesTests {
+
+		@Test
+		void readerProperties() {
+			Map<String, String> props = new HashMap<>();
+
+			props.put("spring.pulsar.reader.topic-names", "my-topic");
+			props.put("spring.pulsar.reader.receiver-queue-size", "100");
+			props.put("spring.pulsar.reader.reader-name", "my-reader");
+			props.put("spring.pulsar.reader.subscription-name", "my-subscription");
+			props.put("spring.pulsar.reader.subscription-role-prefix", "sub-role");
+			props.put("spring.pulsar.reader.read-compacted", "true");
+			props.put("spring.pulsar.reader.reset-include-head", "true");
+			bind(props);
+
+			Map<String, Object> readerProps = properties.buildReaderProperties();
+
+			// Verify that the props can be loaded in a ReaderBuilder
+			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(readerProps,
+					new ReaderConfigurationData<>(), ReaderConfigurationData.class));
+
+			assertThat(readerProps)
+					.hasEntrySatisfying("topicName",
+							topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.array(String[].class))
+									.containsExactly("my-topic"))
+					.containsEntry("receiverQueueSize", 100).containsEntry("readerName", "my-reader")
+					.containsEntry("subscriptionName", "my-subscription")
+					.containsEntry("subscriptionRolePrefix", "sub-role").containsEntry("readCompacted", true)
+					.containsEntry("resetIncludeHead", true);
+		}
+
+	}
+
 }