AC-824 task manager rabbit connection resiliency issues

src/Plugins/RabbitMQ/Factory/CreateChannelArguments.cs

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2022 MONAI Consortium
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using Ardalis.GuardClauses;
+
+namespace Monai.Deploy.Messaging.RabbitMQ
+{
+    public class CreateChannelArguments
+    {
+        public CreateChannelArguments(
+            string hostName,
+            string password,
+            string username,
+            string virtualHost,
+            string useSSL,
+            string portNumber)
+        {
+            Guard.Against.NullOrWhiteSpace(hostName);
+            Guard.Against.NullOrWhiteSpace(password);
+            Guard.Against.NullOrWhiteSpace(username);
+            Guard.Against.NullOrWhiteSpace(virtualHost);
+            Guard.Against.NullOrWhiteSpace(useSSL);
+            Guard.Against.NullOrWhiteSpace(portNumber);
+
+            HostName = hostName;
+            Password = password;
+            Username = username;
+            VirtualHost = virtualHost;
+            UseSSL = useSSL;
+            PortNumber = portNumber;
+        }
+
+        public string HostName { get; set; }
+
+        public string Password { get; set; }
+
+        public string Username { get; set; }
+
+        public string VirtualHost { get; set; }
+
+        public string UseSSL { get; set; }
+
+        public string PortNumber { get; set; }
+    }
+}

src/Plugins/RabbitMQ/Factory/IRabbitMQConnectionFactory.cs

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2022 MONAI Consortium
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using RabbitMQ.Client;
+
+namespace Monai.Deploy.Messaging.RabbitMQ
+{
+    public interface IRabbitMQConnectionFactory
+    {
+        /// <summary>
+        /// Creates a new channel for RabbitMQ client.
+        /// The connection factory maintains a single connection to the specified
+        /// <c>hostName</c>, <c>username</c>, <c>password</c>, and <c>virtualHost</c> combination.
+        /// </summary>
+        /// <param name="hostName">Host name</param>
+        /// <param name="username">User name</param>
+        /// <param name="password">Password</param>
+        /// <param name="virtualHost">Virtual host</param>
+        /// <param name="useSSL">Encrypt communication</param>
+        /// <param name="portNumber">Port Number</param>
+        /// <returns>Instance of <see cref="IModel"/>.</returns>
+        IModel CreateChannel(string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);
+
+        /// <summary>
+        /// Creates a new channel for RabbitMQ client.
+        /// The connection factory maintains a single connection to the specified
+        /// <c>hostName</c>, <c>username</c>, <c>password</c>, and <c>virtualHost</c> combination.
+        /// </summary>
+        /// <param name="virtualHost">Virtual host</param>
+        /// <returns>Instance of <see cref="IModel"/>.</returns>
+        IModel CreateChannel(CreateChannelArguments args);
+    }
+}

src/Plugins/RabbitMQ/RabbitMqConnectionFactory.cs → src/Plugins/RabbitMQ/Factory/RabbitMqConnectionFactory.cs

