feat: restore queue model and consumer on disconnection (#428)

Co-authored-by: t.sarmis <[email protected]>

Author: sarmis

bus/EasyCaching.Bus.RabbitMQ/DefaultRabbitMQBus.cs

@@ -150,20 +150,51 @@ public override void BaseSubscribe(string topic, Action<EasyCachingMessage> acti
                 queueName = _options.QueueName;
             }
 
-            Task.Factory.StartNew(() =>
+            Task.Factory.StartNew(
+                () => StartConsumer(queueName, topic),
+                TaskCreationOptions.LongRunning);
+        }
+
+
+        private void StartConsumer(string queueName, string topic)
+        {
+            var model = _subConnection.CreateModel();
+
+            model.ExchangeDeclare(_options.TopicExchangeName, ExchangeType.Topic, true, false, null);
+            model.QueueDeclare(queueName, false, false, true, null);
+            // bind the queue with the exchange.
+            model.QueueBind(queueName, _options.TopicExchangeName, topic);
+            var consumer = new EventingBasicConsumer(model);
+            consumer.Received += OnMessage;
+            consumer.Shutdown += (sender, e) =>
             {
-                var model = _subConnection.CreateModel();
-                model.ExchangeDeclare(_options.TopicExchangeName, ExchangeType.Topic, true, false, null);
-                model.QueueDeclare(queueName, false, false, true, null);
-                // bind the queue with the exchange.
-                model.QueueBind(queueName, _options.TopicExchangeName, topic);
-                var consumer = new EventingBasicConsumer(model);
-                consumer.Received += OnMessage;
-                consumer.Shutdown += OnConsumerShutdown;
-
-                model.BasicConsume(queueName, true, consumer);
-
-            }, TaskCreationOptions.LongRunning);
+                OnConsumerShutdown(sender, e);
+                OnConsumerError(queueName, topic, model);
+            };
+
+            consumer.ConsumerCancelled += (s, e) =>
+            {
+                OnConsumerError(queueName, topic, model);
+            };
+
+            model.BasicConsume(queueName, true, consumer);
+        }
+
+        private void OnConsumerError(string queueName, string topic, IModel model)
+        {
+            StartConsumer(queueName, topic);
+            BaseOnReconnect();
+            try
+            {
+                if (model?.IsOpen == true)
+                {
+                    model?.Dispose();
+                }
+            }
+            catch
+            {
+                // nothing to do
+            }
         }
 
         /// <summary>

src/EasyCaching.Core/Bus/IEasyCachingSubscriber.cs

@@ -12,6 +12,7 @@ public interface IEasyCachingSubscriber
         /// </summary>
         /// <param name="topic">Topic.</param>
         /// <param name="action">Action.</param>
-        void Subscribe(string topic, Action<EasyCachingMessage> action);
+        /// <param name="reconnectAction"> Reconnect Action.</param>
+        void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null);
     }
 }

src/EasyCaching.Core/Bus/NullEasyCachingBus.cs

@@ -26,7 +26,7 @@ public class NullEasyCachingBus : IEasyCachingBus
         /// <see cref="T:EasyCaching.Core.Bus.NullEasyCachingBus"/> so the garbage collector can reclaim the memory that
         /// the <see cref="T:EasyCaching.Core.Bus.NullEasyCachingBus"/> was occupying.</remarks>
         public void Dispose() { }
-               
+
         /// <summary>
         /// Publish the specified topic and message.
         /// </summary>
@@ -54,7 +54,8 @@ public void Publish(string topic, EasyCachingMessage message)
         /// </summary>
         /// <param name="topic">Topic.</param>
         /// <param name="action">Action.</param>
-        public void Subscribe(string topic, Action<EasyCachingMessage> action)
+        /// <param name="reconnectAction">Reconnect Action.</param>
+        public void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null)
         {
 
         }

src/EasyCaching.Core/EasyCachingAbstractBus.cs

@@ -18,6 +18,8 @@ public abstract class EasyCachingAbstractBus : IEasyCachingBus
 
         protected Action<EasyCachingMessage> _handler;
 
+        protected Action _reconnectHandler;
+        
         protected string BusName { get; set; }
 
         public string Name => this.BusName;
@@ -74,9 +76,10 @@ public void Publish(string topic, EasyCachingMessage message)
             }
         }
 
-        public void Subscribe(string topic, Action<EasyCachingMessage> action)
+        public void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction)
         {
             _handler = action;
+            _reconnectHandler = reconnectAction;
             BaseSubscribe(topic, action);
         }
 
@@ -105,5 +108,10 @@ public virtual void BaseOnMessage(EasyCachingMessage message)
                 }
             }
         }
+
+        public virtual void BaseOnReconnect()
+        {
+            _reconnectHandler?.Invoke();
+        }
     }
 }

src/EasyCaching.HybridCache/Configurations/HybridCachingOptions.cs

@@ -43,5 +43,13 @@ public class HybridCachingOptions
         /// When sending message failed, we will retry some times, default is 3 times.
         /// </remarks>
         public int BusRetryCount { get; set; } = 3;
+
+        /// <summary>
+        /// Flush the local cache on bus disconnection/reconnection
+        /// </summary>
+        /// <remarks>
+        /// Flushing the local cache will avoid using stale data but may cause app jitters until the local cache get's re-populated.
+        /// </remarks>
+        public bool FlushLocalCacheOnBusReconnection { get; set; } = false;
     }
 }

src/EasyCaching.HybridCache/HybridCachingProvider.cs

@@ -93,7 +93,7 @@ string name
             else this._distributedCache = distributed;
 
             this._bus = bus ?? NullEasyCachingBus.Instance;
-            this._bus.Subscribe(_options.TopicName, OnMessage);
+            this._bus.Subscribe(_options.TopicName, OnMessage, OnReconnect);
 
             this._cacheId = Guid.NewGuid().ToString("N");
 
@@ -159,6 +159,21 @@ private void OnMessage(EasyCachingMessage message)
             }
         }
 
+        /// <summary>
+        /// On reconnect (flushes local memory as it could be stale).
+        /// </summary>
+
+        private void OnReconnect()
+        {
+            if (!_options.FlushLocalCacheOnBusReconnection)
+            {
+                return;
+            }
+
+            LogMessage("Flushing local cache due to bus reconnection");
+            _localCache.Flush();
+        }
+
         /// <summary>
         /// Exists the specified cacheKey.
         /// </summary>

test/EasyCaching.UnitTests/Fake/FakeBus.cs

@@ -19,7 +19,7 @@ public void Publish(string topic, EasyCachingMessage message)
             return Task.CompletedTask;
         }
 
-        public void Subscribe(string topic, Action<EasyCachingMessage> action)
+        public void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null)
         {
 
         }