Added Json encoder for Avro

src/net/KEFCore.SerDes.Avro/KEFCore.SerDes.Avro.Compiler.csproj

@@ -15,7 +15,8 @@
 	</PropertyGroup>
 	<ItemGroup>
 	  <Compile Remove="AvroValueContainer.cs" />
-	  <Compile Remove="KEFCoreSerDes.Avro.cs" />
+	  <Compile Remove="KEFCoreSerDes.AvroBinary.cs" />
+	  <Compile Remove="KEFCoreSerDes.AvroJson.cs" />
 	</ItemGroup>
 
 	<ItemGroup>

src/net/KEFCore.SerDes.Avro/KEFCoreSerDes.Avro.cs → src/net/KEFCore.SerDes.Avro/KEFCoreSerDes.AvroBinary.cs

@@ -27,13 +27,13 @@
 namespace MASES.EntityFrameworkCore.KNet.Serialization.Avro;
 
 /// <summary>
-/// Avro extension of <see cref="KNetSerDes{T}"/>, for example <see href="https://masesgroup.github.io/KNet/articles/usageSerDes.html"/>
+/// Avro Binary encoder extension of <see cref="KNetSerDes{T}"/>, for example <see href="https://masesgroup.github.io/KNet/articles/usageSerDes.html"/>
 /// </summary>
 /// <typeparam name="T"></typeparam>
-public class KEFCoreSerDesAvro<T> : KNetSerDes<T>
+public class KEFCoreSerDesAvroBinary<T> : KNetSerDes<T>
 {
-    static SpecificDefaultWriter SpecificWriter = new SpecificDefaultWriter(AvroValueContainer._SCHEMA);
-    static SpecificDefaultReader SpecificReader = new SpecificDefaultReader(AvroValueContainer._SCHEMA, AvroValueContainer._SCHEMA);
+    static readonly SpecificDefaultWriter SpecificWriter = new SpecificDefaultWriter(AvroValueContainer._SCHEMA);
+    static readonly SpecificDefaultReader SpecificReader = new SpecificDefaultReader(AvroValueContainer._SCHEMA, AvroValueContainer._SCHEMA);
 
     /// <inheritdoc cref="KNetSerDes{T}.Serialize(string, T)"/>
     public override byte[] Serialize(string topic, T data)

src/net/KEFCore.SerDes.Avro/KEFCoreSerDes.AvroJson.cs

@@ -0,0 +1,66 @@
+/*
+*  Copyright 2023 MASES s.r.l.
+*
+*  Licensed under the Apache License, Version 2.0 (the "License");
+*  you may not use this file except in compliance with the License.
+*  You may obtain a copy of the License at
+*
+*  http://www.apache.org/licenses/LICENSE-2.0
+*
+*  Unless required by applicable law or agreed to in writing, software
+*  distributed under the License is distributed on an "AS IS" BASIS,
+*  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+*  See the License for the specific language governing permissions and
+*  limitations under the License.
+*
+*  Refer to LICENSE for more information.
+*/
+
+#nullable enable
+
+using Avro.IO;
+using Avro.Specific;
+using MASES.EntityFrameworkCore.KNet.Serialization.Avro.Storage;
+using MASES.KNet.Serialization;
+using Org.Apache.Kafka.Common.Header;
+
+namespace MASES.EntityFrameworkCore.KNet.Serialization.Avro;
+
+/// <summary>
+/// Avro Json encoder extension of <see cref="KNetSerDes{T}"/>, for example <see href="https://masesgroup.github.io/KNet/articles/usageSerDes.html"/>
+/// </summary>
+/// <typeparam name="T"></typeparam>
+public class KEFCoreSerDesAvroJson<T> : KNetSerDes<T>
+{
+    static readonly SpecificDefaultWriter SpecificWriter = new SpecificDefaultWriter(AvroValueContainer._SCHEMA);
+    static readonly SpecificDefaultReader SpecificReader = new SpecificDefaultReader(AvroValueContainer._SCHEMA, AvroValueContainer._SCHEMA);
+
+    /// <inheritdoc cref="KNetSerDes{T}.Serialize(string, T)"/>
+    public override byte[] Serialize(string topic, T data)
+    {
+        return SerializeWithHeaders(topic, null!, data);
+    }
+    /// <inheritdoc cref="KNetSerDes{T}.SerializeWithHeaders(string, Headers, T)"/>
+    public override byte[] SerializeWithHeaders(string topic, Headers headers, T data)
+    {
+        using MemoryStream memStream = new();
+        JsonEncoder encoder = new(AvroValueContainer._SCHEMA, memStream);
+        SpecificWriter.Write(data, encoder);
+        encoder.Flush();
+        return memStream.ToArray();
+    }
+    /// <inheritdoc cref="KNetSerDes{T}.Deserialize(string, byte[])"/>
+    public override T Deserialize(string topic, byte[] data)
+    {
+        return DeserializeWithHeaders(topic, null!, data);
+    }
+    /// <inheritdoc cref="KNetSerDes{T}.DeserializeWithHeaders(string, Headers, byte[])"/>
+    public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data)
+    {
+        using MemoryStream memStream = new(data);
+        JsonDecoder decoder = new(AvroValueContainer._SCHEMA, memStream);
+        T t = (T)Activator.CreateInstance(typeof(T))!;
+        t = SpecificReader.Read(t!, decoder);
+        return t;
+    }
+}

test/Common/ProgramConfig.cs

@@ -26,6 +26,7 @@ namespace MASES.EntityFrameworkCore.KNet.Test
 {
     public class ProgramConfig
     {
+        public bool UseAvroBinary { get; set; } = false;
         public bool EnableKEFCoreTracing { get; set; } = false;
         public bool UseInMemoryProvider { get; set; } = false;
         public bool UseModelBuilder { get; set; } = false;

test/KEFCore.Benchmark.Avro.Test/Program.cs

@@ -54,8 +54,6 @@ static void ReportString(string message)
 
         static void Main(string[] args)
         {
-            //AvroSerialization.TestAvroSerialization();
-
             BloggingContext context = null;
             var testWatcher = new Stopwatch();
             var globalWatcher = new Stopwatch();
@@ -91,7 +89,7 @@ static void Main(string[] args)
                     DbName = databaseName,
                     StreamsConfig = streamConfig,
                     ValueContainerType = typeof(AvroValueContainer<>),
-                    ValueSerializationType = typeof(KEFCoreSerDesAvro<>),
+                    ValueSerializationType = config.UseAvroBinary ? typeof(KEFCoreSerDesAvroBinary<>) : typeof(KEFCoreSerDesAvroJson<>),
                 })
                 {