@@ -16,33 +16,18 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Globalization;
 using System.Linq;
 using System.Net.Security;
 using System.Security.Cryptography;
 using System.Text;
 using Ardalis.GuardClauses;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 
 namespace Monai.Deploy.Messaging.RabbitMQ
 {
-    public interface IRabbitMQConnectionFactory
-    {
-        /// <summary>
-        /// Creates a new channel for RabbitMQ client.
-        /// THe connection factory maintains a single connection to the specified
-        /// <c>hostName</c>, <c>username</c>, <c>password</c>, and <c>virtualHost</c> combination.
-        /// </summary>
-        /// <param name="hostName">Host name</param>
-        /// <param name="username">User name</param>
-        /// <param name="password">Password</param>
-        /// <param name="virtualHost">Virtual host</param>
-        /// <param name="useSSL">Encrypt communication</param>
-        /// <param name="portNumber">Port Number</param>
-        /// <returns>Instance of <see cref="IModel"/>.</returns>
-        IModel CreateChannel(string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);
-    }
-
     public class RabbitMQConnectionFactory : IRabbitMQConnectionFactory, IDisposable
     {
         private readonly ConcurrentDictionary<string, Lazy<ConnectionFactory>> _connectionFactoriess;
@@ -57,25 +42,27 @@ public RabbitMQConnectionFactory(ILogger<RabbitMQConnectionFactory> logger)
             _connections = new ConcurrentDictionary<string, Lazy<IConnection>>();
         }
 
+        public IModel CreateChannel(CreateChannelArguments args) =>
+            CreateChannel(args.HostName, args.Username, args.Password, args.VirtualHost,
+                          args.UseSSL, args.PortNumber);
+
         public IModel CreateChannel(string hostName, string username, string password, string virtualHost, string useSSL, string portNumber)
         {
-            Guard.Against.NullOrWhiteSpace(hostName, nameof(hostName));
-            Guard.Against.NullOrWhiteSpace(username, nameof(username));
-            Guard.Against.NullOrWhiteSpace(password, nameof(password));
-            Guard.Against.NullOrWhiteSpace(virtualHost, nameof(virtualHost));
+            Guard.Against.NullOrWhiteSpace(hostName);
+            Guard.Against.NullOrWhiteSpace(username);
+            Guard.Against.NullOrWhiteSpace(password);
+            Guard.Against.NullOrWhiteSpace(virtualHost);
 
             var key = $"{hostName}{username}{HashPassword(password)}{virtualHost}";
 
             var connection = _connections.AddOrUpdate(key,
-                x =>
-                {
-                    return CreatConnection(hostName, username, password, virtualHost, key, useSSL, portNumber);
-                },
+                x => CreatConnection(hostName, username, password, virtualHost, key, useSSL, portNumber),
                 (updateKey, updateConnection) =>
                 {
                     // If connection to RMQ is lost and:
                     //   - RMQ service returns before calling the next line, then IsOpen returns false
-                    //   - a call is made before RMQ returns, then a new connection is made with error with IsValueFaulted = true && IsValueCreated = false
+                    //   - a call is made before RMQ returns, then a new connection
+                    //      is made with error with IsValueFaulted = true && IsValueCreated = false
                     if (updateConnection.IsValueCreated && updateConnection.Value.IsOpen)
                     {
                         return updateConnection;
@@ -86,7 +73,40 @@ public IModel CreateChannel(string hostName, string username, string password, s
                     }
                 });
 
-            return connection.Value.CreateModel();
+            var model = connection.Value.CreateModel();
+
+            var argsObj = new CreateChannelArguments(hostName, password, username, virtualHost, useSSL, portNumber);
+
+            model.CallbackException += (connection, args) => OnException(args, key, argsObj);
+            model.ModelShutdown += (connection, args) => OnShutdown(args, key, argsObj);
+
+            return model;
+        }
+
+        private void OnShutdown(ShutdownEventArgs args, string key, CreateChannelArguments createChannelArguments)
+        {
+            _logger.ConnectionShutdown(args.ReplyText);
+            _connections.TryRemove(key, out var value);
+
+            if (value is not null)
+            {
+                value?.Value.Dispose();
+            }
+
+            CreateChannel(createChannelArguments);
+        }
+
+        private void OnException(CallbackExceptionEventArgs args, string key, CreateChannelArguments createChannelArguments)
+        {
+            _logger.ConnectionException(args.Exception);
+            _connections.TryRemove(key, out var value);
+
+            if (value is not null)
+            {
+                value?.Value.Dispose();
+            }
+
+            CreateChannel(createChannelArguments);
         }
 
         private Lazy<IConnection> CreatConnection(string hostName, string username, string password, string virtualHost, string key, string useSSL, string portNumber)
@@ -96,7 +116,7 @@ private Lazy<IConnection> CreatConnection(string hostName, string username, stri
                 sslEnabled = false;
             }
 
-            if (!Int32.TryParse(portNumber, out var port))
+            if (!int.TryParse(portNumber, out var port))
             {
                 port = sslEnabled ? 5671 : 5672; // 5671 is default port for SSL/TLS , 5672 is default port for PLAIN.
             }
@@ -115,18 +135,19 @@ private Lazy<IConnection> CreatConnection(string hostName, string username, stri
                 Password = password,
                 VirtualHost = virtualHost,
                 Ssl = sslOptions,
-                Port = port
+                Port = port,
+                RequestedHeartbeat = TimeSpan.FromSeconds(10),
             }));
 
