Add producer name attribute to the Message (#331)

* Add producer name attribute to the Message

* Fix compilation issue

Author: RobertIndie

src/Pulsar.Client/Common/DTO.fs

@@ -213,6 +213,7 @@ type internal Metadata =
         EncryptionAlgo: string
         OrderingKey: byte[]
         ReplicatedFrom: string
+        ProducerName: string
         NullValue: bool
     }
 
@@ -265,7 +266,7 @@ type Message<'T> internal (messageId: MessageId, data: byte[], key: PartitionKey
                   properties: IReadOnlyDictionary<string, string>, encryptionCtx: EncryptionContext option,
                   schemaVersion: byte[], sequenceId: SequenceId, orderingKey: byte[], publishTime: TimeStamp,
                   eventTime: Nullable<TimeStamp>,
-                  redeliveryCount: int32, replicatedFrom: string,
+                  redeliveryCount: int32, replicatedFrom: string, producerName: string,
                   getValue: unit -> 'T) =
     /// Get the unique message ID associated with this message.
     member this.MessageId = messageId
@@ -294,26 +295,28 @@ type Message<'T> internal (messageId: MessageId, data: byte[], key: PartitionKey
     member this.RedeliveryCount = redeliveryCount
     /// Get name of cluster, from which the message is replicated.
     member this.ReplicatedFrom = replicatedFrom
+    /// Get name of producer of the message
+    member this.ProducerName = producerName
 
     /// Get the de-serialized value of the message, according the configured Schema.
     member this.GetValue() =
         getValue()
 
     member internal this.WithMessageId messageId =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated data
     member this.WithData data =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated key
     member this.WithKey (key, hasBase64EncodedKey) =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated properties
     member this.WithProperties properties =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
 
 type Messages<'T> internal(maxNumberOfMessages: int, maxSizeOfMessages: int64) =
 

src/Pulsar.Client/Internal/ClientCnx.fs

@@ -530,6 +530,7 @@ and internal ClientCnx (config: PulsarClientConfiguration,
             EncryptionAlgo = messageMetadata.EncryptionAlgo
             OrderingKey = messageMetadata.OrderingKey
             ReplicatedFrom = messageMetadata.ReplicatedFrom
+            ProducerName = messageMetadata.ProducerName
             NullValue = messageMetadata.NullValue
         }
 

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -691,6 +691,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                             rawMessage.Metadata.EventTime,
                             rawMessage.RedeliveryCount,
                             rawMessage.Metadata.ReplicatedFrom,
+                            rawMessage.Metadata.ProducerName,
                             getValue
                         )
             if (rawMessage.RedeliveryCount >= deadLettersProcessor.MaxRedeliveryCount) then
@@ -1387,6 +1388,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                     eventTime,
                     rawMessage.RedeliveryCount,
                     rawMessage.Metadata.ReplicatedFrom,
+                    rawMessage.Metadata.ProducerName,
                     getValue
                 )
                 if (rawMessage.RedeliveryCount >= deadLettersProcessor.MaxRedeliveryCount) then

tests/IntegrationTests/Basic.fs

@@ -24,7 +24,7 @@ let tests =
 
     testList "Basic" [
 
-        testTask "Sent messageId should be equal to received messageId" {
+        testTask "Sent message/messageId should be equal to received message/messageId" {
 
             Log.Debug("Started Sent messageId should be equal to received messageId")
             let client = getClient()
@@ -33,11 +33,13 @@ let tests =
             let! (producer1 : IProducer<byte[]>) =
                 client.NewProducer()
                     .Topic(topicName)
+                    .ProducerName("producer1")
                     .CreateAsync()
 
             let! (producer2 : IProducer<byte[]>) =
                 client.NewProducer()
                     .Topic(topicName)
+                    .ProducerName("producer2")
                     .EnableBatching(false)
                     .CreateAsync()
 
@@ -55,10 +57,12 @@ let tests =
             Expect.isTrue "" (msg1Id = msg1.MessageId)
             Expect.equal "" [| 0uy |] <| msg1.GetValue()
             Expect.equal "Message ID topic name should match" topicName (string msg1.MessageId.TopicName)
+            Expect.equal "Message producer name should match" "producer1" (string msg1.ProducerName)
 
             Expect.isTrue "" (msg2Id = msg2.MessageId)
             Expect.equal "" [| 1uy |] <| msg2.GetValue()
             Expect.equal "Message ID topic name should match" topicName (string msg2.MessageId.TopicName)
+            Expect.equal "Message producer name should match" "producer2" (string msg2.ProducerName)
 
             Log.Debug("Finished Sent messageId should be equal to received messageId")
         }

tests/UnitTests/Internal/ChunkedMessageTrackerTests.fs

@@ -31,6 +31,7 @@ let tests =
             EventTime = Nullable()
             OrderingKey = [||]
             ReplicatedFrom = ""
+            ProducerName = ""
             NullValue = false
         }