Merge branch 'develop' into master

Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs

@@ -3,11 +3,15 @@ using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
+    // TODO: Rename to "RegisteredClient"
+    // TODO: Add IsConnected
+    // TODO: Add interface
+
     public class ConnectedMqttClient
     {
         public string ClientId { get; set; }
 
-        public MqttProtocolVersion ProtocolVersion { get; set; }
+        public MqttProtocolVersion? ProtocolVersion { get; set; }
 
         public TimeSpan LastPacketReceived { get; set; }
 

Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs

@@ -0,0 +1,8 @@
+namespace MQTTnet.Server
+{
+    public enum MqttClientDisconnectType
+    {
+        Clean,
+        NotClean
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs

@@ -13,12 +13,12 @@ namespace MQTTnet.Server
         private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
 
         private readonly string _clientId;
-        private readonly Func<Task> _timeoutCallback;
+        private readonly Action _timeoutCallback;
         private readonly IMqttNetLogger _logger;
 
         private Task _workerTask;
 
-        public MqttClientKeepAliveMonitor(string clientId, Func<Task> timeoutCallback, IMqttNetLogger logger)
+        public MqttClientKeepAliveMonitor(string clientId, Action timeoutCallback, IMqttNetLogger logger)
         {
             _clientId = clientId;
             _timeoutCallback = timeoutCallback;
@@ -61,10 +61,7 @@ namespace MQTTnet.Server
                     {
                         _logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
 
-                        if (_timeoutCallback != null)
-                        {
-                            await _timeoutCallback().ConfigureAwait(false);
-                        }
+                        _timeoutCallback?.Invoke();
 
                         return;
                     }

Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs

@@ -1,11 +1,11 @@
 using System;
 using System.Collections.Concurrent;
-using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
@@ -14,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;
@@ -66,7 +66,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);
         }
@@ -94,7 +94,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
@@ -105,7 +105,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { packet }, cancellationToken).ConfigureAwait(false);
 
                 _logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }
@@ -132,21 +132,21 @@ namespace MQTTnet.Server
                     if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                     {
                         publishPacket.Dup = true;
-                        _queue.Enqueue(packet);
-                        _queueWaitSemaphore.Release();
+
+                        Enqueue(publishPacket);
                     }
                 }
 
                 if (!cancellationToken.IsCancellationRequested)
                 {
-                    await _clientSession.StopAsync().ConfigureAwait(false);
+                    _clientSession.Stop(MqttClientDisconnectType.NotClean);
                 }
             }
         }
 
         public void Dispose()
         {
-            _queueWaitSemaphore?.Dispose();
+            _queueAutoResetEvent?.Dispose();
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -16,11 +16,11 @@ namespace MQTTnet.Server
     public sealed class MqttClientSession : IDisposable
     {
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
-        private readonly IMqttServerOptions _options;
-        private readonly IMqttNetLogger _logger;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly IMqttNetLogger _logger;
+        private readonly IMqttServerOptions _options;
+        private readonly MqttClientSessionsManager _sessionsManager;
 
-        private IMqttChannelAdapter _adapter;
         private CancellationTokenSource _cancellationTokenSource;
         private MqttApplicationMessage _willMessage;
         private bool _wasCleanDisconnect;
@@ -28,22 +28,22 @@ namespace MQTTnet.Server
         public MqttClientSession(
             string clientId,
             IMqttServerOptions options,
+            MqttClientSessionsManager sessionsManager,
             MqttRetainedMessagesManager retainedMessagesManager,
             IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _sessionsManager = sessionsManager;
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
             ClientId = clientId;
 
-            KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger);
-            SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
+            KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeout, _logger);
+            SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server);
             PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
         }
 
-        public Func<MqttClientSession, MqttApplicationMessage, Task> ApplicationMessageReceivedCallback { get; set; }
-
         public MqttClientSubscriptionsManager SubscriptionsManager { get; }
 
         public MqttClientPendingMessagesQueue PendingMessagesQueue { get; }
@@ -52,9 +52,9 @@ namespace MQTTnet.Server
 
         public string ClientId { get; }
 
-        public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
+        public MqttProtocolVersion? ProtocolVersion { get; private set; }
 
