add extended interface of IContentSerializer and JsonContentSerializer

src/DotNetCore.CAP.Kafka/DotNetCore.CAP.Kafka.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">
-    <WarningsAsErrors>NU1605</WarningsAsErrors>
+    <WarningsAsErrors>NU1605;NU1701</WarningsAsErrors>
     <NoWarn>NU1701</NoWarn>
   </PropertyGroup>
 

src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs

@@ -64,18 +64,19 @@ namespace DotNetCore.CAP.Abstractions
 
         protected abstract Task ExecuteAsync(IDbConnection dbConnection, IDbTransaction dbTransaction, CapPublishedMessage message);
 
-        #region private methods
-
-        private string Serialize<T>(T obj, string callbackName = null)
+        protected virtual string Serialize<T>(T obj, string callbackName = null)
         {
-            var message = new Message(obj)
+            var serializer = (IContentSerializer)ServiceProvider.GetService(typeof(IContentSerializer));
+
+            var message = new CapMessageDto(obj)
             {
                 CallbackName = callbackName
             };
-
-            return Helper.ToJson(message);
+            return serializer.Serialize(message);
         }
 
+        #region private methods
+
         private void PrepareConnectionForAdo(IDbConnection dbConnection, IDbTransaction dbTransaction)
         {
             DbConnection = dbConnection ?? throw new ArgumentNullException(nameof(dbConnection));

src/DotNetCore.CAP/Abstractions/IContentSerializer.cs

+using DotNetCore.CAP.Models;
+
+namespace DotNetCore.CAP.Abstractions
+{
+    public interface IContentSerializer
+    {
+        string Serialize<T>(T obj) where T : CapMessageDto, new();
+
+        T DeSerialize<T>(string content) where T : CapMessageDto, new();
+    }
+}

src/DotNetCore.CAP/CAP.ServiceCollectionExtensions.cs

@@ -33,6 +33,7 @@ namespace Microsoft.Extensions.DependencyInjection
 
             AddSubscribeServices(services);
 
+            services.TryAddSingleton<IContentSerializer, JsonContentSerializer>();
             services.TryAddSingleton<IConsumerServiceSelector, DefaultConsumerServiceSelector>();
             services.TryAddSingleton<IModelBinderFactory, ModelBinderFactory>();
             services.TryAddSingleton<IConsumerInvokerFactory, ConsumerInvokerFactory>();
@@ -42,7 +43,7 @@ namespace Microsoft.Extensions.DependencyInjection
             services.AddSingleton<IProcessingServer, CapProcessingServer>();
             services.AddSingleton<IBootstrapper, DefaultBootstrapper>();
             services.AddSingleton<IStateChanger, StateChanger>();
-
+           
             //Processors
             services.AddTransient<PublishQueuer>();
             services.AddTransient<SubscribeQueuer>();

src/DotNetCore.CAP/Internal/IConsumerInvoker.Default.cs

@@ -35,6 +35,7 @@ namespace DotNetCore.CAP.Internal
         {
             _logger.LogDebug("Executing consumer Topic: {0}", _consumerContext.ConsumerDescriptor.MethodInfo.Name);
 
+            var serializer = _serviceProvider.GetService<IContentSerializer>();
             using (var scope = _serviceProvider.CreateScope())
             {
                 var provider = scope.ServiceProvider;
@@ -42,7 +43,7 @@ namespace DotNetCore.CAP.Internal
                 var obj = ActivatorUtilities.GetServiceOrCreateInstance(provider, serviceType);
 
                 var jsonConent = _consumerContext.DeliverMessage.Content;
-                var message = Helper.FromJson<Message>(jsonConent);
+                var message =  serializer.DeSerialize<CapMessageDto>(jsonConent);
 
                 object result = null;
                 if (_executor.MethodParameters.Length > 0)
@@ -105,7 +106,7 @@ namespace DotNetCore.CAP.Internal
 
         private async Task SentCallbackMessage(string messageId, string topicName, object bodyObj)
         {
-            var callbackMessage = new Message
+            var callbackMessage = new CapMessageDto
             {
                 Id = messageId,
                 Content = bodyObj

src/DotNetCore.CAP/Internal/IContentSerializer.Json.cs

+using DotNetCore.CAP.Abstractions;
+using DotNetCore.CAP.Infrastructure;
+using DotNetCore.CAP.Models;
+
+namespace DotNetCore.CAP.Internal
+{
+    public class JsonContentSerializer : IContentSerializer
+    {
+        public T DeSerialize<T>(string messageObjStr) where T : CapMessageDto, new() {
+
+            return Helper.FromJson<T>(messageObjStr);
+        }
+
+        public string Serialize<T>(T messageObj) where T : CapMessageDto, new() {
+
+            return Helper.ToJson(messageObj);
+        }
+    }
+}