new edits for binary data handling

remove binary logic

update SerializeEnvelope test

add field to message envelope instead

all tests pass

add tests

update tests

refactor

remove unused code

Author: GarrettBeatty

src/AWS.Messaging/AWS.Messaging.csproj

@@ -48,4 +48,4 @@
     <None Include="..\..\README.md" Pack="true" PackagePath="" />
   </ItemGroup>
 
-</Project>
+</Project>

src/AWS.Messaging/MessageEnvelope.cs

@@ -44,6 +44,12 @@ public abstract class MessageEnvelope
     [JsonPropertyName("time")]
     public DateTimeOffset TimeStamp { get; set; } = DateTimeOffset.MinValue;
 
+    /// <summary>
+    /// The data content type.
+    /// </summary>
+    [JsonPropertyName("datacontenttype")]
+    public string? DataContentType { get; set; } = "application/json";
+
     /// <summary>
     /// This stores different metadata that is not modeled as a top-level property in MessageEnvelope class.
     /// These entries will also be serialized as top-level properties when sending the message, which

src/AWS.Messaging/Serialization/EnvelopeSerializer.cs

@@ -67,7 +67,8 @@ public async ValueTask<MessageEnvelope<T>> CreateEnvelopeAsync<T>(T message)
             Version = CLOUD_EVENT_SPEC_VERSION,
             MessageTypeIdentifier = publisherMapping.MessageTypeIdentifier,
             TimeStamp = timeStamp,
-            Message = message
+            Message = message,
+            // DataContentType = "" // TODO
         };
     }
 
@@ -91,10 +92,12 @@ public async ValueTask<string> SerializeAsync<T>(MessageEnvelope<T> envelope)
                 ["source"] = envelope.Source?.ToString(),
                 ["specversion"] = envelope.Version,
                 ["type"] = envelope.MessageTypeIdentifier,
-                ["time"] = envelope.TimeStamp,
-                ["data"] = _messageSerializer.Serialize(message)
+                ["time"] = envelope.TimeStamp
             };
 
+            // See https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/formats/json-format.md#31-handling-of-data for more details.
+            SerializeData(message, blob, envelope.DataContentType);
+
             // Write any Metadata as top-level keys
             // This may be useful for any extensions defined in
             // https://github.com/cloudevents/spec/tree/main/cloudevents/extensions
@@ -107,17 +110,17 @@ public async ValueTask<string> SerializeAsync<T>(MessageEnvelope<T> envelope)
             }
 
             var jsonString = blob.ToJsonString();
-            var serializedMessage = await InvokePostSerializationCallback(jsonString);
+            var finalSerializedMessage  = await InvokePostSerializationCallback(jsonString);
 
             if (_messageConfiguration.LogMessageContent)
             {
-                _logger.LogTrace("Serialized the MessageEnvelope object as the following raw string:\n{SerializedMessage}", serializedMessage);
+                _logger.LogTrace("Serialized the MessageEnvelope object as the following raw string:\n{SerializedMessage}", finalSerializedMessage );
             }
             else
             {
                 _logger.LogTrace("Serialized the MessageEnvelope object to a raw string");
             }
-            return serializedMessage;
+            return finalSerializedMessage ;
         }
         catch (JsonException) when (!_messageConfiguration.LogMessageContent)
         {
@@ -131,6 +134,16 @@ public async ValueTask<string> SerializeAsync<T>(MessageEnvelope<T> envelope)
         }
     }
 
