Binary Payloads

samples/KafkaSchemaRegistry/GreetingsSender/TimedMessageGenerator.cs

@@ -25,7 +25,7 @@ public Task StartAsync(CancellationToken cancellationToken)
         {
             _logger.LogInformation("Kafka Message Generator is starting.");
 
-            //_timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromMilliseconds(500));
+            _timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromMilliseconds(500));
 
             DoWork(null);
 
@@ -36,7 +36,7 @@ public Task StopAsync(CancellationToken cancellationToken)
         {
             _logger.LogInformation("Kafka Message Generator is stopping.");
 
-            //_timer?.Change(Timeout.Infinite, 0);
+            _timer?.Change(Timeout.Infinite, 0);
 
             return Task.CompletedTask;
         }
@@ -56,7 +56,7 @@ private void DoWork(object state)
 
         public void Dispose()
         {
-            //_timer?.Dispose();
+            _timer?.Dispose();
         }
     }
 }

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

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2020 Ian Cooper <[email protected]>
 
@@ -27,32 +28,43 @@ namespace Paramore.Brighter.MessagingGateway.Kafka
     public class HeaderNames
     {
         /// <summary>
-        /// The message type
-        /// </summary>
-        public const string MESSAGE_TYPE = "MessageType";
-        /// <summary>
-        /// The message identifier
+        /// What is the content type of the message§
         /// </summary>
-        public const string MESSAGE_ID = "MessageId";
+        public static string CONTENT_TYPE { get; set; } = "ContentType";
+
         /// <summary>
         /// The correlation id
         /// </summary>
         public const string CORRELATION_ID = "CorrelationId";
+
         /// <summary>
-        /// The time that message was sent
+        /// The message type
         /// </summary>
-        public const string TIMESTAMP = "TimeStamp";
+        public const string MESSAGE_TYPE = "MessageType";
+
         /// <summary>
-        /// The topic
+        /// The message identifier
         /// </summary>
-        public const string TOPIC = "Topic";
+        public const string MESSAGE_ID = "MessageId";
+
         /// <summary>
         /// The key used to partition this message
         /// </summary>
         public static string PARTITIONKEY = "PartitionKey";
+
         /// <summary>
         /// What is the offset into the partition of the message
         /// </summary>
         public static string PARTITION_OFFSET = "TopicPartitionOffset";
+
+        /// <summary>
+        /// The time that message was sent
+        /// </summary>
+        public const string TIMESTAMP = "TimeStamp";
+
+        /// <summary>
+        /// The topic
+        /// </summary>
+        public const string TOPIC = "Topic";
     }
 }

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

