Refactor subscriptions manager

2026-10-19 11:03:07 +08:00 · 2018-05-03 22:22:07 +02:00
parent d14088512b
commit b22b02a0b6
12 changed files with 269 additions and 107 deletions
@@ -290,18 +290,27 @@ namespace MQTTnet.Client

        private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
        {
-            _sendTracker.Restart();
-            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, new[] { packet }, cancellationToken);
+            return SendAsync(new[] { packet }, cancellationToken);
        }

        private Task SendAsync(IEnumerable<MqttBasePacket> packets, CancellationToken cancellationToken)
        {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                throw new TaskCanceledException();
+            }
+
            _sendTracker.Restart();
            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, packets, cancellationToken);
        }

        private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
        {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                throw new TaskCanceledException();
+            }
+
            _sendTracker.Restart();

            ushort identifier = 0;
@@ -528,7 +537,7 @@ namespace MQTTnet.Client

        private static async Task WaitForTaskAsync(Task task, Task sender)
        {
-            if (task == sender)
+            if (task == sender || task == null)
            {
                return;
            }
@@ -0,0 +1,26 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class AsyncAutoResetEvent : IDisposable
+    {
+        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1);
+
+        public Task WaitOneAsync(CancellationToken cancellationToken)
+        {
+            return _semaphore.WaitAsync(cancellationToken);
+        }
+
+        public void Set()
+        {
+            _semaphore.Release();
+        }
+
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+    }
+}
@@ -0,0 +1,26 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class AsyncLock : IDisposable
+    {
+        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
+
+        public Task EnterAsync(CancellationToken cancellationToken)
+        {
+            return _semaphore.WaitAsync(cancellationToken);
+        }
+
+        public void Exit()
+        {
+            _semaphore.Release();
+        }
+
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+    }
+}
@@ -5,6 +5,7 @@ using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;

@@ -13,7 +14,7 @@ namespace MQTTnet.Server
    public sealed class MqttClientPendingMessagesQueue : IDisposable
    {
        private readonly ConcurrentQueue<MqttBasePacket> _queue = new ConcurrentQueue<MqttBasePacket>();
-        private readonly SemaphoreSlim _queueWaitSemaphore = new SemaphoreSlim(0);
+        private readonly AsyncAutoResetEvent _queueAutoResetEvent = new AsyncAutoResetEvent();
        private readonly IMqttServerOptions _options;
        private readonly MqttClientSession _clientSession;
        private readonly IMqttNetLogger _logger;
@@ -54,7 +55,7 @@ namespace MQTTnet.Server
            if (packet == null) throw new ArgumentNullException(nameof(packet));

            _queue.Enqueue(packet);
-            _queueWaitSemaphore.Release();
+            _queueAutoResetEvent.Set();

            _logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
        }
@@ -82,7 +83,7 @@ namespace MQTTnet.Server
            MqttBasePacket packet = null;
            try
            {
-                await _queueWaitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+                await _queueAutoResetEvent.WaitOneAsync(cancellationToken).ConfigureAwait(false);
                if (!_queue.TryDequeue(out packet))
                {
                    throw new InvalidOperationException(); // should not happen
@@ -120,8 +121,8 @@ namespace MQTTnet.Server
                    if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                    {
                        publishPacket.Dup = true;
-                        _queue.Enqueue(packet);
-                        _queueWaitSemaphore.Release();
+
+                        Enqueue(publishPacket);
                    }
                }

@@ -134,7 +135,7 @@ namespace MQTTnet.Server

        public void Dispose()
        {
-            _queueWaitSemaphore?.Dispose();
+            _queueAutoResetEvent?.Dispose();
        }
    }
 }
@@ -134,7 +134,7 @@ namespace MQTTnet.Server
        {
            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));

-            var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
+            var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage).ConfigureAwait(false);
            if (!result.IsSubscribed)
            {
                return;
@@ -155,10 +155,10 @@ namespace MQTTnet.Server
        {
            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));

