Farfetch · joelfoliveira · Oct 30, 2023 · Sep 19, 2023
diff --git a/.github/workflows/deploy-website.yml b/.github/workflows/deploy-website.yml
@@ -36,12 +36,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Admin.dll website/docs/reference/KafkaFlow.Admin --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.BatchConsume.dll website/docs/reference/KafkaFlow.BatchConsume --type-folders
-        shell: bash
-
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.Compressor.dll website/docs/reference/KafkaFlow.Compressor --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Extensions.Hosting.dll website/docs/reference/KafkaFlow.Extensions.Hosting --type-folders
         shell: bash
 
@@ -57,9 +51,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.SchemaRegistry.dll website/docs/reference/KafkaFlow.SchemaRegistry --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.dll website/docs/reference/KafkaFlow.Serializer --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.JsonCore.dll website/docs/reference/KafkaFlow.Serializer.JsonCore --type-folders
         shell: bash
 
@@ -78,9 +69,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf.dll website/docs/reference/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.TypedHandler.dll website/docs/reference/KafkaFlow.TypedHandler --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Unity.dll website/docs/reference/KafkaFlow.Unity --type-folders
         shell: bash
 

diff --git a/.github/workflows/test-deploy-website.yml b/.github/workflows/test-deploy-website.yml
@@ -30,12 +30,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Admin.dll website/docs/reference/KafkaFlow.Admin --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.BatchConsume.dll website/docs/reference/KafkaFlow.BatchConsume --type-folders
-        shell: bash
-
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.Compressor.dll website/docs/reference/KafkaFlow.Compressor --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Extensions.Hosting.dll website/docs/reference/KafkaFlow.Extensions.Hosting --type-folders
         shell: bash
 
@@ -51,9 +45,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.SchemaRegistry.dll website/docs/reference/KafkaFlow.SchemaRegistry --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.dll website/docs/reference/KafkaFlow.Serializer --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.JsonCore.dll website/docs/reference/KafkaFlow.Serializer.JsonCore --type-folders
         shell: bash
 
@@ -72,9 +63,6 @@ jobs:
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf.dll website/docs/reference/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf --type-folders
         shell: bash
 
-      - run: xmldocmd-docusaurus ./drop/KafkaFlow.TypedHandler.dll website/docs/reference/KafkaFlow.TypedHandler --type-folders
-        shell: bash
-
       - run: xmldocmd-docusaurus ./drop/KafkaFlow.Unity.dll website/docs/reference/KafkaFlow.Unity --type-folders
         shell: bash
 

diff --git a/samples/KafkaFlow.Sample.BatchOperations/KafkaFlow.Sample.BatchOperations.csproj b/samples/KafkaFlow.Sample.BatchOperations/KafkaFlow.Sample.BatchOperations.csproj
@@ -22,11 +22,9 @@
 
     <ItemGroup>
       <ProjectReference Include="..\..\src\KafkaFlow.Abstractions\KafkaFlow.Abstractions.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.BatchConsume\KafkaFlow.BatchConsume.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample.BatchOperations/PrintConsoleMiddleware.cs b/samples/KafkaFlow.Sample.BatchOperations/PrintConsoleMiddleware.cs
@@ -1,7 +1,7 @@
 using System;
 using System.Linq;
 using System.Threading.Tasks;
-using KafkaFlow.BatchConsume;
+using KafkaFlow.Batching;
 
 namespace KafkaFlow.Sample.BatchOperations;
 

diff --git a/samples/KafkaFlow.Sample.BatchOperations/Program.cs b/samples/KafkaFlow.Sample.BatchOperations/Program.cs
@@ -1,7 +1,6 @@
 using System;
 using System.Linq;
 using KafkaFlow;
-using KafkaFlow.BatchConsume;
 using KafkaFlow.Producers;
 using KafkaFlow.Sample.BatchOperations;
 using KafkaFlow.Serializer;
@@ -35,8 +34,8 @@
                         .WithWorkersCount(1)
                         .AddMiddlewares(
                             middlewares => middlewares
-                                .AddSerializer<JsonCoreSerializer>()
-                                .BatchConsume(10, TimeSpan.FromSeconds(10))
+                                .AddDeserializer<JsonCoreDeserializer>()
+                                .AddBatching(10, TimeSpan.FromSeconds(10))
                                 .Add<PrintConsoleMiddleware>()
                         )
                 )