@@ -46,7 +46,7 @@ namespace Paramore.Brighter.MessagingGateway.Kafka
     /// </summary>
     public class KafkaMessageConsumer : KafkaMessagingGateway, IAmAMessageConsumer
     {
-        private IConsumer<string, string> _consumer;
+        private IConsumer<string, byte[]> _consumer;
         private readonly KafkaMessageCreator _creator;
         private readonly ConsumerConfig _consumerConfig;
         private List<TopicPartition> _partitions = new List<TopicPartition>();
@@ -154,7 +154,7 @@ public KafkaMessageConsumer(
             _sweepUncommittedInterval = TimeSpan.FromMilliseconds(sweepUncommittedOffsetsIntervalMs);
             _readCommittedOffsetsTimeoutMs = readCommittedOffsetsTimeoutMs;
 
-            _consumer = new ConsumerBuilder<string, string>(_consumerConfig)
+            _consumer = new ConsumerBuilder<string, byte[]>(_consumerConfig)
                 .SetPartitionsAssignedHandler((consumer, list) =>
                 {
                     var partitions = list.Select(p => $"{p.Topic} : {p.Partition.Value}");

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

@@ -12,7 +12,7 @@ internal class KafkaMessageCreator
     {
         private static readonly ILogger s_logger = ApplicationLogging.CreateLogger<KafkaMessageCreator>();
 
-        public Message CreateMessage(ConsumeResult<string, string> consumeResult)
+        public Message CreateMessage(ConsumeResult<string, byte[]> consumeResult)
         {
             var headers = consumeResult.Message.Headers;
             var topic = HeaderResult<string>.Empty();
@@ -21,6 +21,8 @@ public Message CreateMessage(ConsumeResult<string, string> consumeResult)
             var messageType = HeaderResult<MessageType>.Empty();
             var correlationId = HeaderResult<Guid>.Empty();
             var partitionKey = HeaderResult<string>.Empty();
+            var contentType = HeaderResult<string>.Empty(); 
+
 
             Message message;
             try
@@ -31,6 +33,7 @@ public Message CreateMessage(ConsumeResult<string, string> consumeResult)
                 messageType = ReadMessageType(consumeResult.Message.Headers);
                 correlationId = ReadCorrelationId(consumeResult.Message.Headers);
                 partitionKey = ReadPartitionKey(consumeResult.Message.Headers);
+                contentType = ReadContentType(consumeResult.Message.Headers); 
 
 
                 if (false == (topic.Success && messageId.Success && messageType.Success && timeStamp.Success))
@@ -48,8 +51,11 @@ public Message CreateMessage(ConsumeResult<string, string> consumeResult)
 
                     if (partitionKey.Success)
                         messageHeader.PartitionKey = partitionKey.Result;
+
+                    if (contentType.Success)
+                        messageHeader.ContentType =contentType.Result;
 
-                    message = new Message(messageHeader, new MessageBody(consumeResult.Message.Value));
+                    message = new Message(messageHeader, new MessageBody(consumeResult.Message.Value, messageHeader.ContentType));
 
                     headers.Each(header => message.Header.Bag.Add(header.Key, ParseHeaderValue(header)));
 
@@ -66,7 +72,6 @@ public Message CreateMessage(ConsumeResult<string, string> consumeResult)
             return message;
         }
 
-
         private Message FailureMessage(HeaderResult<string> topic, HeaderResult<Guid> messageId)
         {
             var header = new MessageHeader(
@@ -76,6 +81,12 @@ private Message FailureMessage(HeaderResult<string> topic, HeaderResult<Guid> me
             var message = new Message(header, new MessageBody(string.Empty));
             return message;
         }
+
+
+        private HeaderResult<string> ReadContentType(Headers headers)
+        {
+           return ReadHeader(headers, HeaderNames.CONTENT_TYPE,false);
+        }
 
         private HeaderResult<Guid> ReadCorrelationId(Headers headers)
         {
@@ -156,8 +167,8 @@ private HeaderResult<string> ReadPartitionKey(Headers headers)
                 {
                     if (string.IsNullOrEmpty(s))
                     {
-                         s_logger.LogDebug("No partition key found in message");
-                         return new HeaderResult<string>(string.Empty, true);
+                        s_logger.LogDebug("No partition key found in message");
+                        return new HeaderResult<string>(string.Empty, true);
                     }
 
                     return new HeaderResult<string>(s, true);

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

@@ -55,7 +55,7 @@ internal class KafkaMessageProducer : KafkaMessagingGateway, IAmAMessageProducer
         /// </summary>
         public Dictionary<string, object> OutBoxBag { get; set; } = new Dictionary<string, object>();
 
-        private IProducer<string, string> _producer;
+        private IProducer<string, byte[]> _producer;
         private readonly ProducerConfig _producerConfig;
         private KafkaMessagePublisher _publisher;
         private bool _hasFatalProducerError = false;
@@ -127,7 +127,7 @@ public void ConfigHook(Action<ProducerConfig> configHook)
         /// </summary>
         public void Init()
         {
-            _producer = new ProducerBuilder<string, string>(_producerConfig)
+            _producer = new ProducerBuilder<string, byte[]>(_producerConfig)
                 .SetErrorHandler((_, error) =>
                 {
                     s_logger.LogError("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason,

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

@@ -8,37 +8,41 @@ namespace Paramore.Brighter.MessagingGateway.Kafka
 {
     internal class KafkaMessagePublisher
     {
-        private readonly IProducer<string, string> _producer;
-        private static readonly string[] _headersToReset =
+        private readonly IProducer<string, byte[]> _producer;
+
+        private static readonly string[] s_headersToReset;
+
+        public KafkaMessagePublisher(IProducer<string, byte[]> producer)
         {
-            HeaderNames.MESSAGE_TYPE,
-            HeaderNames.TOPIC,
-            HeaderNames.CORRELATION_ID,
-            HeaderNames.TIMESTAMP
-        };
-
+            _producer = producer;
+        }
 
-        public KafkaMessagePublisher(IProducer<string, string> producer)
+        static KafkaMessagePublisher()
         {
-            _producer = producer;
+            s_headersToReset = new[] {
+                HeaderNames.MESSAGE_TYPE,
+                HeaderNames.TOPIC,
+                HeaderNames.CORRELATION_ID,
+                HeaderNames.TIMESTAMP
+            };
         }
 
-        public void PublishMessage(Message message, Action<DeliveryReport<string, string>> deliveryReport)
+        public void PublishMessage(Message message, Action<DeliveryReport<string, byte[]>> deliveryReport)
         {
             var kafkaMessage = BuildMessage(message);
 
             _producer.Produce(message.Header.Topic, kafkaMessage, deliveryReport);
         }
 
-        public async Task PublishMessageAsync(Message message, Action<DeliveryResult<string, string>> deliveryReport)
+        public async Task PublishMessageAsync(Message message, Action<DeliveryResult<string, byte[]>> deliveryReport)
         {
             var kafkaMessage = BuildMessage(message);
 
             var deliveryResult = await _producer.ProduceAsync(message.Header.Topic, kafkaMessage);
             deliveryReport(deliveryResult);
         }
 
-        private static Message<string, string> BuildMessage(Message message)
+        private static Message<string, byte[]> BuildMessage(Message message)
         {
             var headers = new Headers()
             {
@@ -53,10 +57,13 @@ private static Message<string, string> BuildMessage(Message message)
 
             if (!string.IsNullOrEmpty(message.Header.PartitionKey))
                 headers.Add(HeaderNames.PARTITIONKEY, message.Header.PartitionKey.ToByteArray());
+
+            if (!string.IsNullOrEmpty(message.Header.ContentType))
+                headers.Add(HeaderNames.CONTENT_TYPE, message.Header.ContentType.ToByteArray());
 
             message.Header.Bag.Each((header) =>
             {
-                if (!_headersToReset.Any(htr => htr.Equals(header.Key)))
+                if (!s_headersToReset.Any(htr => htr.Equals(header.Key)))
                 {
                     switch (header.Value)
                     {
@@ -73,11 +80,11 @@ private static Message<string, string> BuildMessage(Message message)
                 }
             });
 
-            var kafkaMessage = new Message<string, string>()
+            var kafkaMessage = new Message<string, byte[]>()
             {
                 Headers = headers,
                 Key = message.Header.PartitionKey,
-                Value = message.Body.Value
+                Value = message.Body.Bytes
             };
             return kafkaMessage;
         }

src/Paramore.Brighter.MessagingGateway.RMQ/RmqMessagePublisher.cs

@@ -117,7 +117,7 @@ public void PublishMessage(Message message, int delayMilliseconds)
                 CreateBasicProperties(
                     messageId, 
                     message.Header.TimeStamp, 
-                    message.Body.BodyType, 
+                    message.Body.ContentType, 
                     message.Header.ContentType, 
                     message.Header.ReplyTo,
                     message.Persist,
@@ -169,7 +169,7 @@ public void RequeueMessage(Message message, string queueName, int delayMilliseco
                 CreateBasicProperties(
                     messageId, 
                     message.Header.TimeStamp, 
-                    message.Body.BodyType, 
+                    message.Body.ContentType, 
                     message.Header.ContentType, 
                     message.Header.ReplyTo,
                     message.Persist,

src/Paramore.Brighter.Outbox.DynamoDB/MessageItem.cs

@@ -105,7 +105,7 @@ public MessageItem(Message message)
             CreatedAt = $"{date}";
             HeaderBag = JsonSerializer.Serialize(message.Header.Bag, JsonSerialisationOptions.Options);
             PartitionKey = message.Header.PartitionKey;
-            Body = message.Body.Value;
+            Body = message.Body.ToCharacterEncodedString(CharacterEncoding.Base64);
             DeliveryTime = 0;
         }
 
@@ -132,7 +132,7 @@ public Message ConvertToMessage()
                 header.Bag.Add(key, bag[key]);
             }
 
-            var body = new MessageBody(Body);
+            var body = new MessageBody(Convert.FromBase64String(Body), ContentType);
 
             return new Message(header, body);
         }

src/Paramore.Brighter/CharacterEncoding.cs

@@ -0,0 +1,46 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2023 Ian Cooper <[email protected]>
+
+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
+{
+    /// <summary>
+    /// How is the message body being encoded
+    /// </summary>
+    public enum CharacterEncoding
+    {
+        /// <summary>
+        /// Ascii text
+        /// </summary>
+        ASCII,
+        /// <summary>
+        /// Bse64 or asciiArmor
+        /// </summary>
+        Base64,
+       /// <summary>
+        /// UTF-8 Text
+        /// </summary>
+        UTF8,
+    }
+}

src/Paramore.Brighter/Message.cs

@@ -77,6 +77,7 @@ public Message(MessageHeader header, MessageBody body)
         {
             Header = header;
             Body = body;
+            Header.ContentType = string.IsNullOrEmpty(Header.ContentType) ? Body.ContentType: Header.ContentType;
 
             Header.UpdateTelemetryFromHeaders();
         }