-            await SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
+            SubscriptionsManager.Subscribe(new MqttSubscribePacket
            {
                TopicFilters = topicFilters
-            }).ConfigureAwait(false);
+            });

            await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false);
        }
@@ -167,10 +167,12 @@ namespace MQTTnet.Server
        {
            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));

-            return SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
+            SubscriptionsManager.Unsubscribe(new MqttUnsubscribePacket
            {
                TopicFilters = topicFilters
            });
+
+            return Task.FromResult(0);
        }

        public void Dispose()
@@ -288,7 +290,7 @@ namespace MQTTnet.Server

        private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
        {
-            var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
+            var subscribeResult = SubscriptionsManager.Subscribe(subscribePacket);
            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { subscribeResult.ResponsePacket }, cancellationToken).ConfigureAwait(false);

            if (subscribeResult.CloseConnection)
@@ -302,7 +304,7 @@ namespace MQTTnet.Server

        private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
        {
-            var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
+            var unsubscribeResult = SubscriptionsManager.Unsubscribe(unsubscribePacket);
            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { unsubscribeResult }, cancellationToken);
        }

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -10,7 +11,7 @@ namespace MQTTnet.Server
 {
    public sealed class MqttClientSubscriptionsManager : IDisposable
    {
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
+        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscriptions = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
        private readonly IMqttServerOptions _options;
        private readonly MqttServer _server;
@@ -23,7 +24,7 @@ namespace MQTTnet.Server
            _server = server;
        }

-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
+        public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket)
        {
            if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));

@@ -37,57 +38,41 @@ namespace MQTTnet.Server
                CloseConnection = false
            };

-            await _semaphore.WaitAsync().ConfigureAwait(false);
-            try
+            foreach (var topicFilter in subscribePacket.TopicFilters)
            {
-                foreach (var topicFilter in subscribePacket.TopicFilters)
+                var interceptorContext = InterceptSubscribe(topicFilter);
+                if (!interceptorContext.AcceptSubscription)
                {
-                    var interceptorContext = InterceptSubscribe(topicFilter);
-                    if (!interceptorContext.AcceptSubscription)
-                    {
-                        result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
-                    }
-                    else
-                    {
-                        result.ResponsePacket.SubscribeReturnCodes.Add(ConvertToMaximumQoS(topicFilter.QualityOfServiceLevel));
-                    }
-
-                    if (interceptorContext.CloseConnection)
-                    {
-                        result.CloseConnection = true;
-                    }
-
-                    if (interceptorContext.AcceptSubscription)
-                    {
-                        _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
-                        _server.OnClientSubscribedTopic(_clientId, topicFilter);
-                    }
+                    result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
+                }
+                else
+                {
+                    result.ResponsePacket.SubscribeReturnCodes.Add(ConvertToMaximumQoS(topicFilter.QualityOfServiceLevel));
+                }
+
+                if (interceptorContext.CloseConnection)
+                {
+                    result.CloseConnection = true;
+                }
+
+                if (interceptorContext.AcceptSubscription)
+                {
+                    _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    _server.OnClientSubscribedTopic(_clientId, topicFilter);
                }
-            }
-            finally
-            {
-                _semaphore.Release();
            }

            return result;
        }

-        public async Task<MqttUnsubAckPacket> UnsubscribeAsync(MqttUnsubscribePacket unsubscribePacket)
+        public MqttUnsubAckPacket Unsubscribe(MqttUnsubscribePacket unsubscribePacket)
        {
            if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));

-            await _semaphore.WaitAsync().ConfigureAwait(false);
-            try
+            foreach (var topicFilter in unsubscribePacket.TopicFilters)
            {
-                foreach (var topicFilter in unsubscribePacket.TopicFilters)
-                {
-                    _subscriptions.Remove(topicFilter);
-                    _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
-                }
-            }
-            finally
-            {
-                _semaphore.Release();
+                _subscriptions.TryRemove(topicFilter, out _);
+                _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
            }

            return new MqttUnsubAckPacket