Feature/producer properties (#222)

* Add option to setup producer properties

* Add changelog entry

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [3.3.0] - ?
+
+### Added
+
+- Producer properties can be added when creating a producer
+
 ## [3.2.1] - 2024-04-24
 
 ### Fixed

src/DotPulsar/Abstractions/IProducerBuilder.cs

@@ -65,6 +65,11 @@ public interface IProducerBuilder<TMessage>
     /// </summary>
     IProducerBuilder<TMessage> MaxPendingMessages(uint maxPendingMessages);
 
+    /// <summary>
+    /// Add/Set a property key/value on the producer. This is optional.
+    /// </summary>
+    IProducerBuilder<TMessage> ProducerProperty(string key, string value);
+
     /// <summary>
     /// Create the producer.
     /// </summary>

src/DotPulsar/Internal/Producer.cs

@@ -159,7 +159,8 @@ private SubProducer CreateSubProducer(string topic, int partition)
         var producerName = _options.ProducerName;
         var schema = _options.Schema;
         var producerAccessMode = (PulsarApi.ProducerAccessMode) _options.ProducerAccessMode;
-        var factory = new ProducerChannelFactory(correlationId, _processManager, _connectionPool, topic, producerName, producerAccessMode, schema.SchemaInfo, _compressorFactory);
+        var producerProperties = _options.ProducerProperties;
+        var factory = new ProducerChannelFactory(correlationId, _processManager, _connectionPool, topic, producerName, producerAccessMode, schema.SchemaInfo, _compressorFactory, producerProperties);
         var stateManager = CreateStateManager();
         var initialChannel = new NotReadyChannel<TMessage>();
         var executor = new Executor(correlationId, _processManager, _exceptionHandler);

src/DotPulsar/Internal/ProducerBuilder.cs

@@ -21,6 +21,7 @@ public sealed class ProducerBuilder<TMessage> : IProducerBuilder<TMessage>
 {
     private readonly IPulsarClient _pulsarClient;
     private readonly ISchema<TMessage> _schema;
+    private readonly Dictionary<string, string> _producerProperties;
     private string? _producerName;
     private ProducerAccessMode _producerAccessMode;
     private bool _attachTraceInfoToMessages;
@@ -40,6 +41,7 @@ public ProducerBuilder(IPulsarClient pulsarClient, ISchema<TMessage> schema)
         _initialSequenceId = ProducerOptions<TMessage>.DefaultInitialSequenceId;
         _maxPendingMessages = 500;
         _producerAccessMode = ProducerOptions<TMessage>.DefaultProducerAccessMode;
+        _producerProperties = [];
     }
 
     public IProducerBuilder<TMessage> AttachTraceInfoToMessages(bool attachTraceInfoToMessages)
@@ -96,6 +98,12 @@ public IProducerBuilder<TMessage> MaxPendingMessages(uint maxPendingMessages)
         return this;
     }
 
+    public IProducerBuilder<TMessage> ProducerProperty(string key, string value)
+    {
+        _producerProperties[key] = value;
+        return this;
+    }
+
     public IProducer<TMessage> Create()
     {
         if (string.IsNullOrEmpty(_topic))
@@ -112,7 +120,8 @@ public IProducer<TMessage> Create()
             InitialSequenceId = _initialSequenceId,
             ProducerName = _producerName,
             StateChangedHandler = _stateChangedHandler,
-            MaxPendingMessages = _maxPendingMessages
+            MaxPendingMessages = _maxPendingMessages,
+            ProducerProperties = _producerProperties
         };
 
         if (_messageRouter is not null)

src/DotPulsar/Internal/ProducerChannelFactory.cs

@@ -36,7 +36,8 @@ public ProducerChannelFactory(
         string? producerName,
         ProducerAccessMode producerAccessMode,
         SchemaInfo schemaInfo,
-        ICompressorFactory? compressorFactory)
+        ICompressorFactory? compressorFactory,
+        Dictionary<string,string>? properties)
     {
         _correlationId = correlationId;
         _eventRegister = eventRegister;
@@ -49,6 +50,9 @@ public ProducerChannelFactory(
             Topic = topic
         };
 
+        if (properties is not null)
+            _commandProducer.Metadatas.AddRange(properties.Select(x => new KeyValue { Key = x.Key, Value = x.Value }));
+
         _compressorFactory = compressorFactory;
         _schema = schemaInfo.PulsarSchema;
     }

src/DotPulsar/ProducerOptions.cs

@@ -48,6 +48,7 @@ public ProducerOptions(string topic, ISchema<TMessage> schema)
         Topic = topic;
         Schema = schema;
         MessageRouter = new RoundRobinPartitionRouter();
+        ProducerProperties = [];
     }
 
     /// <summary>
@@ -99,4 +100,9 @@ public ProducerOptions(string topic, ISchema<TMessage> schema)
     /// Set the max size of the queue holding the messages pending to receive an acknowledgment from the broker.
     /// </summary>
     public uint MaxPendingMessages { get; set; }
+
+    /// <summary>
+    /// Add/Set the producers's properties. This is optional.
+    /// </summary>
+    public Dictionary<string, string> ProducerProperties { get; set; }
 }