Avoid double deserialization with Kafka

Fixes #1246

Author: rngcntr

src/Motor.Extensions.Hosting.Kafka/KafkaClientExtensions.cs

@@ -1,4 +1,7 @@
 using System;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Nodes;
 using CloudNative.CloudEvents;
 using CloudNative.CloudEvents.Kafka;
 using Confluent.Kafka;
@@ -25,6 +28,11 @@ public static MotorCloudEvent<byte[]> ToMotorCloudEvent(this Message<string?, by
         {
             throw new ArgumentException("Source property of CloudEvent is null");
         }
+
+        if (cloudEvent.Data is JsonElement element)
+        {
+            cloudEvent.Data = Encoding.UTF8.GetBytes(element.GetRawText());
+        }
         var motorCloudEvent = new MotorCloudEvent<byte[]>(applicationNameService, (byte[])cloudEvent.Data,
             cloudEvent.Type, cloudEvent.Source, cloudEvent.Id, cloudEvent.Time, cloudEvent.DataContentType);
         foreach (var (key, value) in cloudEvent.GetPopulatedAttributes())

src/Motor.Extensions.Hosting.Kafka/Motor.Extensions.Hosting.Kafka.csproj

@@ -22,4 +22,8 @@
 
     <Import Project="$(MSBuildThisFileDirectory)../../shared.csproj" />
 
+    <ItemGroup>
+        <InternalsVisibleTo Include="Motor.Extensions.Hosting.Kafka_UnitTest" />
+    </ItemGroup>
+    
 </Project>

test/Motor.Extensions.Hosting.Kafka_UnitTest/KafkaClientExtensionsRegressionTests.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Globalization;
+using System.Text;
+using CloudNative.CloudEvents.SystemTextJson;
+using Confluent.Kafka;
+using Moq;
+using Motor.Extensions.Hosting.CloudEvents;
+using Motor.Extensions.Hosting.Kafka;
+using Xunit;
+
+namespace Motor.Extensions.Hosting.Kafka_UnitTest;
+
+public class KafkaClientExtensionsRegressionTests
+{
+    [Fact]
+    public void KafkaClientExtensionsRegression_ToMotorCloudEvent_DataIsByteArrayt ()
+    {
+        var applicationNameServiceMock = Mock.Of<IApplicationNameService>();
+        var actualContent = "{\"key\":\"value\"}";
+        var message = new Message<string, byte[]>
+        {
+            Headers = new Headers
+            {
+                { "content-type", "application/cloudevents+json"u8.ToArray() }
+            },
+            Value = Encoding.UTF8.GetBytes($@"
+            {{
+                ""specversion"": ""1.0"",
+                ""type"": ""imaginary.type"",
+                ""source"": ""/imaginary-source"",
+                ""id"": ""{Guid.NewGuid()}"",
+                ""time"": ""{DateTimeOffset.UnixEpoch.ToString("o", CultureInfo.InvariantCulture)}"",
+                ""datacontenttype"": ""application/json"",
+                ""data"": {actualContent}
+            }}")
+        };
+        var cloudEvent = message.ToMotorCloudEvent(applicationNameServiceMock, new JsonEventFormatter());
+        Assert.Equal(Encoding.UTF8.GetBytes(actualContent), cloudEvent.Data);
+    }
+}