Improve performance.

Source/MQTTnet/Server/CheckSubscriptionsResult.cs

@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Server
 {
-    public class CheckSubscriptionsResult
+    public struct CheckSubscriptionsResult
     {
         public bool IsSubscribed { get; set; }
 

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -18,7 +18,7 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSessionsManager : IDisposable
     {
-        readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
+        readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
 
         readonly object _createConnectionSyncRoot = new object();
         readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>();
@@ -176,7 +176,7 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            _messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
+            _messageQueue.Add(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
         }
 
         public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -270,46 +270,41 @@ namespace MQTTnet.Server
         {
             try
             {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
-
-                var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
-                if (!dequeueResult.IsSuccess)
-                {
-                    return;
-                }
-
-                var queuedApplicationMessage = dequeueResult.Item;
+                cancellationToken.ThrowIfCancellationRequested();
 
+                var queuedApplicationMessage = _messageQueue.Take(cancellationToken);
                 var sender = queuedApplicationMessage.Sender;
+                var senderClientId = sender?.ClientId ?? _options.ClientId;
                 var applicationMessage = queuedApplicationMessage.ApplicationMessage;
 
-                var interceptorContext = await InterceptApplicationMessageAsync(sender, applicationMessage).ConfigureAwait(false);
-                if (interceptorContext != null)
+                var interceptor = _options.ApplicationMessageInterceptor;
+                if (interceptor != null)
                 {
-                    if (interceptorContext.CloseConnection)
+                    var interceptorContext = await InterceptApplicationMessageAsync(interceptor, sender, applicationMessage).ConfigureAwait(false);
+                    if (interceptorContext != null)
                     {
-                        if (sender != null)
+                        if (interceptorContext.CloseConnection)
                         {
-                            await sender.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                            if (sender != null)
+                            {
+                                await sender.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                            }
                         }
-                    }
 
-                    if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
-                    {
-                        return;
-                    }
+                        if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
+                        {
+                            return;
+                        }
 
-                    applicationMessage = interceptorContext.ApplicationMessage;
+                        applicationMessage = interceptorContext.ApplicationMessage;
+                    }
                 }
-
-                await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
+                
+                await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(senderClientId, applicationMessage).ConfigureAwait(false);
 
                 if (applicationMessage.Retain)
                 {
-                    await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
+                    await _retainedMessagesManager.HandleMessageAsync(senderClientId, applicationMessage).ConfigureAwait(false);
                 }
 
                 var deliveryCount = 0;
@@ -318,11 +313,7 @@ namespace MQTTnet.Server
                 {
                     foreach (var clientSession in _sessions.Values)
                     {
-                        var isSubscribed = clientSession.EnqueueApplicationMessage(
-                            applicationMessage,
-                            sender?.ClientId,
-                            false);
-
+                        var isSubscribed = clientSession.EnqueueApplicationMessage(applicationMessage, senderClientId, false);
                         if (isSubscribed)
                         {
                             deliveryCount++;
@@ -333,13 +324,12 @@ namespace MQTTnet.Server
                 if (deliveryCount == 0)
                 {
                     var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor;
-
                     if (undeliveredMessageInterceptor == null)
                     {
                         return;
                     }
 
-                    await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage));
+                    await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(senderClientId, sender?.Session?.Items, applicationMessage));
                 }
             }
             catch (OperationCanceledException)
@@ -423,14 +413,8 @@ namespace MQTTnet.Server
             }
         }
 
-        async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
+        async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
         {
-            var interceptor = _options.ApplicationMessageInterceptor;
-            if (interceptor == null)
-            {
-                return null;
-            }
-
             string senderClientId;
             IDictionary<object, object> sessionItems;
 

Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs

@@ -3,23 +3,26 @@ using MQTTnet.Protocol;
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
     public sealed class MqttClientSubscriptionsManager
     {
+        // We use a reader writer lock because the subscriptions are only read most of the time.
+        // Since writing is done by multiple threads (server API or connection thread), we cannot avoid locking
+        // completely by swapping references etc.
+        readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
         readonly Dictionary<string, MqttTopicFilter> _subscriptions = new Dictionary<string, MqttTopicFilter>();
         readonly MqttClientSession _clientSession;
-        readonly IMqttServerOptions _serverOptions;
+        readonly IMqttServerOptions _options;
         readonly MqttServerEventDispatcher _eventDispatcher;
 
         public MqttClientSubscriptionsManager(MqttClientSession clientSession, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
         {
             _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
-
-            // TODO: Consider removing the server options here and build a new class "ISubscriptionInterceptor" and just pass it. The instance is generated in the root server class upon start.
-            _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
+            _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
         }
 
@@ -52,15 +55,22 @@ namespace MQTTnet.Server
                     result.CloseConnection = true;
                 }
 
-                if (interceptorContext.AcceptSubscription && !string.IsNullOrEmpty(finalTopicFilter?.Topic))
+                if (!interceptorContext.AcceptSubscription || string.IsNullOrEmpty(finalTopicFilter?.Topic))
                 {
-                    lock (_subscriptions)
-                    {
-                        _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
-                    }
-
-                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
+                    continue;
                 }
+
+                _subscriptionsLock.EnterWriteLock();
+                try
+                {
+                    _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
+
+                await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
             }
 
             return result;
@@ -78,15 +88,22 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                if (interceptorContext.AcceptSubscription)
+                if (!interceptorContext.AcceptSubscription)
                 {
-                    lock (_subscriptions)
-                    {
-                        _subscriptions[topicFilter.Topic] = topicFilter;
-                    }
-
-                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                    continue;
                 }
+
+                _subscriptionsLock.EnterWriteLock();
+                try
+                {
+                    _subscriptions[topicFilter.Topic] = topicFilter;
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
+
+                await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
             }
         }
 
@@ -105,16 +122,16 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                lock (_subscriptions)
+                _subscriptionsLock.EnterWriteLock();
+                try
                 {
-                    if (_subscriptions.Remove(topicFilter))
-                    {
-                        reasonCodes.Add(MqttUnsubscribeReasonCode.Success);
-                    }
-                    else
-                    {
-                        reasonCodes.Add(MqttUnsubscribeReasonCode.NoSubscriptionExisted);
-                    }
+                    reasonCodes.Add(_subscriptions.Remove(topicFilter)
+                        ? MqttUnsubscribeReasonCode.Success
+                        : MqttUnsubscribeReasonCode.NoSubscriptionExisted);
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
                 }
             }
 
@@ -138,28 +155,40 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                lock (_subscriptions)
+                _subscriptionsLock.EnterWriteLock();
+                try
                 {
                     _subscriptions.Remove(topicFilter);
                 }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
             }
         }
 
         public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel)
         {
-            var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
-
-            lock (_subscriptions)
+            Dictionary<string, MqttTopicFilter> subscriptionsCopy;
+            _subscriptionsLock.EnterReadLock();
+            try
             {
-                foreach (var subscription in _subscriptions)
-                {
-                    if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Key))
-                    {
-                        continue;
-                    }
+                subscriptionsCopy = new Dictionary<string, MqttTopicFilter>(_subscriptions);
+            }
+            finally
+            {
+                _subscriptionsLock.ExitReadLock();
+            }
 
