Using sync version of subscribe in HybridCachingProvider (dotnetcore#455

) * fix: using sync version of subscribe in HybridCachingProvider(dotnetcore#421) * feat: enable SubscribeAsync
Memoyu · Feb 9, 2023 · 5cf3e25 · 5cf3e25
1 parent 3e321cc
commit 5cf3e25
Show file tree

Hide file tree

Showing 10 changed files with 130 additions and 5 deletions.
diff --git a/bus/EasyCaching.Bus.CSRedis/DefaultCSRedisBus.cs b/bus/EasyCaching.Bus.CSRedis/DefaultCSRedisBus.cs
@@ -64,6 +64,21 @@ public override void BaseSubscribe(string topic, Action<EasyCachingMessage> acti
             );
         }
 
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public override Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            _client.Subscribe(
+                (topic, msg => OnMessage(msg.Body))
+            );
+            return Task.CompletedTask;
+        }
+
+
         /// <summary>
         /// Ons the message.
         /// </summary>

diff --git a/bus/EasyCaching.Bus.ConfluentKafka/DefaultConfluentKafkaBus.cs b/bus/EasyCaching.Bus.ConfluentKafka/DefaultConfluentKafkaBus.cs
@@ -12,7 +12,7 @@
     using Microsoft.Extensions.Options;
 
     public class DefaultConfluentKafkaBus : EasyCachingAbstractBus
-    {     
+    {
 
 
         /// <summary>
@@ -77,7 +77,7 @@ public override void BasePublish(string topic, EasyCachingMessage message)
         /// <param name="cancellationToken">Cancellation token.</param>
         public override async Task BasePublishAsync(string topic, EasyCachingMessage message, CancellationToken cancellationToken = default(CancellationToken))
         {
-            var msg =  _serializer.Serialize(message);
+            var msg = _serializer.Serialize(message);
 
             await _producer.ProduceAsync(topic, new Message<Null, byte[]> { Value = msg });
         }
@@ -89,7 +89,27 @@ public override void BasePublish(string topic, EasyCachingMessage message)
         /// <param name="action">Action.</param>
         public override void BaseSubscribe(string topic, Action<EasyCachingMessage> action)
         {
-            Task.Factory.StartNew(() =>
+            _ = StartConsumer(topic);
+        }
+
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public override async Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            await StartConsumer(topic);
+        }
+
+        /// <summary>
+        /// Ons the consumer.
+        /// </summary>
+        /// <param name="topic">Topic</param>
+        private Task StartConsumer(string topic)
+        {
+            return Task.Factory.StartNew(() =>
             {
                 for (int i = 0; i < this._kafkaBusOptions.ConsumerCount; i++)
                 {

diff --git a/bus/EasyCaching.Bus.RabbitMQ/DefaultRabbitMQBus.cs b/bus/EasyCaching.Bus.RabbitMQ/DefaultRabbitMQBus.cs
@@ -65,7 +65,7 @@ IPooledObjectPolicy<IModel> _objectPolicy
                 RequestedConnectionTimeout = System.TimeSpan.FromMilliseconds(_options.RequestedConnectionTimeout),
                 SocketReadTimeout = System.TimeSpan.FromMilliseconds(_options.SocketReadTimeout),
                 SocketWriteTimeout = System.TimeSpan.FromMilliseconds(_options.SocketWriteTimeout),
-                ClientProvidedName = _options.ClientProvidedName
+                ClientProvidedName = _options.ClientProvidedName,
             };
 
             _subConnection = factory.CreateConnection();
@@ -156,6 +156,28 @@ public override void BaseSubscribe(string topic, Action<EasyCachingMessage> acti
         }
 
 
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public override Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var queueName = string.Empty;
+            if (string.IsNullOrWhiteSpace(_options.QueueName))
+            {
+                queueName = $"rmq.queue.undurable.easycaching.subscriber.{_busId}";
+            }
+            else
+            {
+                queueName = _options.QueueName;
+            }
+
+            StartConsumer(queueName, topic);
+            return Task.CompletedTask;
+        }
+
         private void StartConsumer(string queueName, string topic)
         {
             var model = _subConnection.CreateModel();

diff --git a/bus/EasyCaching.Bus.Redis/DefaultRedisBus.cs b/bus/EasyCaching.Bus.Redis/DefaultRedisBus.cs
@@ -104,5 +104,16 @@ public override void BaseSubscribe(string topic, Action<EasyCachingMessage> acti
         {
             _subscriber.Subscribe(topic, OnMessage);
         }
+
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public override async Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            await _subscriber.SubscribeAsync(topic, OnMessage);
+        }
     }
 }
diff --git a/bus/EasyCaching.Bus.Zookeeper/DefaultZookeeperBus.cs b/bus/EasyCaching.Bus.Zookeeper/DefaultZookeeperBus.cs
@@ -116,6 +116,18 @@ public override void BaseSubscribe(string topic, Action<EasyCachingMessage> acti
             }, TaskCreationOptions.LongRunning);
         }
 
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public override async Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var path = $"{topic}";
+            await SubscribeDataChangeAsync(path, SubscribeDataChange);
+        }
+
         /// <summary>
         /// Ons the message.
         /// </summary>