-        public bool IsConnected => _adapter != null;
+        public bool IsConnected { get; private set; }
 
         public async Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
         {
@@ -63,34 +63,45 @@ namespace MQTTnet.Server
 
             try
             {
-                var cancellationTokenSource = new CancellationTokenSource();
+                _cancellationTokenSource = new CancellationTokenSource();
 
                 _wasCleanDisconnect = false;
                 _willMessage = connectPacket.WillMessage;
-                _adapter = adapter;
-                _cancellationTokenSource = cancellationTokenSource;
 
-                PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
-                KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, cancellationTokenSource.Token);
+                IsConnected = true;
+                ProtocolVersion = adapter.PacketSerializer.ProtocolVersion;
 
-                await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
+                PendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token);
+                KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token);
+
+                await ReceivePacketsAsync(adapter, _cancellationTokenSource.Token).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
             }
             catch (MqttCommunicationException exception)
             {
-                _logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
+                _logger.Warning<MqttClientSession>(exception,
+                    "Client '{0}': Communication exception while processing client packets.", ClientId);
             }
             catch (Exception exception)
             {
-                _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
+                _logger.Error<MqttClientSession>(exception,
+                    "Client '{0}': Unhandled exception while processing client packets.", ClientId);
+            }
+            finally
+            {
+                ProtocolVersion = null;
+                IsConnected = false;
+
+                _cancellationTokenSource?.Dispose();
+                _cancellationTokenSource = null;
             }
 
             return _wasCleanDisconnect;
         }
 
-        public async Task StopAsync(bool wasCleanDisconnect = false)
+        public void Stop(MqttClientDisconnectType type)
         {
             try
             {
@@ -99,37 +110,31 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                _wasCleanDisconnect = wasCleanDisconnect;
+                _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
 
                 _cancellationTokenSource?.Cancel(false);
-
                 PendingMessagesQueue.WaitForCompletion();
                 KeepAliveMonitor.WaitForCompletion();
 
-                _cancellationTokenSource?.Dispose();
-                _cancellationTokenSource = null;
-
-                _adapter = null;
-
-                _logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
-            }
-            finally
-            {
                 var willMessage = _willMessage;
                 _willMessage = null; // clear willmessage so it is send just once
 
-                if (willMessage != null && !wasCleanDisconnect)
+                if (willMessage != null && !_wasCleanDisconnect)
                 {
-                    await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
+                    _sessionsManager.StartDispatchApplicationMessage(this, willMessage);
                 }
             }
+            finally
+            {
+                _logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
+            }
         }
 
         public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage)
         {
             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;
@@ -153,10 +158,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);
         }
@@ -165,26 +170,26 @@ 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()
         {
-            ApplicationMessageReceivedCallback = null;
-
             SubscriptionsManager?.Dispose();
             PendingMessagesQueue?.Dispose();
 
             _cancellationTokenSource?.Dispose();
         }
 
-        private Task StopDueToKeepAliveTimeoutAsync()
+        private void StopDueToKeepAliveTimeout()
         {
             _logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
-            return StopAsync();
+            Stop(MqttClientDisconnectType.NotClean);
         }
 
         private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -201,15 +206,18 @@ namespace MQTTnet.Server
             catch (OperationCanceledException)
             {
             }
-            catch (MqttCommunicationException exception)
-            {
-                _logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
-                await StopAsync().ConfigureAwait(false);
-            }
             catch (Exception exception)
             {
-                _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
-                await StopAsync().ConfigureAwait(false);
+                if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
+                }
+                else
+                {
+                    _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
+                }
+
+                Stop(MqttClientDisconnectType.NotClean);
             }
         }
 
@@ -222,7 +230,7 @@ namespace MQTTnet.Server
 
             if (packet is MqttPingReqPacket)
             {
-                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() });
+                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { new MqttPingRespPacket() }, cancellationToken);
             }
 
             if (packet is MqttPubRelPacket pubRelPacket)
@@ -237,7 +245,7 @@ namespace MQTTnet.Server
                     PacketIdentifier = pubRecPacket.PacketIdentifier
                 };
 
-                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket });
+                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { responsePacket }, cancellationToken);
             }
 
             if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -258,35 +266,20 @@ namespace MQTTnet.Server
 
             if (packet is MqttDisconnectPacket)
             {
-                return StopAsync(true);
+                Stop(MqttClientDisconnectType.Clean);
+                return Task.FromResult(0);
             }
 
             if (packet is MqttConnectPacket)
             {
-                return StopAsync();
+                Stop(MqttClientDisconnectType.NotClean);
+                return Task.FromResult(0);
             }
 
             _logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            return StopAsync();
-        }
+            Stop(MqttClientDisconnectType.NotClean);
 
-        private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
-        {
-            var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false);
-
-            if (subscribeResult.CloseConnection)
-            {
-                await StopAsync().ConfigureAwait(false);
-            }
-
-            await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
-        }
-
-        private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
-        {
-            var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult });
+            return Task.FromResult(0);
         }
 
         private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
@@ -298,6 +291,26 @@ namespace MQTTnet.Server
             }
         }
 
