refactor

src/DotNetCore.CAP/ICallbackPublisher.cs

-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
+using System.Threading.Tasks;
+using DotNetCore.CAP.Models;
 
 namespace DotNetCore.CAP
 {
     public interface ICallbackPublisher
     {
-        Task PublishAsync(string name, object obj);
+        Task PublishAsync(CapPublishedMessage obj);
     }
 }

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

@@ -23,16 +23,14 @@ namespace DotNetCore.CAP.Internal
         {
             _modelBinderFactory = modelBinderFactory;
             _serviceProvider = serviceProvider;
-            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+            _logger = logger;
+            _consumerContext = consumerContext;
 
-            _consumerContext = consumerContext ?? throw new ArgumentNullException(nameof(consumerContext));
             _executor = ObjectMethodExecutor.Create(_consumerContext.ConsumerDescriptor.MethodInfo,
                 _consumerContext.ConsumerDescriptor.ImplTypeInfo);
         }
 
         public async Task InvokeAsync()
-        {
-            using (_logger.BeginScope("consumer invoker begin"))
         {
             _logger.LogDebug("Executing consumer Topic: {0}", _consumerContext.ConsumerDescriptor.MethodInfo.Name);
 
@@ -40,47 +38,86 @@ namespace DotNetCore.CAP.Internal
                     _consumerContext.ConsumerDescriptor.ImplTypeInfo.AsType());
 
             var jsonConent = _consumerContext.DeliverMessage.Content;
-
             var message = Helper.FromJson<Message>(jsonConent);
 
-                object returnObj = null;
+            object result = null;
             if (_executor.MethodParameters.Length > 0)
+            {
+                result = await ExecuteWithParameterAsync(obj, message.Content.ToString());
+            }
+            else
+            {
+                result = await ExecuteAsync(obj);
+            }
+
+            if (!string.IsNullOrEmpty(message.CallbackName))
+            {
+                await SentCallbackMessage(message.Id, message.CallbackName, result);
+            }
+        }
+
+        private async Task<object> ExecuteAsync(object @class)
+        {
+            if (_executor.IsMethodAsync)
+            {
+                return await _executor.ExecuteAsync(@class);
+            }
+            else
+            {
+                return _executor.Execute(@class);
+            }
+        }
+
+        private async Task<object> ExecuteWithParameterAsync(object @class, string parameterString)
         {
             var firstParameter = _executor.MethodParameters[0];
             try
             {
                 var binder = _modelBinderFactory.CreateBinder(firstParameter);
-                        var result = await binder.BindModelAsync(message.Content.ToString());
-                        if (result.IsSuccess)
+                var bindResult = await binder.BindModelAsync(parameterString);
+                if (bindResult.IsSuccess)
                 {
-                            returnObj = _executor.Execute(obj, result.Model);
+                    if (_executor.IsMethodAsync)
+                    {
+                        return await _executor.ExecuteAsync(@class, bindResult.Model);
                     }
                     else
                     {
-                            _logger.LogWarning($"Parameters:{firstParameter.Name} bind failed! the content is:" + jsonConent);
+                        return _executor.Execute(@class, bindResult.Model);
                     }
                 }
-                    catch (FormatException ex)
+                else
                 {
-                        _logger.ModelBinderFormattingException(_executor.MethodInfo?.Name, firstParameter.Name, jsonConent, ex);
+                    throw new MethodBindException($"Parameters:{firstParameter.Name} bind failed! ParameterString is: {parameterString} ");
                 }
             }
-                else
+            catch (FormatException ex)
             {
-                    returnObj = _executor.Execute(obj);
+                _logger.ModelBinderFormattingException(_executor.MethodInfo?.Name, firstParameter.Name, parameterString, ex);
+                return null;
+            }
         }
 
-                //TODO :refactor
-                if (returnObj != null && !string.IsNullOrEmpty(message.CallbackName))
+        private async Task SentCallbackMessage(string messageId, string topicName, object bodyObj)
         {
-                    var publisher = _serviceProvider.GetRequiredService<ICallbackPublisher>();
-                    var callbackMessage = new Message(returnObj)
+            var callbackMessage = new Message
             {
-                        Id = message.Id,
-                        Timestamp = DateTime.Now
+                Id = messageId,
+                Content = bodyObj
             };
-                    await publisher.PublishAsync(message.CallbackName, callbackMessage);
-                }
+
+            using (var scope = _serviceProvider.CreateScope())
+            {
+                var provider = scope.ServiceProvider;
+                var publisher = provider.GetRequiredService<ICallbackPublisher>();
+
+                var publishedMessage = new CapPublishedMessage
+                {
+                    Name = topicName,
+                    Content = Helper.ToJson(callbackMessage),
+                    StatusName = StatusName.Scheduled
+                };
+                await publisher.PublishAsync(publishedMessage);
             }
         }
     }