do not throw exception for existing cap-msg-group message header. ove…

…rwrite it instead. tolerate kafka message duplicate headers (#1623)
dotnetcore · Dec 5, 2024 · 09f2a05 · 09f2a05
1 parent 40a56d2
commit 09f2a05
Show file tree

Hide file tree

Showing 6 changed files with 8 additions and 7 deletions.
diff --git a/src/DotNetCore.CAP.AmazonSQS/AmazonSQSConsumerClient.cs b/src/DotNetCore.CAP.AmazonSQS/AmazonSQSConsumerClient.cs
@@ -111,7 +111,7 @@ Task Consume()
 
                     var message = new TransportMessage(header, body != null ? Encoding.UTF8.GetBytes(body) : null);
 
-                    message.Headers.Add(Headers.Group, _groupId);
+                    message.Headers[Headers.Group] = _groupId;
 
                     return OnMessageCallback!(message, response.Messages[0].ReceiptHandle);
                 }

diff --git a/src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClient.cs b/src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClient.cs
@@ -278,7 +278,7 @@ private TransportMessage ConvertMessage(ServiceBusReceivedMessage message)
         var headers = message.ApplicationProperties
             .ToDictionary(x => x.Key, y => y.Value?.ToString());
 
-        headers.Add(Headers.Group, _subscriptionName);
+        headers[Headers.Group] = _subscriptionName;
 
         if (_asbOptions.CustomHeadersBuilder != null)
         {

diff --git a/src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs b/src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs
@@ -171,10 +171,10 @@ private async Task Consume(ConsumeResult<string, byte[]> consumerResult)
         foreach (var header in consumerResult.Message.Headers)
         {
             var val = header.GetValueBytes();
-            headers.Add(header.Key, val != null ? Encoding.UTF8.GetString(val) : null);
+            headers[header.Key] = val != null ? Encoding.UTF8.GetString(val) : null;
         }
 
-        headers.Add(Headers.Group, _groupId);
+        headers[Headers.Group] = _groupId;
 
         if (_kafkaOptions.CustomHeadersBuilder != null)
         {

diff --git a/src/DotNetCore.CAP.NATS/NATSConsumerClient.cs b/src/DotNetCore.CAP.NATS/NATSConsumerClient.cs
@@ -166,7 +166,7 @@ Task Consume()
                 headers.Add(h, e.Message.Header[h]);
             }
 
-            headers.Add(Headers.Group, _groupName);
+            headers[Headers.Group] = _groupName;
 
             if (_natsOptions.CustomHeadersBuilder != null)
             {

diff --git a/src/DotNetCore.CAP.Pulsar/PulsarConsumerClient.cs b/src/DotNetCore.CAP.Pulsar/PulsarConsumerClient.cs
@@ -78,7 +78,7 @@ Task Consume()
                         headers.Add(header.Key, header.Value);
                     }
 
-                    headers.Add(Headers.Group, _groupId);
+                    headers[Headers.Group] = _groupId;
 
                     var message = new TransportMessage(headers, consumerResult.Data);
 

diff --git a/src/DotNetCore.CAP.RabbitMQ/RabbitMQBasicConsumer.cs b/src/DotNetCore.CAP.RabbitMQ/RabbitMQBasicConsumer.cs
@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Text;
+using System.Text.RegularExpressions;
 using System.Threading;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Messages;
@@ -67,7 +68,7 @@ Task Consume()
                         headers.Add(header.Key, header.Value?.ToString());
                 }
 
-            headers.Add(Messages.Headers.Group, _groupName);
+            headers[Messages.Headers.Group] = _groupName;
 
             if (_customHeadersBuilder != null)
             {