+        private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+        {
+            var subscribeResult = SubscriptionsManager.Subscribe(subscribePacket);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { subscribeResult.ResponsePacket }, cancellationToken).ConfigureAwait(false);
+
+            if (subscribeResult.CloseConnection)
+            {
+                Stop(MqttClientDisconnectType.NotClean);
+                return;
+            }
+
+            await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
+        }
+
+        private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        {
+            var unsubscribeResult = SubscriptionsManager.Unsubscribe(unsubscribePacket);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { unsubscribeResult }, cancellationToken);
+        }
+
         private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             var applicationMessage = publishPacket.ToApplicationMessage();
@@ -306,7 +319,8 @@ namespace MQTTnet.Server
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        return ApplicationMessageReceivedCallback?.Invoke(this, applicationMessage);
+                        _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
+                        return Task.FromResult(0);
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
@@ -325,25 +339,25 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
-            await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
+            _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
 
             var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken).ConfigureAwait(false);
         }
 
         private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-            await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
+            _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
 
             var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken).ConfigureAwait(false);
         }
 
         private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
             var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
-            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response });
+            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken);
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -8,38 +8,34 @@ using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
-using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
     public sealed class MqttClientSessionsManager : IDisposable
     {
         private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
-        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
+        private readonly SemaphoreSlim _sessionsLock = new SemaphoreSlim(1, 1);
 
-        private readonly IMqttServerOptions _options;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly IMqttServerOptions _options;
         private readonly IMqttNetLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
+        public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            Server = server ?? throw new ArgumentNullException(nameof(server));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
-        public Action<ConnectedMqttClient, bool> ClientDisconnectedCallback { get; set; }
-        public Action<string, TopicFilter> ClientSubscribedTopicCallback { get; set; }
-        public Action<string, string> ClientUnsubscribedTopicCallback { get; set; }
-        public Action<string, MqttApplicationMessage> ApplicationMessageReceivedCallback { get; set; }
+        public MqttServer Server { get; }
 
         public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
         {
             var clientId = string.Empty;
             var wasCleanDisconnect = false;
             MqttClientSession clientSession = null;
-            
+
             try
             {
                 if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken)
@@ -57,30 +53,30 @@ namespace MQTTnet.Server
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
-                    await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
+                    await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[]
                     {
                         new MqttConnAckPacket
                         {
                             ConnectReturnCode = connectReturnCode
                         }
-                    }).ConfigureAwait(false);
+                    }, cancellationToken).ConfigureAwait(false);
 
                     return;
                 }
 
-                var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
+                var result = await PrepareClientSessionAsync(connectPacket).ConfigureAwait(false);
                 clientSession = result.Session;
 
-                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
+                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[]
                 {
                     new MqttConnAckPacket
                     {
                         ConnectReturnCode = connectReturnCode,
                         IsSessionPresent = result.IsExistingSession
                     }
-                }).ConfigureAwait(false);
+                }, cancellationToken).ConfigureAwait(false);
 
