Kafka Headers

src/Paramore.Brighter.DynamoDb/GUIDConverter.cs

@@ -8,10 +8,8 @@ public class GUIDConverter : IPropertyConverter
     {
         public DynamoDBEntry ToEntry(object value)
         {
-            var uuid = value as Guid? ??
-                       throw new InvalidOperationException(
-                           $"Supplied type was of type {value.GetType().Name} not Guid");
 
+            var uuid = (Guid)value;
             var json = uuid.ToString();
 
             DynamoDBEntry entry = new Primitive

src/Paramore.Brighter.MessagingGateway.Kafka/BrighterDefinedHeaders.cs

@@ -0,0 +1,47 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+namespace Paramore.Brighter.MessagingGateway.Kafka;
+
+public static class BrighterDefinedHeaders
+{
+    public static readonly string[] HeadersToReset;
+
+    static BrighterDefinedHeaders()
+    {
+        HeadersToReset = new[] {
+            HeaderNames.MESSAGE_ID,
+            HeaderNames.MESSAGE_TYPE,
+            HeaderNames.TOPIC,
+            HeaderNames.CORRELATION_ID,
+            HeaderNames.TIMESTAMP,
+            HeaderNames.PARTITIONKEY,
+            HeaderNames.CONTENT_TYPE,
+            HeaderNames.REPLY_TO,
+            HeaderNames.DELAYED_MILLISECONDS,
+            HeaderNames.HANDLED_COUNT
+        };  
+    }
+
+}

src/Paramore.Brighter.MessagingGateway.Kafka/HeaderNames.cs

@@ -27,6 +27,11 @@ namespace Paramore.Brighter.MessagingGateway.Kafka
 {
     public class HeaderNames
     {
+        /// <summary>
+        /// A dictionary of user defined values
+        /// </summary>
+        public static string BAG { get; } = "Bag";
+
         /// <summary>
         /// What is the content type of the message§
         /// </summary>
@@ -37,6 +42,17 @@ public class HeaderNames
         /// </summary>
         public const string CORRELATION_ID = "CorrelationId";
 
+        /// <summary>
+        /// If the message was deferred, how long for?
+        /// </summary>
+        public static string DELAYED_MILLISECONDS { get; } =  "x-delay";
+
+        /// <summary>
+        /// How many times has the message been retried with a delay
+        /// </summary>
+        public static string HANDLED_COUNT { get; } = "HandledCount" ;
+
+
         /// <summary>
         /// The message type
         /// </summary>

src/Paramore.Brighter.MessagingGateway.Kafka/IKafkaMessageHeaderBuilder.cs

@@ -1,4 +1,28 @@
-using Confluent.Kafka;
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using Confluent.Kafka;
 
 namespace Paramore.Brighter.MessagingGateway.Kafka
 {

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaDefaultMessageHeaderBuilder.cs

@@ -1,24 +1,41 @@
-using System;
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using System;
+using System.Globalization;
 using System.Linq;
 using Confluent.Kafka;
 using Paramore.Brighter.Extensions;
 
 namespace Paramore.Brighter.MessagingGateway.Kafka
 {
-    internal class KafkaDefaultMessageHeaderBuilder : IKafkaMessageHeaderBuilder
+    /// <summary>
+    /// This class serializes Brighter headers to Kafka. Kafka uses a byte[] for its header values. We convert all
+    /// header values into a string, and then get a UTF8 encoded set of bytes for that string.
+    /// </summary>
+    public class KafkaDefaultMessageHeaderBuilder : IKafkaMessageHeaderBuilder
     {
-        private static readonly string[] s_headersToReset;
-
-        static KafkaDefaultMessageHeaderBuilder()
-        {
-            s_headersToReset = new[] {
-                HeaderNames.MESSAGE_TYPE,
-                HeaderNames.TOPIC,
-                HeaderNames.CORRELATION_ID,
-                HeaderNames.TIMESTAMP
-            };
-        }
-
         public static KafkaDefaultMessageHeaderBuilder Instance => new KafkaDefaultMessageHeaderBuilder();
 
         public Headers Build(Message message)
@@ -31,9 +48,9 @@ public Headers Build(Message message)
             };
 
             if (message.Header.TimeStamp != default)
-                headers.Add(HeaderNames.TIMESTAMP, BitConverter.GetBytes(new DateTimeOffset(message.Header.TimeStamp).ToUnixTimeMilliseconds()));
+                headers.Add(HeaderNames.TIMESTAMP, new DateTimeOffset(message.Header.TimeStamp).ToString().ToByteArray());
             else
-                headers.Add(HeaderNames.TIMESTAMP, BitConverter.GetBytes(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()));
+                headers.Add(HeaderNames.TIMESTAMP, DateTimeOffset.UtcNow.ToString().ToByteArray());
 
             if (message.Header.CorrelationId != Guid.Empty)
                 headers.Add(HeaderNames.CORRELATION_ID, message.Header.CorrelationId.ToString().ToByteArray());
@@ -47,20 +64,39 @@ public Headers Build(Message message)
             if (!string.IsNullOrEmpty(message.Header.ReplyTo))
                 headers.Add(HeaderNames.REPLY_TO, message.Header.ReplyTo.ToByteArray());
 
+            headers.Add(HeaderNames.DELAYED_MILLISECONDS, message.Header.DelayedMilliseconds.ToString().ToByteArray());
+
+            headers.Add(HeaderNames.HANDLED_COUNT, message.Header.HandledCount.ToString().ToByteArray());
+
             message.Header.Bag.Each((header) =>
             {
-                if (!s_headersToReset.Any(htr => htr.Equals(header.Key)))
+                if (!BrighterDefinedHeaders.HeadersToReset.Any(htr => htr.Equals(header.Key)))
                 {
                     switch (header.Value)
                     {
                         case string stringValue:
                             headers.Add(header.Key, stringValue.ToByteArray());
                             break;
+                        case DateTime dateTimeValue:
+                            headers.Add(header.Key, dateTimeValue.ToString(CultureInfo.InvariantCulture).ToByteArray());
+                            break;
+                       case Guid guidValue:
+                            headers.Add(header.Key, guidValue.ToString().ToByteArray());
+                            break;
+                       case bool boolValue:
+                            headers.Add(header.Key, boolValue.ToString().ToByteArray());
+                            break;
                         case int intValue:
-                            headers.Add(header.Key, BitConverter.GetBytes(intValue));
+                            headers.Add(header.Key, intValue.ToString().ToByteArray());
+                            break; 
+                        case double doubleValue:
+                            headers.Add(header.Key, doubleValue.ToString(CultureInfo.InvariantCulture).ToByteArray());
+                            break;
+                        case float floatValue:
+                            headers.Add(header.Key, floatValue.ToString(CultureInfo.InvariantCulture).ToByteArray());
                             break;
-                        case Guid guidValue:
-                            headers.Add(header.Key, guidValue.ToByteArray());
+                        case long longValue:
+                            headers.Add(header.Key, longValue.ToString().ToByteArray());
                             break;
                         case byte[] byteArray:
                             headers.Add(header.Key, byteArray);
@@ -71,7 +107,7 @@ public Headers Build(Message message)
                     }
                 }
             });
-
+            
             return headers;
         }
     }

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaHeadersTools.cs

@@ -1,7 +1,29 @@
-using System;
-using System.Collections.Generic;
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using System;
 using System.Linq;
-using System.Text;
 using Confluent.Kafka;
 
 namespace Paramore.Brighter.MessagingGateway.Kafka

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageConsumer.cs

@@ -20,6 +20,7 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE. */
 #endregion
+
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;