-                    qosLevels.Add(subscription.Value.QualityOfServiceLevel);
+            var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
+            foreach (var subscription in subscriptionsCopy)
+            {
+                if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Key))
+                {
+                    continue;
                 }
+                
+                qosLevels.Add(subscription.Value.QualityOfServiceLevel);
             }
 
             if (qosLevels.Count == 0)
@@ -175,32 +204,20 @@ namespace MQTTnet.Server
 
         static MqttSubscribeReturnCode ConvertToSubscribeReturnCode(MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
-                default: return MqttSubscribeReturnCode.Failure;
-            }
+            return (MqttSubscribeReturnCode)(int)qualityOfServiceLevel;
         }
 
         static MqttSubscribeReasonCode ConvertToSubscribeReasonCode(MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReasonCode.GrantedQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReasonCode.GrantedQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReasonCode.GrantedQoS2;
-                default: return MqttSubscribeReasonCode.UnspecifiedError;
-            }
+            return (MqttSubscribeReasonCode)(int)qualityOfServiceLevel;
         }
 
         async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(MqttTopicFilter topicFilter)
         {
             var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
-            if (_serverOptions.SubscriptionInterceptor != null)
+            if (_options.SubscriptionInterceptor != null)
             {
-                await _serverOptions.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
+                await _options.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
             }
 
             return context;
@@ -209,9 +226,9 @@ namespace MQTTnet.Server
         async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribeAsync(string topicFilter)
         {
             var context = new MqttUnsubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
-            if (_serverOptions.UnsubscriptionInterceptor != null)
+            if (_options.UnsubscriptionInterceptor != null)
             {
-                await _serverOptions.UnsubscriptionInterceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false);
+                await _options.UnsubscriptionInterceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false);
             }
 
             return context;

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -1,17 +1,17 @@
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
-using MQTTnet.Internal;
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
-    public class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
+    public sealed class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
     {
-        readonly byte[] _emptyArray = new byte[0];
-        readonly AsyncLock _messagesLock = new AsyncLock();
+        readonly AsyncLock _storageAccessLock = new AsyncLock();
         readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
 
         IMqttNetScopedLogger _logger;
@@ -37,12 +37,13 @@ namespace MQTTnet.Server
             try
             {
                 var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
-                if (retainedMessages?.Any() == true)
-                {
-                    using (await _messagesLock.WaitAsync().ConfigureAwait(false))
-                    {
-                        _messages.Clear();
 
+                lock (_messages)
+                {
+                    _messages.Clear();
+
+                    if (retainedMessages != null)
+                    {
                         foreach (var retainedMessage in retainedMessages)
                         {
                             _messages[retainedMessage.Topic] = retainedMessage;
@@ -62,9 +63,11 @@ namespace MQTTnet.Server
 
             try
             {
-                using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+                List<MqttApplicationMessage> messagesForSave = null;
+                var saveIsRequired = false;
+                
+                lock (_messages)
                 {
-                    var saveIsRequired = false;
                     var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
 
                     if (!hasPayload)
@@ -81,7 +84,7 @@ namespace MQTTnet.Server
                         }
                         else
                         {
-                            if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? _emptyArray))
+                            if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray))
                             {
                                 _messages[applicationMessage.Topic] = applicationMessage;
                                 saveIsRequired = true;
@@ -93,9 +96,16 @@ namespace MQTTnet.Server
 
                     if (saveIsRequired)
                     {
-                        if (_options.Storage != null)
+                        messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
+                    }
+                }
+
+                if (saveIsRequired)
+                {
+                    if (_options.Storage != null)
+                    {
+                        using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                         {
-                            var messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
                             await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
                         }
                     }
@@ -107,14 +117,14 @@ namespace MQTTnet.Server
             }
         }
 
-        public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
+        public Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
             var matchingRetainedMessages = new List<MqttApplicationMessage>();
 
             List<MqttApplicationMessage> retainedMessages;
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
                 retainedMessages = _messages.Values.ToList();
             }
@@ -133,24 +143,27 @@ namespace MQTTnet.Server
                 }
             }
 
-            return matchingRetainedMessages;
+            return Task.FromResult((IList<MqttApplicationMessage>)matchingRetainedMessages);
         }
 
