Fix message consumer bugs

samples/Sample.RabbitMQ.MySql/Controllers/ValuesController.cs

@@ -71,7 +71,7 @@ namespace Sample.RabbitMQ.MySql.Controllers
         [CapSubscribe("sample.rabbitmq.mysql")]
         public void Subscriber(DateTime time)
         {
-            //Console.WriteLine($@"{DateTime.Now}, Subscriber invoked, Sent time:{time}");
+            Console.WriteLine($@"{DateTime.Now}, Subscriber invoked, Sent time:{time}");
         }
     }
 }

src/DotNetCore.CAP.MySql/MySqlDataStorage.cs

@@ -103,7 +103,7 @@ namespace DotNetCore.CAP.MySql
             return message;
         }
 
-        public async Task<MediumMessage> StoreMessageAsync(string name, string group, Message content, CancellationToken cancellationToken = default)
+        public Task<MediumMessage> StoreMessageAsync(string name, string group, Message content, CancellationToken cancellationToken = default)
         {
             var sql = $@"INSERT INTO `{_options.Value.TableNamePrefix}.received`(`Id`,`Version`,`Name`,`Group`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`) VALUES(@Id,'{_options.Value.Version}',@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
 
@@ -125,14 +125,15 @@ namespace DotNetCore.CAP.MySql
                 Retries = message.Retries,
                 Added = message.Added,
                 ExpiresAt = message.ExpiresAt,
-                StatusName = StatusName.Scheduled
+                StatusName = nameof(StatusName.Scheduled)
             };
 
             using (var connection = new MySqlConnection(_options.Value.ConnectionString))
             {
-                await connection.ExecuteAsync(sql, po);
+                connection.Execute(sql, po);
             }
-            return message;
+
+            return Task.FromResult(message);
         }
 
         public async Task<int> DeleteExpiresAsync(string table, DateTime timeout, int batchCount = 1000, CancellationToken token = default)

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Text;
 using System.Threading;
 using DotNetCore.CAP.Messages;
 using Microsoft.Extensions.Options;
@@ -163,11 +164,14 @@ namespace DotNetCore.CAP.RabbitMQ
         {
             _deliveryTag = e.DeliveryTag;
 
-            var header = e.BasicProperties.Headers
-                .ToDictionary(x => x.Key, x => x.Value.ToString());
-            header.Add(Headers.Group, _queueName);
+            var headers = new Dictionary<string, string>();
+            foreach (var header in e.BasicProperties.Headers)
+            {
+                headers.Add(header.Key, header.Value == null ? null : Encoding.UTF8.GetString((byte[])header.Value));
+            }
+            headers.Add(Headers.Group, _queueName);
 
-            var message = new TransportMessage(header, e.Body);
+            var message = new TransportMessage(headers, e.Body);
 
             OnMessageReceived?.Invoke(sender, message);
         }

src/DotNetCore.CAP/CAP.ServiceCollectionExtensions.cs

@@ -36,6 +36,8 @@ namespace Microsoft.Extensions.DependencyInjection
 
             services.TryAddSingleton<CapMarkerService>();
 
+            services.TryAddSingleton<ICapPublisher, CapPublisher>();
+
             //Serializer and model binder
             services.TryAddSingleton<IContentSerializer, JsonContentSerializer>();
             services.TryAddSingleton<IMessagePacker, DefaultMessagePacker>();
@@ -55,6 +57,7 @@ namespace Microsoft.Extensions.DependencyInjection
             //Queue's message processor
             services.TryAddSingleton<MessageNeedToRetryProcessor>();
             services.TryAddSingleton<TransportCheckProcessor>();
+            services.TryAddSingleton<CollectorProcessor>();
 
             //Sender and Executors   
             services.TryAddSingleton<IMessageSender, MessageSender>();

src/DotNetCore.CAP/IConsumerRegister.Default.cs

@@ -160,6 +160,7 @@ namespace DotNetCore.CAP
                     var stopwatch = Stopwatch.StartNew();
 
                     var message = await _serializer.DeserializeAsync(messageContext);
+
                     var mediumMessage = await _storage.StoreMessageAsync(message.GetName(), message.GetGroup(), message);
 
                     client.Commit();

src/DotNetCore.CAP/ISubscribeExecutor.Default.cs

@@ -14,14 +14,15 @@ using DotNetCore.CAP.Persistence;
 using DotNetCore.CAP.Processor;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace DotNetCore.CAP
 {
     internal class DefaultSubscriberExecutor : ISubscriberExecutor
     {
-        private readonly ICapPublisher _sender;
         private readonly IDataStorage _dataStorage;
         private readonly ILogger _logger;
+        private readonly IServiceProvider _provider;
         private readonly CapOptions _options;
         private readonly MethodMatcherCache _selector;
 
@@ -33,18 +34,17 @@ namespace DotNetCore.CAP
         public DefaultSubscriberExecutor(
             ILogger<DefaultSubscriberExecutor> logger,
             IOptions<CapOptions> options,
-            IConsumerInvokerFactory consumerInvokerFactory,
-            ICapPublisher sender,
-            IDataStorage dataStorage,
+            IServiceProvider provider,
             MethodMatcherCache selector)
         {
             _selector = selector;
-            _sender = sender;
-            _options = options.Value;
-            _dataStorage = dataStorage;
-            _logger = logger;
 
-            Invoker = consumerInvokerFactory.CreateInvoker();
+            _provider = provider;
+            _logger = logger;
+            _options = options.Value;
+           
+            _dataStorage = _provider.GetService<IDataStorage>();
+            Invoker = _provider.GetService<IConsumerInvokerFactory>().CreateInvoker();
         }
 
         private IConsumerInvoker Invoker { get; }
@@ -190,7 +190,7 @@ namespace DotNetCore.CAP
                         [Headers.CorrelationSequence] = (message.Origin.GetCorrelationSequence() + 1).ToString()
                     };
 
-                    await _sender.PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken);
+                    await _provider.GetService<ICapPublisher>().PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken);
                 }
             }
             catch (OperationCanceledException)