-                ClientConnectedCallback?.Invoke(new ConnectedMqttClient
+                Server.OnClientConnected(new ConnectedMqttClient
                 {
                     ClientId = clientId,
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
@@ -99,15 +95,15 @@ namespace MQTTnet.Server
             {
                 try
                 {
-                    await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                    await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
                     clientAdapter.Dispose();
                 }
                 catch (Exception exception)
                 {
                     _logger.Error<MqttClientSessionsManager>(exception, exception.Message);
                 }
-                
-                ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
+
+                Server.OnClientDisconnected(new ConnectedMqttClient
                 {
                     ClientId = clientId,
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
@@ -119,31 +115,31 @@ namespace MQTTnet.Server
 
         public async Task StopAsync()
         {
-            await _semaphore.WaitAsync().ConfigureAwait(false);
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 foreach (var session in _sessions)
                 {
-                    await session.Value.StopAsync().ConfigureAwait(false);
+                    session.Value.Stop(MqttClientDisconnectType.NotClean);
                 }
 
                 _sessions.Clear();
             }
             finally
             {
-                _semaphore.Release();
+                _sessionsLock.Release();
             }
         }
 
         public async Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
         {
-            await _semaphore.WaitAsync().ConfigureAwait(false);
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                 {
                     ClientId = s.Value.ClientId,
-                    ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
+                    ProtocolVersion = s.Value.ProtocolVersion,
                     LastPacketReceived = s.Value.KeepAliveMonitor.LastPacketReceived,
                     LastNonKeepAlivePacketReceived = s.Value.KeepAliveMonitor.LastNonKeepAlivePacketReceived,
                     PendingApplicationMessages = s.Value.PendingMessagesQueue.Count
@@ -151,49 +147,13 @@ namespace MQTTnet.Server
             }
             finally
             {
-                _semaphore.Release();
+                _sessionsLock.Release();
             }
         }
 
-        public async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
-            try
-            {
-                var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
-                if (interceptorContext.CloseConnection)
-                {
-                    await senderClientSession.StopAsync().ConfigureAwait(false);
-                }
-
-                if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
-                {
-                    return;
-                }
-
-                if (applicationMessage.Retain)
-                {
-                    await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
-                }
-
-                ApplicationMessageReceivedCallback?.Invoke(senderClientSession?.ClientId, applicationMessage);
-            }
-            catch (Exception exception)
-            {
-                _logger.Error<MqttClientSessionsManager>(exception, "Error while processing application message");
-            }
-
-            await _semaphore.WaitAsync().ConfigureAwait(false);
-            try
-            {
-                foreach (var clientSession in _sessions.Values)
-                {
-                    await clientSession.EnqueueApplicationMessageAsync(applicationMessage);
-                }
-            }
-            finally
-            {
-                _semaphore.Release();
-            }
+            Task.Run(() => DispatchApplicationMessageAsync(senderClientSession, applicationMessage));
         }
 
         public async Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
@@ -201,19 +161,19 @@ namespace MQTTnet.Server
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
-            await _semaphore.WaitAsync().ConfigureAwait(false);
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 if (!_sessions.TryGetValue(clientId, out var session))
                 {
-                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                    throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
                 }
 
-                await session.SubscribeAsync(topicFilters);
+                await session.SubscribeAsync(topicFilters).ConfigureAwait(false);
             }
             finally
             {
-                _semaphore.Release();
+                _sessionsLock.Release();
             }
         }
 
@@ -222,36 +182,25 @@ namespace MQTTnet.Server
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
-            await _semaphore.WaitAsync().ConfigureAwait(false);
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 if (!_sessions.TryGetValue(clientId, out var session))
                 {
-                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                    throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
                 }
 
-                await session.UnsubscribeAsync(topicFilters);
+                await session.UnsubscribeAsync(topicFilters).ConfigureAwait(false);
             }
             finally
             {
-                _semaphore.Release();
+                _sessionsLock.Release();
             }
         }
 
-        private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public void Dispose()
         {
-            var interceptorContext = new MqttApplicationMessageInterceptorContext(
-                senderClientSession?.ClientId,
-                applicationMessage);
-
-            var interceptor = _options.ApplicationMessageInterceptor;
-            if (interceptor == null)
-            {
-                return interceptorContext;
-            }
-            
-            interceptor(interceptorContext);
-            return interceptorContext;
+            _sessionsLock?.Dispose();
         }
 
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
@@ -271,9 +220,9 @@ namespace MQTTnet.Server
             return context.ReturnCode;
         }
 
-        private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
+        private async Task<GetOrCreateClientSessionResult> PrepareClientSessionAsync(MqttConnectPacket connectPacket)
         {
-            await _semaphore.WaitAsync().ConfigureAwait(false);
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
             try
             {
                 var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
@@ -283,7 +232,7 @@ namespace MQTTnet.Server
                     {
                         _sessions.Remove(connectPacket.ClientId);
 
-                        await clientSession.StopAsync().ConfigureAwait(false);
+                        clientSession.Stop(MqttClientDisconnectType.Clean);
                         clientSession.Dispose();
                         clientSession = null;
 
@@ -300,14 +249,7 @@ namespace MQTTnet.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, _logger)
-                    {
-                        ApplicationMessageReceivedCallback = DispatchApplicationMessageAsync
-                    };
-
-                    clientSession.SubscriptionsManager.TopicSubscribedCallback = ClientSubscribedTopicCallback;
-                    clientSession.SubscriptionsManager.TopicUnsubscribedCallback = ClientUnsubscribedTopicCallback;
-
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _logger);
                     _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.Verbose<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -317,19 +259,65 @@ namespace MQTTnet.Server
             }
             finally
             {
-                _semaphore.Release();
+                _sessionsLock.Release();
             }
         }
 
