Try to fix transport healthy check bug. #503

Author: yang-xiaodong

src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClient.cs

@@ -109,9 +109,7 @@ public void Dispose()
             _consumerClient?.CloseAsync().Wait(1500);
         }
 
-        #region private methods
-
-        private async Task ConnectAsync()
+        public async Task ConnectAsync()
         {
             if (_consumerClient != null)
             {
@@ -157,6 +155,8 @@ private async Task ConnectAsync()
             }
         }
 
+        #region private methods
+
         private Task OnConsumerReceived(Message message, CancellationToken token)
         {
             var header = message.UserProperties.ToDictionary(x => x.Key, y => y.Value?.ToString());

src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClientFactory.cs

@@ -25,7 +25,9 @@ public IConsumerClient Create(string groupId)
             try
             {
                 var logger = _loggerFactory.CreateLogger(typeof(AzureServiceBusConsumerClient));
-                return new AzureServiceBusConsumerClient(logger, groupId, _asbOptions);
+                var client = new AzureServiceBusConsumerClient(logger, groupId, _asbOptions);
+                client.ConnectAsync().GetAwaiter().GetResult();
+                return client;
             }
             catch (System.Exception e)
             {

src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs

@@ -95,9 +95,7 @@ public void Dispose()
             _consumerClient?.Dispose();
         }
 
-        #region private methods
-
-        private void Connect()
+        public void Connect()
         {
             if (_consumerClient != null)
             {
@@ -134,7 +132,5 @@ private void ConsumerClient_OnConsumeError(IConsumer<string, byte[]> consumer, E
             };
             OnLog?.Invoke(null, logArgs);
         }
-
-        #endregion private methods
     }
 }

src/DotNetCore.CAP.Kafka/KafkaConsumerClientFactory.cs

@@ -19,7 +19,9 @@ public IConsumerClient Create(string groupId)
         {
             try
             {
-                return new KafkaConsumerClient(groupId, _kafkaOptions);
+                var client = new KafkaConsumerClient(groupId, _kafkaOptions);
+                client.Connect();
+                return client;
             }
             catch (System.Exception e)
             {

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs

@@ -95,9 +95,7 @@ public void Dispose()
             _connection?.Dispose();
         }
 
-        #region events
-
-        private void Connect()
+        public void Connect()
         {
             if (_connection != null)
             {
@@ -129,6 +127,8 @@ private void Connect()
             }
         }
 
+        #region events
+
         private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e)
         {
             var args = new LogMessageEventArgs

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClientFactory.cs

@@ -21,7 +21,9 @@ public IConsumerClient Create(string groupId)
         {
             try
             {
-                return new RabbitMQConsumerClient(groupId, _connectionChannelPool, _rabbitMQOptions);
+               var client = new RabbitMQConsumerClient(groupId, _connectionChannelPool, _rabbitMQOptions);
+               client.Connect();
+               return client;
             }
             catch (System.Exception e)
             {

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

@@ -235,6 +235,7 @@ private void WriteLog(object sender, LogMessageEventArgs logmsg)
                     _logger.LogWarning("RabbitMQ consumer unregistered. --> " + logmsg.Reason);
                     break;
                 case MqLogType.ConsumerShutdown:
+                    _isHealthy = false;
                     _logger.LogWarning("RabbitMQ consumer shutdown. --> " + logmsg.Reason);
                     break;
                 case MqLogType.ConsumeError: