Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement topic pattern on PulsarListener (#5) (#77)

Browse Source 
* Implement topicPattern parameter in PulsarListener

* Fix Typos
This commit is contained in:
Alexander Preuß
2022-08-31 21:01:32 +02:00
committed by GitHub
parent ada6086cd8
commit e364dee85a
9 changed files with 83 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -109,6 +109,7 @@ import org.springframework.validation.Validator;
* @param <V> the value type.
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
* @see PulsarListener
* @see EnablePulsar
* @see PulsarListenerConfigurer
@@ -277,14 +278,15 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
String beanRef = pulsarListener.beanRef();
this.listenerScope.addListener(beanRef, bean);
String[] topics = resolveTopics(pulsarListener);
processListener(endpoint, pulsarListener, bean, beanName, topics);
String topicPattern = getTopicPattern(pulsarListener);
processListener(endpoint, pulsarListener, bean, beanName, topics, topicPattern);
this.listenerScope.removeListener(beanRef);
}
protected void processListener(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener PulsarListener, Object bean,
String beanName, String[] topics) {
String beanName, String[] topics, String topicPattern) {
processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics);
processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics, topicPattern);
String containerFactory = resolve(PulsarListener.containerFactory());
PulsarListenerContainerFactory<?> listenerContainerFactory = resolveContainerFactory(PulsarListener,
@@ -335,13 +337,14 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
}
private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint<?> endpoint,
PulsarListener pulsarListener, Object bean, String[] topics) {
PulsarListener pulsarListener, Object bean, String[] topics, String topicPattern) {
endpoint.setBean(bean);
endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
endpoint.setSubscriptionName(getEndpointSubscriptionName(pulsarListener));
endpoint.setId(getEndpointId(pulsarListener));
endpoint.setTopics(topics);
endpoint.setTopicPattern(topicPattern);
endpoint.setSubscriptionType(getEndpointSubscriptionType(pulsarListener));
endpoint.setSchemaType(pulsarListener.schemaType());
@@ -464,6 +467,10 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
}
}
private String getTopicPattern(PulsarListener pulsarListener) {
return resolveExpressionAsString(pulsarListener.topicPattern(), "topicPattern");
}
private String resolveExpressionAsString(String value, String attribute) {
Object resolved = resolveExpression(value);
if (resolved instanceof String) {

18
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java
View File

@@ -41,12 +41,14 @@ import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter;
import org.springframework.pulsar.support.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Base implementation for {@link PulsarListenerEndpoint}.
*
* @param <K> Message payload type.
* @author Soby Chacko
* @author Alexander Preuß
*/
public abstract class AbstractPulsarListenerEndpoint<K>
implements PulsarListenerEndpoint, BeanFactoryAware, InitializingBean {
@@ -63,6 +65,8 @@ public abstract class AbstractPulsarListenerEndpoint<K>
private final Collection<String> topics = new ArrayList<>();
private String topicPattern;
private BeanFactory beanFactory;
private BeanExpressionResolver resolver;
@@ -97,8 +101,8 @@ public abstract class AbstractPulsarListenerEndpoint<K>
@Override
public void afterPropertiesSet() {
boolean topicsEmpty = getTopics().isEmpty();
if (!topicsEmpty) {
throw new IllegalStateException("Topics or topicPartitions must be provided but not both for " + this);
if (!topicsEmpty && !StringUtils.hasText(getTopicPattern())) {
throw new IllegalStateException("Topics or topicPattern must be provided but not both for " + this);
}
}
@@ -148,6 +152,16 @@ public abstract class AbstractPulsarListenerEndpoint<K>
return Collections.unmodifiableCollection(this.topics);
}
public void setTopicPattern(String topicPattern) {
Assert.notNull(topicPattern, "'topicPattern' must not be null");
this.topicPattern = topicPattern;
}
@Override
public String getTopicPattern() {
return this.topicPattern;
}
@Override
@Nullable
public Boolean getAutoStartup() {

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java
View File

@@ -31,6 +31,7 @@ import org.springframework.util.StringUtils;
* @param <T> message type in the listener.
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
*/
public class ConcurrentPulsarListenerContainerFactory<T>
extends AbstractPulsarListenerContainerFactory<ConcurrentPulsarMessageListenerContainer<T>, T> {
@@ -50,12 +51,17 @@ public class ConcurrentPulsarListenerContainerFactory<T>
PulsarContainerProperties properties = new PulsarContainerProperties();
Collection<String> topics = endpoint.getTopics();
String topicPattern = endpoint.getTopicPattern();
if (!topics.isEmpty()) {
final String[] topics1 = topics.toArray(new String[0]);
properties.setTopics(topics1);
}
if (StringUtils.hasText(topicPattern)) {
properties.setTopicsPattern(topicPattern);
}
final String subscriptionName = endpoint.getSubscriptionName();
if (StringUtils.hasText(subscriptionName)) {

3
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java
View File

@@ -32,6 +32,7 @@ import org.springframework.pulsar.support.MessageConverter;
* endpoints programmatically.
*
* @author Soby Chacko
* @author Alexander Preuß
*/
public interface PulsarListenerEndpoint {
@@ -46,6 +47,8 @@ public interface PulsarListenerEndpoint {
Collection<String> getTopics();
String getTopicPattern();
@Nullable
Boolean getAutoStartup();

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java
View File

@@ -31,6 +31,7 @@ import org.springframework.pulsar.support.MessageConverter;
* Adapter to avoid having to implement all methods.
*
* @author Soby Chacko
* @author Alexander Preuß
*/
public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint {
@@ -54,6 +55,11 @@ public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint {
return Collections.emptyList();
}
@Override
public String getTopicPattern() {
return null;
}
@Override
public Boolean getAutoStartup() {
return null;

15
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -56,6 +56,7 @@ import org.springframework.util.StringUtils;
*
* @param <T> message type.
* @author Soby Chacko
* @author Alexander Preuß
*/
public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMessageListenerContainer<T> {
@@ -223,6 +224,14 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
propertiesToOverride.put("topicNames", listenerDefinedTopics);
}
}
if (!propertiesToOverride.containsKey("topicsPattern")) {
final String topicsPattern = pulsarContainerProperties.getTopicsPattern();
if (topicsPattern != null) {
propertiesToOverride.put("topicsPattern", topicsPattern);
}
}
if (!propertiesToOverride.containsKey("subscriptionName")) {
if (StringUtils.hasText(pulsarContainerProperties.getSubscriptionName())) {
propertiesToOverride.put("subscriptionName", pulsarContainerProperties.getSubscriptionName());
@@ -267,7 +276,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
try {
if (isSharedSubsriptionType()) {
if (isSharedSubscriptionType()) {
this.consumer.acknowledge(messages);
}
else {
@@ -322,7 +331,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
}
private boolean isSharedSubsriptionType() {
private boolean isSharedSubscriptionType() {
return this.containerProperties.getSubscriptionType() == SubscriptionType.Shared
|| this.containerProperties.getSubscriptionType() == SubscriptionType.Key_Shared;
}
@@ -331,7 +340,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
if (this.nackableMessages.isEmpty()) {
try {
if (messages.size() > 0) {
if (isSharedSubsriptionType()) {
if (isSharedSubscriptionType()) {
this.consumer.acknowledge(messages);
}
else {

12
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java
View File

@@ -18,7 +18,6 @@ package org.springframework.pulsar.listener;
import java.time.Duration;
import java.util.Properties;
import java.util.regex.Pattern;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -31,6 +30,7 @@ import org.springframework.util.Assert;
* Contains runtime properties for a listener container.
*
* @author Soby Chacko
* @author Alexander Preuß
*/
public class PulsarContainerProperties {
@@ -60,7 +60,7 @@ public class PulsarContainerProperties {
private String[] topics;
private Pattern topicsPattern;
private String topicsPattern;
private String subscriptionName;
@@ -91,7 +91,7 @@ public class PulsarContainerProperties {
this.topicsPattern = null;
}
public PulsarContainerProperties(Pattern topicPattern) {
public PulsarContainerProperties(String topicPattern) {
this.topicsPattern = topicPattern;
this.topics = null;
}
@@ -170,7 +170,7 @@ public class PulsarContainerProperties {
* @param consumerStartTimeout the consumer start timeout.
*/
public void setConsumerStartTimeout(Duration consumerStartTimeout) {
Assert.notNull(consumerStartTimeout, "'consumerStartTimout' cannot be null");
Assert.notNull(consumerStartTimeout, "'consumerStartTimeout' cannot be null");
this.consumerStartTimeout = consumerStartTimeout;
}
@@ -190,11 +190,11 @@ public class PulsarContainerProperties {
this.topics = topics;
}
public Pattern getTopicsPattern() {
public String getTopicsPattern() {
return this.topicsPattern;
}
public void setTopicsPattern(Pattern topicsPattern) {
public void setTopicsPattern(String topicsPattern) {
this.topicsPattern = topicsPattern;
}

2
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainierTests.java → spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java
View File

@@ -39,7 +39,7 @@ import org.springframework.pulsar.core.PulsarTemplate;
/**
* @author Soby Chacko
*/
class DefaultPulsarMessageListenerContainierTests extends AbstractContainerBaseTests {
class DefaultPulsarMessageListenerContainerTests extends AbstractContainerBaseTests {
@Test
void basicDefaultConsumer() throws Exception {

22
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -55,6 +55,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Soby Chacko
* @author Alexander Preuß
*/
@SpringJUnitConfig
@DirtiesContext
@@ -62,6 +63,7 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
static CountDownLatch latch = new CountDownLatch(1);
static CountDownLatch latch1 = new CountDownLatch(3);
static CountDownLatch latch2 = new CountDownLatch(3);
@Autowired
PulsarTemplate<String> pulsarTemplate;
@@ -125,6 +127,19 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
@ContextConfiguration(classes = PulsarListenerBasicTestCases.TestPulsarListenersForBasicScenario.class)
class PulsarListenerBasicTestCases {
@Test
void testPulsarListenerWithTopicsPattern(@Autowired PulsarListenerEndpointRegistry registry) throws Exception {
PulsarMessageListenerContainer baz = registry.getListenerContainer("baz");
PulsarContainerProperties containerProperties = baz.getContainerProperties();
assertThat(containerProperties.getTopicsPattern()).isEqualTo("persistent://public/default/pattern.*");
pulsarTemplate.send("persistent://public/default/pattern-1", "hello baz");
pulsarTemplate.send("persistent://public/default/pattern-2", "hello baz");
pulsarTemplate.send("persistent://public/default/pattern-3", "hello baz");
assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void testPulsarListenerProvidedConsumerProperties(@Autowired PulsarListenerEndpointRegistry registry)
throws Exception {
@@ -192,6 +207,13 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
latch1.countDown();
}
@PulsarListener(id = "baz", topicPattern = "persistent://public/default/pattern.*",
subscriptionName = "subscription-3",
properties = { "patternAutoDiscoveryPeriod=5", "subscriptionInitialPosition=Earliest" })
void listen3(String message) {
latch2.countDown();
}
}
}