diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc index 43c58847..aa060a93 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc @@ -1086,22 +1086,20 @@ Spring Boot provides this reader factory which you can further configure by spec ==== PulsarReader Annotation -While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides a convenient annotation called `PulsarReader` that you can use to quickly read from a topic without setting up any reader factories yourselves. +While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides the `PulsarReader` annotation that you can use to quickly read from a topic without setting up any reader factories yourselves. This is similar to the same ideas behind `PulsarListener.` Here is a quick example. [source, java] ---- -@PulsarReader(id = "pulsar-reader-demo-id", subscriptionName = "pulsar-reader-demo-subscription", -topics = "pulsar-reader-demo-topic", startMessageId = "earliest") +@PulsarReader(id = "reader-demo-id", topics = "reader-demo-topic", startMessageId = "earliest") void read(String message) { //... } ---- - -As you can see from the above example, `PulsarReader` is a quick way to read from a Pulsar topic. -The `id` and `subscriptionName` attributes are optional, but always a best practice to provide them. -`PulsarReader` requires `topics` and the `startMessageId` as mandatory attributes. +The `id` attribute is optional, but it is a best practice to provide a value that is meaningful to your application. +When not specified an auto-generated id will be used. +On the other hand, the `topics` and `startMessageId` attributes are mandatory. The `topics` attribute can be a single topic or a comma-separated list of topics. The `startMessageId` attribute instructs the reader to start from a particular message in the topic. The valid values for `startMessageId` are `earliest` or `latest.` @@ -1114,8 +1112,8 @@ You can provide a `@Bean` from `ReaderBuilderCustomizer` and then make it availa [source, java] ---- -@PulsarReader(id = "with-customizer-reader", subscriptionName = "with-customizer-reader-subscription", - topics = "with-customizer-reader-topic", readerCustomizer = "myCustomizer") +@PulsarReader(id = "reader-customizer-demo-id", topics = "reader-customizer-demo-topic", + readerCustomizer = "myCustomizer") void read(String message) { //... } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReader.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReader.java index dfff8a65..3dbb9a06 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReader.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReader.java @@ -46,12 +46,6 @@ public @interface PulsarReader { */ String id() default ""; - /** - * Pulsar subscription name associated with this listener. - * @return the {@code subscriptionName} for this Pulsar listener endpoint. - */ - String subscriptionName() default ""; - /** * Pulsar schema type for this listener. * @return the {@code schemaType} for this listener diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java index 25e7d37b..5d3ca48e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java @@ -210,7 +210,6 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn Object bean, String[] topics) { endpoint.setBean(bean); endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); - endpoint.setSubscriptionName(getEndpointSubscriptionName(pulsarReader)); endpoint.setId(getEndpointId(pulsarReader)); endpoint.setTopics(topics); endpoint.setSchemaType(pulsarReader.schemaType()); @@ -248,13 +247,6 @@ public class PulsarReaderAnnotationBeanPostProcessor extends AbstractPulsarAn } } - private String getEndpointSubscriptionName(PulsarReader pulsarReader) { - if (StringUtils.hasText(pulsarReader.subscriptionName())) { - return resolveExpressionAsString(pulsarReader.subscriptionName(), "subscriptionName"); - } - return GENERATED_ID_PREFIX + this.counter.getAndIncrement(); - } - private String getEndpointId(PulsarReader pulsarReader) { if (StringUtils.hasText(pulsarReader.id())) { return resolveExpressionAsString(pulsarReader.id(), "id"); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderEndpoint.java index 38ebbbd1..758f1fbd 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderEndpoint.java @@ -109,12 +109,6 @@ public abstract class AbstractPulsarReaderEndpoint this.subscriptionName = subscriptionName; } - @Nullable - @Override - public String getSubscriptionName() { - return this.subscriptionName; - } - public void setId(String id) { this.id = id; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java index 99eb94c2..24ff0318 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java @@ -21,7 +21,6 @@ import org.springframework.pulsar.reader.DefaultPulsarMessageReaderContainer; import org.springframework.pulsar.reader.PulsarMessageReaderContainer; import org.springframework.pulsar.reader.PulsarReaderContainerProperties; import org.springframework.util.CollectionUtils; -import org.springframework.util.StringUtils; /** * Concrete implementation for {@link PulsarReaderContainerFactory}. @@ -40,21 +39,13 @@ public class DefaultPulsarReaderContainerFactory @Override protected DefaultPulsarMessageReaderContainer createContainerInstance( PulsarReaderEndpoint endpoint) { - PulsarReaderContainerProperties properties = new PulsarReaderContainerProperties(); properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver()); - if (!CollectionUtils.isEmpty(endpoint.getTopics())) { properties.setTopics(endpoint.getTopics()); } - - if (StringUtils.hasText(endpoint.getSubscriptionName())) { - properties.setSubscriptionName(endpoint.getSubscriptionName()); - } - properties.setSchemaType(endpoint.getSchemaType()); properties.setStartMessageId(endpoint.getStartMessageId()); - return new DefaultPulsarMessageReaderContainer<>(this.getReaderFactory(), properties); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java index 5cb588fd..fdb7209a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java @@ -121,16 +121,12 @@ public class GenericReaderEndpointRegistry factory) { - C readerContainer = factory.createReaderContainer(endpoint); - if (readerContainer instanceof InitializingBean) { try { ((InitializingBean) readerContainer).afterPropertiesSet(); @@ -151,7 +145,6 @@ public class GenericReaderEndpointRegistry { @Nullable String getId(); - /** - * Return the subscription name for this endpoint's container. - * @return the subscription name. - */ - @Nullable - String getSubscriptionName(); - /** * Return the topics for this endpoint's container. * @return the topics. diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpointRegistrar.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpointRegistrar.java index 18a4c842..f9f4537c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpointRegistrar.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpointRegistrar.java @@ -117,7 +117,7 @@ public class PulsarReaderEndpointRegistrar implements BeanFactoryAware, Initiali public void registerEndpoint(PulsarReaderEndpoint endpoint, @Nullable ReaderContainerFactory factory) { Assert.notNull(endpoint, "Endpoint must be set"); - Assert.hasText(endpoint.getSubscriptionName(), "Endpoint id must be set"); + Assert.hasText(endpoint.getId(), "Endpoint id must be set"); // Factory may be null, we defer the resolution right before actually creating the // container PulsarReaderEndpointDescriptor descriptor = new PulsarReaderEndpointDescriptor(endpoint, factory); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java index 85d70d60..ef254151 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java @@ -54,8 +54,6 @@ public class PulsarReaderContainerProperties { private SchemaResolver schemaResolver; - private String subscriptionName; - public Object getReaderListener() { return this.readerListener; } @@ -114,14 +112,6 @@ public class PulsarReaderContainerProperties { this.schema = schema; } - public String getSubscriptionName() { - return this.subscriptionName; - } - - public void setSubscriptionName(String subscriptionName) { - this.subscriptionName = subscriptionName; - } - public SchemaType getSchemaType() { return this.schemaType; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java index 8f25a8f4..97f461b8 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java @@ -50,7 +50,7 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** - * {@link PulsarReader} integration tests. + * Tests for {@link PulsarReader}. * * @author Soby Chacko * @author Chris Bono @@ -113,9 +113,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @Configuration static class PulsarReaderStartMessageIdEarliest { - @PulsarReader(id = "pulsarReaderBasicScenario-id-1", - subscriptionName = "pulsarReaderBasicScenario-subscription-1", - topics = "pulsarReaderBasicScenario-topic-1", startMessageId = "earliest") + @PulsarReader(id = "pulsarReaderBasicScenario-id-1", topics = "pulsarReaderBasicScenario-topic-1", + startMessageId = "earliest") void read(String ignored) { latch.countDown(); } @@ -140,9 +139,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @EnablePulsar static class PulsarReaderStartMessageIdMissing { - @PulsarReader(id = "pulsarReaderBasicScenario-id-2", - subscriptionName = "pulsarReaderBasicScenario-subscription-2", - topics = "pulsarReaderBasicScenario-topic-2") + @PulsarReader(id = "pulsarReaderBasicScenario-id-2", topics = "pulsarReaderBasicScenario-topic-2") void readWithoutStartMessageId(String ignored) { } @@ -169,9 +166,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @Configuration static class PulsarReaderStartMessageIdLatest { - @PulsarReader(id = "pulsarReaderBasicScenario-id-3", - subscriptionName = "pulsarReaderBasicScenario-subscription-3", - topics = "pulsarReaderBasicScenario-topic-3", startMessageId = "latest") + @PulsarReader(id = "pulsarReaderBasicScenario-id-3", topics = "pulsarReaderBasicScenario-topic-3", + startMessageId = "latest") void read(String msg) { latch.countDown(); assertThat(msg).isEqualTo("hello foobar"); @@ -200,8 +196,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { MessageId[] messageIds = new MessageId[10]; - @PulsarReader(id = "with-customizer-reader", subscriptionName = "with-customizer-reader-subscription", - topics = "with-customizer-reader-topic", readerCustomizer = "myCustomizer") + @PulsarReader(id = "with-customizer-reader", topics = "with-customizer-reader-topic", + readerCustomizer = "myCustomizer") void listen(Message message) { assertThat(message.getMessageId()).isEqualTo(messageIds[currentIndex++]); latch.countDown();