-        public void Dispose()
+        private async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
-            ClientConnectedCallback = null;
-            ClientDisconnectedCallback = null;
-            ClientSubscribedTopicCallback = null;
-            ClientUnsubscribedTopicCallback = null;
-            ApplicationMessageReceivedCallback = null;
+            try
+            {
+                var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
+                if (interceptorContext.CloseConnection)
+                {
+                    senderClientSession.Stop(MqttClientDisconnectType.NotClean);
+                }
 
-            _semaphore?.Dispose();
+                if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
+                {
+                    return;
+                }
+
+                if (applicationMessage.Retain)
+                {
+                    await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
+                }
+
+                Server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error<MqttClientSessionsManager>(exception, "Error while processing application message");
+            }
+
+            await _sessionsLock.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                foreach (var clientSession in _sessions.Values)
+                {
+                    await clientSession.EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                _sessionsLock.Release();
+            }
+        }
+
+        private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        {
+            var interceptorContext = new MqttApplicationMessageInterceptorContext(
+                senderClientSession?.ClientId,
+                applicationMessage);
+
+            var interceptor = _options.ApplicationMessageInterceptor;
+            if (interceptor == null)
+            {
+                return interceptorContext;
+            }
+
+            interceptor(interceptorContext);
+            return interceptorContext;
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -10,21 +11,20 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSubscriptionsManager : IDisposable
     {
+        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscriptions = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
+        private readonly MqttServer _server;
         private readonly string _clientId;
 
-        public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId)
+        public MqttClientSubscriptionsManager(string clientId, IMqttServerOptions options, MqttServer server)
         {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
             _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+            _server = server;
         }
 
-        public Action<string, TopicFilter> TopicSubscribedCallback { get; set; }
-        public Action<string, string> TopicUnsubscribedCallback { get; set; }
-
-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
+        public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
@@ -38,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;
-                        TopicSubscribedCallback?.Invoke(_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);
-                    TopicUnsubscribedCallback?.Invoke(_clientId, topicFilter);
-                }
-            }
-            finally
-            {
-                _semaphore.Release();
+                _subscriptions.TryRemove(topicFilter, out _);
+                _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
             }
 
             return new MqttUnsubAckPacket

Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs

@@ -57,7 +57,7 @@ namespace MQTTnet.Server
             return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
         }
 
-        public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
+        public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
 
@@ -65,8 +65,10 @@ namespace MQTTnet.Server
 
             foreach (var applicationMessage in applicationMessages)
             {
-                await _clientSessionsManager.DispatchApplicationMessageAsync(null, applicationMessage);
+                _clientSessionsManager.StartDispatchApplicationMessage(null, applicationMessage);
             }
+
+            return Task.FromResult(0);
         }
 
         public async Task StartAsync(IMqttServerOptions options)
@@ -80,14 +82,7 @@ namespace MQTTnet.Server
             _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync();
 
-            _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _logger)
-            {
-                ClientConnectedCallback = OnClientConnected,
-                ClientDisconnectedCallback = OnClientDisconnected,
-                ClientSubscribedTopicCallback = OnClientSubscribedTopic,
-                ClientUnsubscribedTopicCallback = OnClientUnsubscribedTopic,
-                ApplicationMessageReceivedCallback = OnApplicationMessageReceived
-            };
+            _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _logger);
 
             foreach (var adapter in _adapters)
             {
@@ -132,29 +127,29 @@ namespace MQTTnet.Server
             }
         }
 
-        private void OnClientConnected(ConnectedMqttClient client)
+        internal void OnClientConnected(ConnectedMqttClient client)
         {
             _logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
             ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
         }
 
-        private void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect)
+        internal void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect)
         {
             _logger.Info<MqttServer>("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect);
             ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client, wasCleanDisconnect));
         }
 
-        private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
+        internal void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
         {
             ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
         }
 
-        private void OnClientUnsubscribedTopic(string clientId, string topicFilter)
+        internal void OnClientUnsubscribedTopic(string clientId, string topicFilter)
         {
             ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
         }
 
-        private void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
+        internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
         {
             ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
         }

Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs

@@ -19,12 +19,17 @@ namespace MQTTnet.Server
             var fragmentsTopic = topic.Split(TopicLevelSeparator, StringSplitOptions.None);
             var fragmentsFilter = filter.Split(TopicLevelSeparator, StringSplitOptions.None);
 
+            // # > In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2].
             for (var i = 0; i < fragmentsFilter.Length; i++)
             {
-                switch (fragmentsFilter[i])
+                if (fragmentsFilter[i] == "+")
                 {
-                    case "+": continue;
-                    case "#" when i == fragmentsFilter.Length - 1: return true;
+                    continue;
+                }
+
+                if (fragmentsFilter[i] == "#")
+                {
+                    return true;
                 }
 
                 if (i >= fragmentsTopic.Length)
@@ -38,7 +43,7 @@ namespace MQTTnet.Server
                 }
             }
 
-            return fragmentsTopic.Length <= fragmentsFilter.Length;
+            return fragmentsTopic.Length == fragmentsFilter.Length;
         }
     }
 }