-            return new Lazy<IConnection>(() => connectionFactory.Value.CreateConnection());
+            return new Lazy<IConnection>(connectionFactory.Value.CreateConnection);
         }
 
-        private object HashPassword(string password)
+        private static object HashPassword(string password)
         {
-            Guard.Against.NullOrWhiteSpace(password, nameof(password));
+            Guard.Against.NullOrWhiteSpace(password);
             var sha256 = SHA256.Create();
             var hash = sha256.ComputeHash(Encoding.UTF8.GetBytes(password));
-            return hash.Select(x => x.ToString("x2"));
+            return hash.Select(x => x.ToString("x2", CultureInfo.InvariantCulture));
         }
 
         protected virtual void Dispose(bool disposing)

src/Plugins/RabbitMQ/Logger.cs

@@ -61,5 +61,11 @@ public static partial class Logger
 
         [LoggerMessage(EventId = 10012, Level = LogLevel.Error, Message = "Health check failure.")]
         public static partial void HealthCheckError(this ILogger logger, Exception ex);
+
+        [LoggerMessage(EventId = 10013, Level = LogLevel.Error, Message = "RabbitMQ connection shutdown ({replyText}) attempting to reconnect.")]
+        public static partial void ConnectionShutdown(this ILogger logger, string replyText);
+
+        [LoggerMessage(EventId = 10014, Level = LogLevel.Error, Message = "RabbitMQ connection exception attempting to reconnect.")]
+        public static partial void ConnectionException(this ILogger logger, Exception ex);
     }
 }

src/Plugins/RabbitMQ/RabbitMqMessageSubscriberService.cs → src/Plugins/RabbitMQ/Subscriber/RabbitMqMessageSubscriberService.cs

