refactor MQ log message

Author: yang-xiaodong

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs

@@ -2,7 +2,6 @@
 using System.Collections.Generic;
 using System.Text;
 using System.Threading;
-using System.Threading.Tasks;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 
@@ -32,7 +31,7 @@ public RabbitMQConsumerClient(string queueName,
 
         public event EventHandler<MessageContext> OnMessageReceived;
 
-        public event EventHandler<string> OnError;
+        public event EventHandler<LogMessageEventArgs> OnLog;
 
         public void Subscribe(IEnumerable<string> topics)
         {
@@ -47,9 +46,17 @@ public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
             var consumer = new EventingBasicConsumer(_channel);
             consumer.Received += OnConsumerReceived;
             consumer.Shutdown += OnConsumerShutdown;
+            consumer.Registered += OnConsumerRegistered;
+            consumer.Unregistered += OnConsumerUnregistered;
+            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;
+
             _channel.BasicConsume(_queueName, false, consumer);
+
             while (true)
-                Task.Delay(timeout, cancellationToken).GetAwaiter().GetResult();
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                cancellationToken.WaitHandle.WaitOne(timeout);
+            }
         }
 
         public void Commit()
@@ -69,9 +76,9 @@ public void Dispose()
 
         private void InitClient()
         {
-            var connection = _connectionChannelPool.GetConnection();
+            var _connection = _connectionChannelPool.GetConnection();
 
-            _channel = connection.CreateModel();
+            _channel = _connection.CreateModel();
 
             _channel.ExchangeDeclare(
                 _exchageName,
@@ -84,6 +91,38 @@ private void InitClient()
             _channel.QueueDeclare(_queueName, true, false, false, arguments);
         }
 
+        #region events
+
+        private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e)
+        {
+            var args = new LogMessageEventArgs
+            {
+                LogType =  MqLogType.ConsumerCancelled,
+                Reason = e.ConsumerTag
+            };
+            OnLog?.Invoke(sender, args);
+        }
+
+        private void OnConsumerUnregistered(object sender, ConsumerEventArgs e)
+        {
+            var args = new LogMessageEventArgs
+            {
+                LogType = MqLogType.ConsumerUnregistered,
+                Reason = e.ConsumerTag
+            };
+            OnLog?.Invoke(sender, args);
+        }
+
+        private void OnConsumerRegistered(object sender, ConsumerEventArgs e)
+        {
+            var args = new LogMessageEventArgs
+            {
+                LogType = MqLogType.ConsumerRegistered,
+                Reason = e.ConsumerTag
+            };
+            OnLog?.Invoke(sender, args);
+        }
+
         private void OnConsumerReceived(object sender, BasicDeliverEventArgs e)
         {
             _deliveryTag = e.DeliveryTag;
@@ -98,7 +137,14 @@ private void OnConsumerReceived(object sender, BasicDeliverEventArgs e)
 
         private void OnConsumerShutdown(object sender, ShutdownEventArgs e)
         {
-            OnError?.Invoke(sender, e.Cause?.ToString());
+            var args = new LogMessageEventArgs
+            {
+                LogType = MqLogType.ConsumerShutdown,
+                Reason = e.ToString()
+            };
+            OnLog?.Invoke(sender, args);
         }
+
+        #endregion
     }
 }