feat: support relative topicPattern without requiring full tenant/namespace prefix #1240

Signed-off-by: oneby-wang <[email protected]>

Author: oneby-wang

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java

@@ -18,7 +18,9 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.regex.Pattern;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
@@ -98,6 +100,12 @@ protected void ensureTopicNamesFullyQualified(ReactiveMessageConsumerBuilder<T>
 			var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
 			mutableSpec.setTopicNames(fullyQualifiedTopics);
 		}
+		var topicsPattern = mutableSpec.getTopicsPattern();
+		if (topicsPattern != null && StringUtils.isNoneBlank(topicsPattern.pattern())) {
+			var topicsPatternStr = topicsPattern.pattern();
+			var fullyQualifiedTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
+			mutableSpec.setTopicsPattern(Pattern.compile(fullyQualifiedTopicsPatternStr));
+		}
 	}
 
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java

@@ -23,7 +23,9 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.TreeMap;
+import java.util.regex.Pattern;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -146,6 +148,12 @@ protected void ensureTopicNamesFullyQualified(ConsumerBuilder<T> builder) {
 			var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
 			builderImpl.getConf().setTopicNames(new HashSet<>(fullyQualifiedTopics));
 		}
+		var topicsPattern = builderImpl.getConf().getTopicsPattern();
+		if (topicsPattern != null && StringUtils.isNoneBlank(topicsPattern.pattern())) {
+			var topicsPatternStr = topicsPattern.pattern();
+			var fullyQualifiedTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
+			builderImpl.getConf().setTopicsPattern(Pattern.compile(fullyQualifiedTopicsPatternStr));
+		}
 	}
 
 }