feat: support producer access mode.

Author: shibd

src/config.cc

@@ -223,7 +223,9 @@ void export_config(py::module_& m) {
         .def("batching_type", &ProducerConfiguration::setBatchingType, return_value_policy::reference)
         .def("batching_type", &ProducerConfiguration::getBatchingType)
         .def("encryption_key", &ProducerConfiguration::addEncryptionKey, return_value_policy::reference)
-        .def("crypto_key_reader", &ProducerConfiguration::setCryptoKeyReader, return_value_policy::reference);
+        .def("crypto_key_reader", &ProducerConfiguration::setCryptoKeyReader, return_value_policy::reference)
+        .def("access_mode", &ProducerConfiguration::setAccessMode, return_value_policy::reference)
+        .def("access_mode", &ProducerConfiguration::getAccessMode, return_value_policy::copy);
 
     class_<BatchReceivePolicy>(m, "BatchReceivePolicy")
         .def(init<int, int, long>())

src/enums.cc

@@ -124,6 +124,12 @@ void export_enums(py::module_& m) {
         .value("Default", ProducerConfiguration::DefaultBatching)
         .value("KeyBased", ProducerConfiguration::KeyBasedBatching);
 
+    enum_<ProducerConfiguration::ProducerAccessMode>(m, "ProducerAccessMode", "Producer Access Mode")
+        .value("Shared", ProducerConfiguration::ProducerAccessMode::Shared)
+        .value("Exclusive", ProducerConfiguration::ProducerAccessMode::Exclusive)
+        .value("WaitForExclusive", ProducerConfiguration::ProducerAccessMode::WaitForExclusive)
+        .value("ExclusiveWithFencing", ProducerConfiguration::ProducerAccessMode::ExclusiveWithFencing);
+
     enum_<Logger::Level>(m, "LoggerLevel")
         .value("Debug", Logger::LEVEL_DEBUG)
         .value("Info", Logger::LEVEL_INFO)

tests/pulsar_test.py

@@ -42,6 +42,7 @@
     InitialPosition,
     CryptoKeyReader,
     ConsumerBatchReceivePolicy,
+    ProducerAccessMode,
 )
 from pulsar.schema import JsonSchema, Record, Integer
 
@@ -166,6 +167,61 @@ def test_producer_send(self):
         self.assertEqual(msg_id, msg.message_id())
         client.close()
 
+    def test_producer_access_mode_exclusive(self):
+        client = Client(self.serviceUrl)
+        topic_name = "test-access-mode-exclusive"
+        client.create_producer(topic_name, producer_name="p1", access_mode=pulsar.ProducerAccessMode.Exclusive)
+        with self.assertRaises(pulsar.ProducerFenced):
+            client.create_producer(topic_name, producer_name="p2", access_mode=pulsar.ProducerAccessMode.Exclusive)
+        client.close()
+
+    def test_producer_access_mode_wait_exclusive(self):
+        client = Client(self.serviceUrl)
+        topic_name = "test_producer_access_mode_wait_exclusive"
+        producer1 = client.create_producer(
+            topic=topic_name,
+            producer_name='p-1',
+            access_mode=pulsar.ProducerAccessMode.Exclusive
+        )
+        assert producer1.producer_name() == 'p-1'
+
+        # when p1 close, p2 success created.
+        producer1.close()
+        producer2 = client.create_producer(
+            topic=topic_name,
+            producer_name='p-2',
+            access_mode=pulsar.ProducerAccessMode.WaitForExclusive
+        )
+        assert producer2.producer_name() == 'p-2'
+
+        producer2.close()
+        client.close()
+
+    def test_producer_access_mode_exclusive_with_fencing(self):
+        client = Client(self.serviceUrl)
+        topic_name = 'test_producer_access_mode_exclusive_with_fencing'
+
+        producer1 = client.create_producer(
+            topic=topic_name,
+            producer_name='p-1',
+            access_mode=pulsar.ProducerAccessMode.Exclusive
+        )
+        assert producer1.producer_name() == 'p-1'
+
+        producer2 = client.create_producer(
+            topic=topic_name,
+            producer_name='p-2',
+            access_mode=pulsar.ProducerAccessMode.ExclusiveWithFencing
+        )
+        assert producer2.producer_name() == 'p-2'
+
+        # producer1 will be fenced.
+        with self.assertRaises(pulsar.ProducerFenced):
+            producer1.send('test-msg'.encode('utf-8'))
+
+        producer2.close()
+        client.close()
+
     def test_producer_is_connected(self):
         client = Client(self.serviceUrl)
         topic = "test_producer_is_connected"