diff --git a/samples/KafkaFlow.Sample.ConsumerThrottling/KafkaFlow.Sample.ConsumerThrottling.csproj b/samples/KafkaFlow.Sample.ConsumerThrottling/KafkaFlow.Sample.ConsumerThrottling.csproj
@@ -22,8 +22,6 @@
         <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
-        <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
-        <ProjectReference Include="..\..\src\KafkaFlow.TypedHandler\KafkaFlow.TypedHandler.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample.ConsumerThrottling/Program.cs b/samples/KafkaFlow.Sample.ConsumerThrottling/Program.cs
@@ -48,7 +48,7 @@ The ThrottleConsumer mechanism works by checking the lag of "consumerA" every 1
                                     .WithWorkersCount(1)
                                     .AddMiddlewares(
                                         middlewares => middlewares
-                                            .AddSerializer<JsonCoreSerializer>()
+                                            .AddDeserializer<JsonCoreDeserializer>()
                                             .Add<ProcessMessagesMiddleware>()
                                     )
                             )
@@ -68,7 +68,7 @@ The ThrottleConsumer mechanism works by checking the lag of "consumerA" every 1
                                                     .AddAction(a => a.AboveThreshold(10).ApplyDelay(1_000))
                                                     .AddAction(a => a.AboveThreshold(20).ApplyDelay(5_000))
                                                     .AddAction(a => a.AboveThreshold(30).ApplyDelay(10_000)))
-                                            .AddSerializer<JsonCoreSerializer>()
+                                            .AddDeserializer<JsonCoreDeserializer>()
                                             .Add<ProcessMessagesMiddleware>()
                                     )
                             )

diff --git a/samples/KafkaFlow.Sample.FlowControl/KafkaFlow.Sample.FlowControl.csproj b/samples/KafkaFlow.Sample.FlowControl/KafkaFlow.Sample.FlowControl.csproj
@@ -26,7 +26,6 @@
       <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.NewtonsoftJson\KafkaFlow.Serializer.NewtonsoftJson.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample.FlowControl/Program.cs b/samples/KafkaFlow.Sample.FlowControl/Program.cs
@@ -35,7 +35,7 @@
                             .WithWorkersCount(1)
                             .AddMiddlewares(
                                 m => m
-                                    .AddSingleTypeSerializer<SampleMessage, NewtonsoftJsonSerializer>()
+                                    .AddSingleTypeDeserializer<SampleMessage, NewtonsoftJsonDeserializer>()
                                     .Add<PrintConsoleMiddleware>()
                             )
                     );

diff --git a/samples/KafkaFlow.Sample.PauseConsumerOnError/KafkaFlow.Sample.PauseConsumerOnError.csproj b/samples/KafkaFlow.Sample.PauseConsumerOnError/KafkaFlow.Sample.PauseConsumerOnError.csproj
@@ -14,8 +14,6 @@
       <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.TypedHandler\KafkaFlow.TypedHandler.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample.PauseConsumerOnError/MessageHandler.cs b/samples/KafkaFlow.Sample.PauseConsumerOnError/MessageHandler.cs
@@ -1,4 +1,4 @@
-using KafkaFlow.TypedHandler;
+using KafkaFlow.Middlewares.TypedHandler;
 
 namespace KafkaFlow.Sample.PauseConsumerOnError;
 

diff --git a/samples/KafkaFlow.Sample.PauseConsumerOnError/Program.cs b/samples/KafkaFlow.Sample.PauseConsumerOnError/Program.cs
@@ -2,7 +2,6 @@
 using KafkaFlow.Producers;
 using KafkaFlow.Sample.PauseConsumerOnError;
 using KafkaFlow.Serializer;
-using KafkaFlow.TypedHandler;
 using Microsoft.Extensions.DependencyInjection;
 
 var services = new ServiceCollection();
@@ -33,7 +32,7 @@
                             middlewares =>
                                 middlewares
                                     .Add<PauseConsumerOnExceptionMiddleware>()
-                                    .AddSerializer<JsonCoreSerializer>()
+                                    .AddDeserializer<JsonCoreDeserializer>()
                                     .AddTypedHandlers(h => h.AddHandler<MessageHandler>())
                         )
                 )

diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/AvroMessageHandler.cs b/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/AvroMessageHandler.cs
@@ -1,20 +1,21 @@
-namespace KafkaFlow.Sample.SchemaRegistry.Handlers;
-
-using System;
-using System.Threading.Tasks;
-using global::SchemaRegistry;
-using TypedHandler;
-
-public class AvroMessageHandler : IMessageHandler<AvroLogMessage>
+namespace KafkaFlow.Sample.SchemaRegistry.Handlers
 {
-    public Task Handle(IMessageContext context, AvroLogMessage message)
+    using System;
+    using System.Threading.Tasks;
+    using KafkaFlow.Middlewares.TypedHandler;
+    using global::SchemaRegistry;
+
+    public class AvroMessageHandler : IMessageHandler<AvroLogMessage>
     {
-        Console.WriteLine(
-            "Partition: {0} | Offset: {1} | Message: {2} | Avro",
-            context.ConsumerContext.Partition,
-            context.ConsumerContext.Offset,
-            message.Severity.ToString());
+        public Task Handle(IMessageContext context, AvroLogMessage message)
+        {
+            Console.WriteLine(
+                "Partition: {0} | Offset: {1} | Message: {2} | Avro",
+                context.ConsumerContext.Partition,
+                context.ConsumerContext.Offset,
+                message.Severity.ToString());
 
-        return Task.CompletedTask;
+            return Task.CompletedTask;
+        }
     }
 }
diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/AvroMessageHandler2.cs b/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/AvroMessageHandler2.cs
@@ -3,7 +3,7 @@
 using System;
 using System.Threading.Tasks;
 using global::SchemaRegistry;