+    private string ExtractDataContent(JsonElement dataElement, string? dataContentType)
+    {
+        return IsJsonContentType(dataContentType)
+            ? dataElement.ValueKind == JsonValueKind.String
+                ? dataElement.GetString()!
+                : dataElement.GetRawText()
+            : dataElement.GetString()
+              ?? throw new InvalidDataException("Data must be a string for non-JSON content type");
+    }
+
     /// <inheritdoc/>
     public async ValueTask<ConvertToEnvelopeResult> ConvertToEnvelopeAsync(Message sqsMessage)
     {
@@ -143,6 +156,7 @@ public async ValueTask<ConvertToEnvelopeResult> ConvertToEnvelopeAsync(Message s
             var subscriberMapping = GetAndValidateSubscriberMapping(parsedResult.Envelope.MessageTypeIdentifier);
             var deserializedMessage = DeserializeDataContent(
                 parsedResult.Envelope.Message!,
+                parsedResult.Envelope.DataContentType,
                 subscriberMapping.MessageType);
 
             // Create and populate final envelope
@@ -174,11 +188,35 @@ public async ValueTask<ConvertToEnvelopeResult> ConvertToEnvelopeAsync(Message s
     private async Task<ParsedEnvelopeResult> ParseMessageEnvelope(Message sqsMessage)
     {
         sqsMessage.Body = await InvokePreDeserializationCallback(sqsMessage.Body);
-        var messageEnvelopeConfiguration = GetMessageEnvelopeConfiguration(sqsMessage);
-        var intermediateEnvelope = JsonSerializer.Deserialize<MessageEnvelope<string>>(messageEnvelopeConfiguration.MessageEnvelopeBody!)!;
+        var config = GetMessageEnvelopeConfiguration(sqsMessage);
+
+        using var document = JsonDocument.Parse(config.MessageEnvelopeBody!);
+        var root = document.RootElement;
+
+        if (!root.TryGetProperty("data", out var dataElement))
+        {
+            throw new InvalidDataException("Message envelope is missing required 'data' field");
+        }
+
+        string? dataContentType = root.TryGetProperty("datacontenttype", out var contentTypeElement)
+            ? contentTypeElement.GetString()
+            : "application/json";
+
+        string dataContent = ExtractDataContent(dataElement, dataContentType);
+
+        // Create intermediate envelope with all properties
+        var envelopeJson = new JsonObject();
+        foreach (var property in root.EnumerateObject())
+        {
+            envelopeJson[property.Name] = property.Name == "data"
+                ? JsonValue.Create(dataContent)
+                : JsonNode.Parse(property.Value.GetRawText());
+        }
+
+        var intermediateEnvelope = JsonSerializer.Deserialize<MessageEnvelope<string>>(envelopeJson.ToJsonString())!;
         ValidateMessageEnvelope(intermediateEnvelope);
 
-        return new ParsedEnvelopeResult(intermediateEnvelope, messageEnvelopeConfiguration);
+        return new ParsedEnvelopeResult(intermediateEnvelope, config);
     }
 
     private MessageEnvelope CreateFinalEnvelope(
@@ -204,6 +242,7 @@ private MessageEnvelope CreateFinalEnvelope(
         finalEnvelope.MessageTypeIdentifier = intermediateEnvelope.MessageTypeIdentifier;
         finalEnvelope.TimeStamp = intermediateEnvelope.TimeStamp;
         finalEnvelope.Metadata = intermediateEnvelope.Metadata;
+        finalEnvelope.DataContentType = intermediateEnvelope.DataContentType;
         finalEnvelope.SQSMetadata = config.SQSMetadata;
         finalEnvelope.SNSMetadata = config.SNSMetadata;
         finalEnvelope.EventBridgeMetadata = config.EventBridgeMetadata;
@@ -232,9 +271,87 @@ private SubscriberMapping GetAndValidateSubscriberMapping(string messageTypeIden
         return subscriberMapping;
     }
 
-    private object DeserializeDataContent(string dataContent, Type messageType)
+    private object DeserializeDataContent(string dataContent, string? dataContentType, Type messageType)
     {
-        return _messageSerializer.Deserialize(dataContent, messageType);
+        if (IsJsonContentType(dataContentType))
+        {
+            return _messageSerializer.Deserialize(dataContent, messageType);
+        }
+
+        if (messageType == typeof(string))
+        {
+            return dataContent;
+        }
+
+        throw new InvalidOperationException(
+            $"Cannot deserialize non-JSON content type {dataContentType} to type {messageType}");
+    }
+
+    private void SerializeData<T>(T message, JsonObject blob, string? dataContentType)
+    {
+        if (message == null)
+        {
+            throw new ArgumentNullException("The underlying application message cannot be null");
+        }
+
+        // Serialize the message
+        var serializedMessage = _messageSerializer.Serialize(message);
+
+        // Determine if the serialized message is valid JSON
+        // Wed do this because _messageSerializer is injected and there is no guarantee that it serializes to json.
+        bool isJson = IsValidJson(serializedMessage);
+        blob["datacontenttype"] = dataContentType;
+
+        if (IsJsonContentType(dataContentType))
+        {
+            if (isJson)
+            {
+                // If it's valid JSON, parse it to maintain structure
+                blob["data"] = JsonNode.Parse(serializedMessage);
+            }
+            else
+            {
+                // If it's not valid JSON but content type indicates JSON,
+                // log warning and store as string
+                _logger.LogWarning("Data was serialized as non-JSON, but datacontenttype indicates JSON format. Storing as string.");
+                blob["data"] = serializedMessage;
+            }
+        }
+        else
+        {
+            // For non-JSON content types, store as string
+            blob["data"] = serializedMessage;
+        }
+    }
+
+    private bool IsJsonContentType(string? contentType)
+    {
+        if (string.IsNullOrEmpty(contentType))
+        {
+            // If datacontenttype is unspecified, treat as application/json
+            return true;
+        }
+
+        // Strip any parameters (anything after ';')
+        var mediaType = contentType.Split(';')[0].Trim().ToLowerInvariant();
+
+        return mediaType.EndsWith("/json") || // Matches */json
+               (mediaType.Contains('/') && mediaType.EndsWith("+json")); // Matches */*+json
+    }
+
+    private bool IsValidJson(string strInput)
+    {
+        if (string.IsNullOrWhiteSpace(strInput)) return false;
+
+        try
+        {
+            JsonDocument.Parse(strInput);
+            return true;
+        }
+        catch (JsonException)
+        {
+            return false;
+        }
     }
 
     private void ValidateMessageEnvelope<T>(MessageEnvelope<T>? messageEnvelope)

test/AWS.Messaging.UnitTests/MessageHandlers/Handlers.cs

@@ -28,6 +28,14 @@ public Task<MessageProcessStatus> HandleAsync(MessageEnvelope<AddressInfo> messa
     }
 }
 
+public class PlainTextHandler : IMessageHandler<string>
+{
+    public Task<MessageProcessStatus> HandleAsync(MessageEnvelope<string> messageEnvelope, CancellationToken token = default)
+    {
+        return Task.FromResult(MessageProcessStatus.Success());
+    }
+}
+
 public class ChatMessageHandlerWithDependencies : IMessageHandler<ChatMessage>
 {
     private readonly IDependentThing _thingDoer;