diff --git a/src/EasyCaching.Core/Bus/IEasyCachingSubscriber.cs b/src/EasyCaching.Core/Bus/IEasyCachingSubscriber.cs
@@ -1,6 +1,8 @@
 namespace EasyCaching.Core.Bus
 {
     using System;
+    using System.Threading;
+    using System.Threading.Tasks;
 
     /// <summary>
     /// EasyCaching subscriber.
@@ -14,5 +16,14 @@ public interface IEasyCachingSubscriber
         /// <param name="action">Action.</param>
         /// <param name="reconnectAction"> Reconnect Action.</param>
         void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null);
+
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="reconnectAction"> Reconnect Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null, CancellationToken cancellationToken = default(CancellationToken));
     }
 }
diff --git a/src/EasyCaching.Core/Bus/NullEasyCachingBus.cs b/src/EasyCaching.Core/Bus/NullEasyCachingBus.cs
@@ -59,5 +59,17 @@ public void Subscribe(string topic, Action<EasyCachingMessage> action, Action re
         {
 
         }
+
+        /// <summary>
+        /// Subscribe the specified topic and action async.
+        /// </summary>
+        /// <param name="topic">Topic.</param>
+        /// <param name="action">Action.</param>
+        /// <param name="reconnectAction">Reconnect Action.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        public Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            return Task.CompletedTask;
+        }
     }
 }
diff --git a/src/EasyCaching.Core/EasyCachingAbstractBus.cs b/src/EasyCaching.Core/EasyCachingAbstractBus.cs
@@ -15,6 +15,7 @@ public abstract class EasyCachingAbstractBus : IEasyCachingBus
         public abstract void BasePublish(string topic, EasyCachingMessage message);
         public abstract Task BasePublishAsync(string topic, EasyCachingMessage message, CancellationToken cancellationToken = default(CancellationToken));
         public abstract void BaseSubscribe(string topic, Action<EasyCachingMessage> action);
+        public abstract Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken));
 
         protected Action<EasyCachingMessage> _handler;
 
@@ -83,6 +84,13 @@ public void Subscribe(string topic, Action<EasyCachingMessage> action, Action re
             BaseSubscribe(topic, action);
         }
 
+        public async Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            _handler = action;
+            _reconnectHandler = reconnectAction;
+            await BaseSubscribeAsync(topic, action);
+        }
+
         public virtual void BaseOnMessage(EasyCachingMessage message)
         {
             var operationId = s_diagnosticListener.WriteSubscribeMessageBefore(new BeforeSubscribeMessageRequestEventData(message));

diff --git a/src/EasyCaching.HybridCache/HybridCachingProvider.cs b/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, OnReconnect);
+            _ = SubscribeAsync();
 
             this._cacheId = Guid.NewGuid().ToString("N");
 
@@ -117,6 +117,15 @@ string name
             _busAsyncWrap = Policy.WrapAsync(fallbackAsyncPolicy, retryAsyncPolicy);
         }
 
+        /// <summary>
+        /// Subscribe the topic
+        /// </summary>
+        /// <returns></returns>
+        private async Task SubscribeAsync()
+        {
+            await _bus.SubscribeAsync(_options.TopicName, OnMessage, OnReconnect);
+        }
+
         /// <summary>
         /// Ons the message.
         /// </summary>

diff --git a/test/EasyCaching.UnitTests/Fake/FakeBus.cs b/test/EasyCaching.UnitTests/Fake/FakeBus.cs
@@ -23,5 +23,10 @@ public void Subscribe(string topic, Action<EasyCachingMessage> action, Action re
         {
 
         }
+
+        public Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            return Task.CompletedTask;
+        }
     }
 }
-Original file line number
+Diff line change
@@ Expand Up @@
             {
             }
+            public Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction = null, CancellationToken cancellationToken = default(CancellationToken))
+            {
+                return Task.CompletedTask;
+            }
         }
     }