-using TypedHandler;
+using KafkaFlow.Middlewares.TypedHandler;
 
 public class AvroMessageHandler2 : IMessageHandler<AvroLogMessage2>
 {

diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/JsonMessageHandler.cs b/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/JsonMessageHandler.cs
@@ -3,7 +3,7 @@
 using System;
 using System.Threading.Tasks;
 using global::SchemaRegistry;
-using TypedHandler;
+using KafkaFlow.Middlewares.TypedHandler;
 
 public class JsonMessageHandler : IMessageHandler<JsonLogMessage>
 {

diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/ProtobufMessageHandler.cs b/samples/KafkaFlow.Sample.SchemaRegistry/Handlers/ProtobufMessageHandler.cs
@@ -1,7 +1,7 @@
 
 using System;
 using System.Threading.Tasks;
-using KafkaFlow.TypedHandler;
+using KafkaFlow.Middlewares.TypedHandler;
 using SchemaRegistry;
 
 namespace KafkaFlow.Sample.SchemaRegistry.Handlers;

diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/KafkaFlow.Sample.SchemaRegistry.csproj b/samples/KafkaFlow.Sample.SchemaRegistry/KafkaFlow.Sample.SchemaRegistry.csproj
@@ -24,8 +24,6 @@
         <ProjectReference Include="..\..\src\KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro\KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow.Serializer.SchemaRegistry.ConfluentJson\KafkaFlow.Serializer.SchemaRegistry.ConfluentJson.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf\KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf.csproj" />
-        <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
-        <ProjectReference Include="..\..\src\KafkaFlow.TypedHandler\KafkaFlow.TypedHandler.csproj" />
         <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample.SchemaRegistry/Program.cs b/samples/KafkaFlow.Sample.SchemaRegistry/Program.cs
@@ -5,7 +5,6 @@
 using KafkaFlow;
 using KafkaFlow.Producers;
 using KafkaFlow.Sample.SchemaRegistry.Handlers;
-using KafkaFlow.TypedHandler;
 using Microsoft.Extensions.DependencyInjection;
 using SchemaRegistry;
 
@@ -74,7 +73,7 @@
                         .WithAutoOffsetReset(AutoOffsetReset.Latest)
                         .AddMiddlewares(
                             middlewares => middlewares
-                                .AddSchemaRegistryAvroSerializer()
+                                .AddSchemaRegistryAvroDeserializer()
                                 .AddTypedHandlers(
                                     handlers => handlers
                                         .AddHandler<AvroMessageHandler>()
@@ -103,7 +102,7 @@
                         .WithAutoOffsetReset(AutoOffsetReset.Latest)
                         .AddMiddlewares(
                             middlewares => middlewares
-                                .AddSchemaRegistryProtobufSerializer()
+                                .AddSchemaRegistryProtobufDeserializer()
                                 .AddTypedHandlers(handlers => handlers.AddHandler<ProtobufMessageHandler>())
                         )
                 )

diff --git a/samples/KafkaFlow.Sample/KafkaFlow.Sample.csproj b/samples/KafkaFlow.Sample/KafkaFlow.Sample.csproj
@@ -19,13 +19,10 @@
     <ItemGroup>
       <ProjectReference Include="..\..\src\KafkaFlow.Admin\KafkaFlow.Admin.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Compressor.Gzip\KafkaFlow.Compressor.Gzip.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.Compressor\KafkaFlow.Compressor.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.ProtobufNet\KafkaFlow.Serializer.ProtobufNet.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
-      <ProjectReference Include="..\..\src\KafkaFlow.TypedHandler\KafkaFlow.TypedHandler.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow\KafkaFlow.csproj" />
     </ItemGroup>
 

diff --git a/samples/KafkaFlow.Sample/PrintConsoleHandler.cs b/samples/KafkaFlow.Sample/PrintConsoleHandler.cs
@@ -1,6 +1,6 @@
 using System;
 using System.Threading.Tasks;
-using KafkaFlow.TypedHandler;
+using KafkaFlow.Middlewares.TypedHandler;
 
 namespace KafkaFlow.Sample;
 

diff --git a/samples/KafkaFlow.Sample/Program.cs b/samples/KafkaFlow.Sample/Program.cs
@@ -4,7 +4,6 @@
 using KafkaFlow.Producers;
 using KafkaFlow.Sample;
 using KafkaFlow.Serializer;
-using KafkaFlow.TypedHandler;
 using Microsoft.Extensions.DependencyInjection;
 
 var services = new ServiceCollection();
@@ -33,7 +32,7 @@
                         .WithWorkersCount(3)
                         .AddMiddlewares(
                             middlewares => middlewares
-                                .AddSerializer<ProtobufNetSerializer>()
+                                .AddDeserializer<ProtobufNetDeserializer>()
                                 .AddTypedHandlers(h => h.AddHandler<PrintConsoleHandler>())
                         )
                 )

diff --git a/...kaFlow.Abstractions/IMessageCompressor.cs → src/KafkaFlow.Abstractions/ICompressor.cs b/...kaFlow.Abstractions/IMessageCompressor.cs → src/KafkaFlow.Abstractions/ICompressor.cs
@@ -3,20 +3,13 @@ namespace KafkaFlow
     /// <summary>
     /// Used to create a message compressor
     /// </summary>
-    public interface IMessageCompressor
+    public interface ICompressor
     {
         /// <summary>
         /// Compress the given message
         /// </summary>
         /// <param name="message">The message to be compressed</param>
         /// <returns>The compressed message</returns>
         byte[] Compress(byte[] message);
-
-        /// <summary>
-        /// Decompress the given message
-        /// </summary>
-        /// <param name="message">The message to be decompressed</param>
-        /// <returns>The decompressed message</returns>
-        byte[] Decompress(byte[] message);
     }
 }
diff --git a/src/KafkaFlow.Abstractions/IDecompressor.cs b/src/KafkaFlow.Abstractions/IDecompressor.cs
@@ -0,0 +1,15 @@
+namespace KafkaFlow
+{
+    /// <summary>
+    /// Used to create a message decompressor
+    /// </summary>
+    public interface IDecompressor
+    {
+        /// <summary>
+        /// Decompress the given message
+        /// </summary>
+        /// <param name="message">The message to be decompressed</param>
+        /// <returns>The decompressed message</returns>
+        byte[] Decompress(byte[] message);
+    }
+}
diff --git a/src/KafkaFlow.Abstractions/IDeserializer.cs b/src/KafkaFlow.Abstractions/IDeserializer.cs
@@ -0,0 +1,21 @@
+namespace KafkaFlow
+{
+    using System;
+    using System.IO;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Used to implement a message serializer
+    /// </summary>
+    public interface IDeserializer
+    {
+        /// <summary>
+        /// Deserializes the given message
+        /// </summary>
+        /// <param name="input">A stream to read the data to be deserialized</param>
+        /// <param name="type">The type to be created</param>
+        /// <param name="context">An object containing metadata</param>
+        /// <returns>The deserialized message</returns>
+        Task<object> DeserializeAsync(Stream input, Type type, ISerializerContext context);
+    }
+}
diff --git a/...KafkaFlow.TypedHandler/IMessageHandler.cs → ...KafkaFlow.Abstractions/IMessageHandler.cs b/...KafkaFlow.TypedHandler/IMessageHandler.cs → ...KafkaFlow.Abstractions/IMessageHandler.cs
@@ -1,4 +1,4 @@
-namespace KafkaFlow.TypedHandler
+namespace KafkaFlow
 {
     using System.Threading.Tasks;