-        public async Task<IList<MqttApplicationMessage>> GetMessagesAsync()
+        public Task<IList<MqttApplicationMessage>> GetMessagesAsync()
         {
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
-                return _messages.Values.ToList();
+                return Task.FromResult((IList<MqttApplicationMessage>)_messages.Values.ToList());
             }
         }
 
         public async Task ClearMessagesAsync()
         {
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
                 _messages.Clear();
+            }
 
-                if (_options.Storage != null)
+            if (_options.Storage != null)
+            {
+                using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                 {
                     await _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>()).ConfigureAwait(false);
                 }

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -108,7 +108,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false); ;
+                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Server/MqttTopicFilterComparer.cs

@@ -4,14 +4,14 @@ namespace MQTTnet.Server
 {
     public static class MqttTopicFilterComparer
     {
-        private const char LevelSeparator = '/';
-        private const char MultiLevelWildcard = '#';
-        private const char SingleLevelWildcard = '+';
+        const char LevelSeparator = '/';
+        const char MultiLevelWildcard = '#';
+        const char SingleLevelWildcard = '+';
 
         public static bool IsMatch(string topic, string filter)
         {
-            if (string.IsNullOrEmpty(topic)) throw new ArgumentNullException(nameof(topic));
-            if (string.IsNullOrEmpty(filter)) throw new ArgumentNullException(nameof(filter));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+            if (filter == null) throw new ArgumentNullException(nameof(filter));
 
             var sPos = 0;
             var sLen = filter.Length;