feat: support pattern subscription non persistent topic.

Author: shibd

pulsar/__init__.py

@@ -48,7 +48,7 @@
 import _pulsar
 
 from _pulsar import Result, CompressionType, ConsumerType, InitialPosition, PartitionsRoutingMode, BatchingType, \
-    LoggerLevel, BatchReceivePolicy, KeySharedPolicy, KeySharedMode, ProducerAccessMode  # noqa: F401
+    LoggerLevel, BatchReceivePolicy, KeySharedPolicy, KeySharedMode, ProducerAccessMode, RegexSubscriptionMode  # noqa: F401
 
 from pulsar.__about__ import __version__
 
@@ -708,6 +708,7 @@ def subscribe(self, topic, subscription_name,
                   batch_receive_policy=None,
                   key_shared_policy=None,
                   batch_index_ack_enabled=False,
+                  regex_subscription_mode=RegexSubscriptionMode.PersistentOnly,
                   ):
         """
         Subscribe to the given topic and subscription combination.
@@ -819,9 +820,11 @@ def my_listener(consumer, message):
         _check_type_or_none(ConsumerBatchReceivePolicy, batch_receive_policy, 'batch_receive_policy')
         _check_type_or_none(ConsumerKeySharedPolicy, key_shared_policy, 'key_shared_policy')
         _check_type(bool, batch_index_ack_enabled, 'batch_index_ack_enabled')
+        _check_type(RegexSubscriptionMode, regex_subscription_mode, 'regex_subscription_mode')
 
         conf = _pulsar.ConsumerConfiguration()
         conf.consumer_type(consumer_type)
+        conf.regex_subscription_mode(regex_subscription_mode)
         conf.read_compacted(is_read_compacted)
         if message_listener:
             conf.message_listener(_listener_wrapper(message_listener, schema))

src/config.cc

@@ -267,6 +267,8 @@ void export_config(py::module_& m) {
         .def("property", &ConsumerConfiguration::setProperty, return_value_policy::reference)
         .def("subscription_initial_position", &ConsumerConfiguration::getSubscriptionInitialPosition)
         .def("subscription_initial_position", &ConsumerConfiguration::setSubscriptionInitialPosition)
+        .def("regex_subscription_mode", &ConsumerConfiguration::setRegexSubscriptionMode)
+        .def("regex_subscription_mode", &ConsumerConfiguration::getRegexSubscriptionMode, return_value_policy::reference)
         .def("crypto_key_reader", &ConsumerConfiguration::setCryptoKeyReader, return_value_policy::reference)
         .def("replicate_subscription_state_enabled",
              &ConsumerConfiguration::setReplicateSubscriptionStateEnabled)

src/enums.cc

@@ -120,6 +120,11 @@ void export_enums(py::module_& m) {
         .value("Latest", InitialPositionLatest)
         .value("Earliest", InitialPositionEarliest);
 
+    enum_<RegexSubscriptionMode>(m, "RegexSubscriptionMode", "Regex subscription mode")
+        .value("PersistentOnly", PersistentOnly)
+        .value("NonPersistentOnly", NonPersistentOnly)
+        .value("AllTopics", AllTopics);
+
     enum_<ProducerConfiguration::BatchingType>(m, "BatchingType", "Supported batching types")
         .value("Default", ProducerConfiguration::DefaultBatching)
         .value("KeyBased", ProducerConfiguration::KeyBasedBatching);

tests/pulsar_test.py

@@ -46,7 +46,7 @@
 )
 from pulsar.schema import JsonSchema, Record, Integer
 
-from _pulsar import ProducerConfiguration, ConsumerConfiguration
+from _pulsar import ProducerConfiguration, ConsumerConfiguration, RegexSubscriptionMode
 
 from schema_test import *
 
@@ -1718,6 +1718,39 @@ def test_batch_index_ack(self):
 
         client.close()
 
+    def test_regex_subscription(self):
+        import re
+
+        client = Client(self.serviceUrl)
+        topic1 = "persistent://public/default/test-regex-sub-1"
+        topic2 = "persistent://public/default/test-regex-sub-2"
+        topic3 = "non-persistent://public/default/test-regex-sub-3"
+        topic4 = "persistent://public/default/no-match-test-regex-sub-3"  # no match pattern rule topic.
+
+        producer1 = client.create_producer(topic1)
+        producer2 = client.create_producer(topic2)
+        producer3 = client.create_producer(topic3)
+        producer4 = client.create_producer(topic4)
+
+        consumer = client.subscribe(
+            re.compile('public/default/test-regex-sub-.*'), "regex-sub", consumer_type=ConsumerType.Shared,
+            regex_subscription_mode=RegexSubscriptionMode.AllTopics
+        )
+
+        num = 10
+        for i in range(num):
+            producer1.send(b"hello-1-%d" % i)
+            producer2.send(b"hello-2-%d" % i)
+            producer3.send(b"hello-3-%d" % i)
+            producer4.send(b"hello-4-%d" % i)
+
+        for i in range(3 * num):
+            msg = consumer.receive(TM)
+            consumer.acknowledge(msg)
+
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
+        client.close()
 
 if __name__ == "__main__":
     main()