@@ -53,7 +53,7 @@ public RabbitMQMessageSubscriberService(IOptions<MessageBrokerServiceConfigurati
                                                 ILogger<RabbitMQMessageSubscriberService> logger,
                                                 IRabbitMQConnectionFactory rabbitMqConnectionFactory)
         {
-            Guard.Against.Null(options, nameof(options));
+            Guard.Against.Null(options);
 
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
@@ -74,7 +74,7 @@ public RabbitMQMessageSubscriberService(IOptions<MessageBrokerServiceConfigurati
             }
             else
             {
-                _useSSL = String.Empty;
+                _useSSL = string.Empty;
             }
 
             if (configuration.SubscriberSettings.ContainsKey(ConfigurationKeys.Port))
@@ -83,7 +83,7 @@ public RabbitMQMessageSubscriberService(IOptions<MessageBrokerServiceConfigurati
             }
             else
             {
-                _portNumber = String.Empty;
+                _portNumber = string.Empty;
             }
 
             _logger.ConnectingToRabbitMQ(Name, _endpoint, _virtualHost);
@@ -95,7 +95,7 @@ public RabbitMQMessageSubscriberService(IOptions<MessageBrokerServiceConfigurati
 
         internal static void ValidateConfiguration(Dictionary<string, string> configuration)
         {
-            Guard.Against.Null(configuration, nameof(configuration));
+            Guard.Against.Null(configuration);
 
             foreach (var key in ConfigurationKeys.SubscriberRequiredKeys)
             {
@@ -126,8 +126,8 @@ public void Subscribe(string topic, string queue, Action<MessageReceivedEventArg
 
         public void Subscribe(string[] topics, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0)
         {
-            Guard.Against.Null(topics, nameof(topics));
-            Guard.Against.Null(messageReceivedCallback, nameof(messageReceivedCallback));
+            Guard.Against.Null(topics);
+            Guard.Against.Null(messageReceivedCallback);
 
             var arguments = new Dictionary<string, object>()
             {
@@ -188,8 +188,8 @@ public void SubscribeAsync(string topic, string queue, Func<MessageReceivedEvent
 
         public void SubscribeAsync(string[] topics, string queue, Func<MessageReceivedEventArgs, Task> messageReceivedCallback, ushort prefetchCount = 0)
         {
-            Guard.Against.Null(topics, nameof(topics));
-            Guard.Against.Null(messageReceivedCallback, nameof(messageReceivedCallback));
+            Guard.Against.Null(topics);
+            Guard.Against.Null(messageReceivedCallback);
 
             var arguments = new Dictionary<string, object>()
             {
@@ -250,7 +250,7 @@ public void SubscribeAsync(string[] topics, string queue, Func<MessageReceivedEv
 
         public void Acknowledge(MessageBase message)
         {
-            Guard.Against.Null(message, nameof(message));
+            Guard.Against.Null(message);
 
             _logger.SendingAcknowledgement(message.MessageId);
             _channel.BasicAck(ulong.Parse(message.DeliveryTag, CultureInfo.InvariantCulture), multiple: false);
@@ -281,7 +281,7 @@ public async Task RequeueWithDelay(MessageBase message)
 
         public void Reject(MessageBase message, bool requeue = true)
         {
-            Guard.Against.Null(message, nameof(message));
+            Guard.Against.Null(message);
 
             _logger.SendingNAcknowledgement(message.MessageId);
             _channel.BasicNack(ulong.Parse(message.DeliveryTag, CultureInfo.InvariantCulture), multiple: false, requeue: requeue);
@@ -312,8 +312,8 @@ public void Dispose()
 
         private void BindToRoutingKeys(string[] topics, string queue, string deadLetterQueue = "")
         {
-            Guard.Against.Null(topics, nameof(topics));
-            Guard.Against.NullOrWhiteSpace(queue, nameof(queue));
+            Guard.Against.Null(topics);
+            Guard.Against.NullOrWhiteSpace(queue);
 
             foreach (var topic in topics)
             {
@@ -331,17 +331,17 @@ private void BindToRoutingKeys(string[] topics, string queue, string deadLetterQ
 
         private static MessageReceivedEventArgs CreateMessage(string topic, BasicDeliverEventArgs eventArgs)
         {
-            Guard.Against.NullOrWhiteSpace(topic, nameof(topic));
-            Guard.Against.Null(eventArgs, nameof(eventArgs));
-
-            Guard.Against.Null(eventArgs.Body, nameof(eventArgs.Body));
-            Guard.Against.Null(eventArgs.BasicProperties, nameof(eventArgs.BasicProperties));
-            Guard.Against.Null(eventArgs.BasicProperties.MessageId, nameof(eventArgs.BasicProperties.MessageId));
-            Guard.Against.Null(eventArgs.BasicProperties.AppId, nameof(eventArgs.BasicProperties.AppId));
-            Guard.Against.Null(eventArgs.BasicProperties.ContentType, nameof(eventArgs.BasicProperties.ContentType));
-            Guard.Against.Null(eventArgs.BasicProperties.CorrelationId, nameof(eventArgs.BasicProperties.CorrelationId));
-            Guard.Against.Null(eventArgs.BasicProperties.Timestamp, nameof(eventArgs.BasicProperties.Timestamp));
-            Guard.Against.Null(eventArgs.DeliveryTag, nameof(eventArgs.DeliveryTag));
+            Guard.Against.NullOrWhiteSpace(topic);
+            Guard.Against.Null(eventArgs);
+
+            Guard.Against.Null(eventArgs.Body);
+            Guard.Against.Null(eventArgs.BasicProperties);
+            Guard.Against.Null(eventArgs.BasicProperties.MessageId);
+            Guard.Against.Null(eventArgs.BasicProperties.AppId);
+            Guard.Against.Null(eventArgs.BasicProperties.ContentType);
+            Guard.Against.Null(eventArgs.BasicProperties.CorrelationId);
+            Guard.Against.Null(eventArgs.BasicProperties.Timestamp);
+            Guard.Against.Null(eventArgs.DeliveryTag);
 
             return new MessageReceivedEventArgs(
                 new Message(