[feat] Support set producer access mode for C client.

include/pulsar/c/producer_configuration.h

@@ -78,6 +78,21 @@ typedef enum
     pulsar_ProducerSend
 } pulsar_producer_crypto_failure_action;
 
+typedef enum
+{
+    // By default multiple producers can publish on a topic.
+    pulsar_ProducerAccessModeShared = 0,
+    // Require exclusive access for producer.
+    // Fail immediately if there's already a producer connected.
+    pulsar_ProducerAccessModeExclusive = 1,
+    // Producer creation is pending until it can acquire exclusive access.
+    pulsar_ProducerAccessModeWaitForExclusive = 2,
+    // Acquire exclusive access for the producer.
+    // Any existing producer will be removed and invalidated immediately.
+    pulsar_ProducerAccessModeExclusiveWithFencing = 3
+
+} pulsar_producer_access_mode;
+
 typedef struct _pulsar_producer_configuration pulsar_producer_configuration_t;
 
 typedef struct _pulsar_crypto_key_reader pulsar_crypto_key_reader;
@@ -210,6 +225,12 @@ PULSAR_PUBLIC void pulsar_producer_configuration_set_chunking_enabled(pulsar_pro
 
 PULSAR_PUBLIC int pulsar_producer_configuration_is_chunking_enabled(pulsar_producer_configuration_t *conf);
 
+PULSAR_PUBLIC pulsar_producer_access_mode
+pulsar_producer_configuration_get_access_mode(pulsar_producer_configuration_t *conf);
+
+PULSAR_PUBLIC void pulsar_producer_configuration_set_access_mode(pulsar_producer_configuration_t *conf,
+                                                                 pulsar_producer_access_mode accessMode);
+
 // const CryptoKeyReaderPtr getCryptoKeyReader() const;
 // ProducerConfiguration &setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader);
 //

lib/c/c_ProducerConfiguration.cc

@@ -232,3 +232,13 @@ void pulsar_producer_configuration_set_chunking_enabled(pulsar_producer_configur
 int pulsar_producer_configuration_is_chunking_enabled(pulsar_producer_configuration_t *conf) {
     return conf->conf.isChunkingEnabled();
 }
+
+pulsar_producer_access_mode pulsar_producer_configuration_get_access_mode(
+    pulsar_producer_configuration_t *conf) {
+    return (pulsar_producer_access_mode)conf->conf.getAccessMode();
+}
+
+void pulsar_producer_configuration_set_access_mode(pulsar_producer_configuration_t *conf,
+                                                   pulsar_producer_access_mode accessMode) {
+    conf->conf.setAccessMode((pulsar::ProducerConfiguration::ProducerAccessMode)accessMode);
+}

tests/c/c_ProducerConfigurationTest.cc

@@ -25,4 +25,8 @@ TEST(C_ProducerConfigurationTest, testCApiConfig) {
     ASSERT_EQ(pulsar_producer_configuration_is_chunking_enabled(producer_conf), 0);
     pulsar_producer_configuration_set_chunking_enabled(producer_conf, 1);
     ASSERT_EQ(pulsar_producer_configuration_is_chunking_enabled(producer_conf), 1);
+
+    pulsar_producer_configuration_set_access_mode(producer_conf, pulsar_ProducerAccessModeExclusive);
+    ASSERT_EQ(pulsar_producer_configuration_get_access_mode(producer_conf),
+              pulsar_ProducerAccessModeExclusive);
 }