diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java index 4d1be34a..a7bc61be 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java +++ b/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")); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index efeb6da5..af444083 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/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 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 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 { /** diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java index 6f831a60..e00c2bbb 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java +++ b/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 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 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); + } + + } + }