Fix broken connection takeover.

improved behavior when multiple connections fight over a session

Build/MQTTnet.nuspec

@@ -12,16 +12,13 @@
     <requireLicenseAcceptance>true</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
     <releaseNotes>
-* [LowLevelClient] Fixed a null reference exception when connecting to a not existing server (thanks to @SGStino).      
-* [RcpClient] Adjusted some namespaces (BREAKING CHANGE!).
-* [AspNetCore] Adjusted some namespaces (BREAKING CHANGE!).
-* [Server] Adjusted some namespaces (BREAKING CHANGE!).
-* [Server] Added state checks (throw if not started etc.) for most server APIs.
-* [Server] Exposed real X509Certificate2 (instead byte array) to TLS options (thanks to @borigas).
-* [Server] Fixed memory leak with TCP sockets (MqttServer is now Disposable!) (BREAKING CHANGE!).
-* [Core] Fixed a null reference exception in the MqttTcpChannel with WriteAsync and ReadAsync.
-* [Core] Added server interceptor for undelivered messages (thanks to @cshark-inator).
-* [nuget] Added support for SourceLink (thanks to @JTOne123).
+* [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder).
+* [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause).
+* [Client] handle disconnect package and propagate disconnect reason code (thanks to @JanEggers)
+* [ManagedClient] Exposed the internal MQTT client.
+* [Server] Added client message queue interceptor for QoS level (thanks to @msallin).
+* [Server] improved behavior when multiple connections fight over a session (thanks to @JanEggers)
+* [Server] Exposed _ClientDisconnectedInterceptor_ (thanks to @SeppPenner).
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2020</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</tags> 
@@ -44,6 +41,12 @@
         <dependency id="System.Net.WebSockets" version="4.3.0" />
         <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
       </group>
+      <group targetFramework="netcoreapp3.1">
+        <dependency id="NETStandard.Library" version="2.0.0" />
+        <dependency id="System.Net.Security" version="4.3.2" />
+        <dependency id="System.Net.WebSockets" version="4.3.0" />
+        <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
+      </group>
       <group targetFramework="uap10.0">
         <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.2.10" />
       </group>
@@ -66,6 +69,9 @@
     <!-- .NET Standard 2.1 -->
     <file src="..\Source\MQTTnet\bin\Release\netstandard2.1\MQTTnet.*" target="lib\netstandard2.1\"/>
 
+    <!-- .NET Standard 3.1 -->
+    <file src="..\Source\MQTTnet\bin\Release\netcoreapp3.1\MQTTnet.*" target="lib\netcoreapp3.1\"/>
+
     <!-- Universal Windows -->
     <file src="..\Source\MQTTnet\bin\Release\uap10.0\MQTTnet.*" target="lib\uap10.0\"/>
 

Build/build.ps1

@@ -29,6 +29,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp3.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
 # Build the ASP.NET Core 2.0 extension

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -4,30 +4,46 @@ using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
     public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
-        bool IsStarted { get; }
-        bool IsConnected { get; }
-        int PendingApplicationMessagesCount { get; }
-        IManagedMqttClientOptions Options { get; }
+        /// <summary>
+        /// Gets the internally used MQTT client.
+        /// This property should be used with caution because manipulating the internal client might break the managed client.
+        /// </summary>
+        IMqttClient InternalClient { get; }
 
+        bool IsStarted { get; }
+        
+        bool IsConnected { get; }
+        
+        int PendingApplicationMessagesCount { get; }
+        
+        IManagedMqttClientOptions Options { get; }
+        
         IMqttClientConnectedHandler ConnectedHandler { get; set; }
+        
         IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
 
         IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; }
+        
         IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; }
 
         IConnectingFailedHandler ConnectingFailedHandler { get; set; }
+        
         ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
 
         Task StartAsync(IManagedMqttClientOptions options);
+        
         Task StopAsync();
+        
         Task PingAsync(CancellationToken cancellationToken);
 
         Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters);
+        
         Task UnsubscribeAsync(IEnumerable<string> topics);
 
         Task PublishAsync(ManagedMqttApplicationMessage applicationMessages);

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs

@@ -8,6 +8,8 @@ namespace MQTTnet.Extensions.ManagedClient
     {
         IMqttClientOptions ClientOptions { get; }
 
+        bool AutoReconnect { get; }
+
         TimeSpan AutoReconnectDelay { get; }
 
         TimeSpan ConnectionCheckInterval { get; }

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -18,7 +18,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClient : Disposable, IManagedMqttClient
     {
-        private readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
+        readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
 
         /// <summary>
         /// The subscriptions are managed in 2 separate buckets:
@@ -27,33 +27,39 @@ namespace MQTTnet.Extensions.ManagedClient
         /// any thread and are therefore mutex'ed. <see cref="_reconnectSubscriptions"/> get sent to the broker
         ///  at reconnect and are solely owned by <see cref="MaintainConnectionAsync"/>.
         /// </summary>
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly HashSet<string> _unsubscriptions = new HashSet<string>();
-        private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
+        readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
+        readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
+        readonly HashSet<string> _unsubscriptions = new HashSet<string>();
+        readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
 
-        private readonly IMqttClient _mqttClient;
-        private readonly IMqttNetScopedLogger _logger;
+        readonly IMqttClient _mqttClient;
+        readonly IMqttNetScopedLogger _logger;
 
-        private readonly AsyncLock _messageQueueLock = new AsyncLock();
+        readonly AsyncLock _messageQueueLock = new AsyncLock();
 
-        private CancellationTokenSource _connectionCancellationToken;
-        private CancellationTokenSource _publishingCancellationToken;
-        private Task _maintainConnectionTask;
+        CancellationTokenSource _connectionCancellationToken;
+        CancellationTokenSource _publishingCancellationToken;
+        Task _maintainConnectionTask;
 
-        private ManagedMqttClientStorageManager _storageManager;
+        ManagedMqttClientStorageManager _storageManager;
 
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
         {
             _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
+            InternalClient = mqttClient;
 
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateScopedLogger(nameof(ManagedMqttClient));
         }
 
         public bool IsConnected => _mqttClient.IsConnected;
+
         public bool IsStarted => _connectionCancellationToken != null;
+        
+        public IMqttClient InternalClient { get; }
+
         public int PendingApplicationMessagesCount => _messageQueue.Count;
+        
         public IManagedMqttClientOptions Options { get; private set; }
 
         public IMqttClientConnectedHandler ConnectedHandler
@@ -88,8 +94,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
             if (options == null) throw new ArgumentNullException(nameof(options));
             if (options.ClientOptions == null) throw new ArgumentException("The client options are not set.", nameof(options));
-
-
+            
             if (!_maintainConnectionTask?.IsCompleted ?? false) throw new InvalidOperationException("The managed client is already started.");
 
             Options = options;
@@ -105,11 +110,16 @@ namespace MQTTnet.Extensions.ManagedClient
                 }
             }
 
-            _connectionCancellationToken = new CancellationTokenSource();
-
-            _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token);
-            _maintainConnectionTask.Forget(_logger);
+            if (Options.AutoReconnect)
+            {
+                var cancellationTokenSource = new CancellationTokenSource();
+                var cancellationToken = cancellationTokenSource.Token;
+                _connectionCancellationToken = cancellationTokenSource;
 
+                _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(cancellationToken), cancellationToken);
+                _maintainConnectionTask.Forget(_logger);
+            }
+            
             _logger.Info("Started");
         }
 
@@ -260,10 +270,11 @@ namespace MQTTnet.Extensions.ManagedClient
                 _mqttClient.Dispose();
                 _subscriptionsQueuedSignal.Dispose();
             }
+
             base.Dispose(disposing);
         }
 
-        private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
+        async Task MaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -303,7 +314,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
+        async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -346,7 +357,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
+        async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -384,7 +395,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
+        async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
             try
@@ -449,7 +460,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             var endTime = DateTime.UtcNow + timeout;
             while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
@@ -501,7 +512,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishReconnectSubscriptionsAsync()
+        async Task PublishReconnectSubscriptionsAsync()
         {
             _logger.Info("Publishing subscriptions at reconnect");
 
@@ -519,7 +530,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task HandleSubscriptionExceptionAsync(Exception exception)
+        async Task HandleSubscriptionExceptionAsync(Exception exception)
         {
             _logger.Warning(exception, "Synchronizing subscriptions failed.");
 
@@ -530,7 +541,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task<ReconnectionResult> ReconnectIfRequiredAsync()
+        async Task<ReconnectionResult> ReconnectIfRequiredAsync()
         {
             if (_mqttClient.IsConnected)
             {
@@ -554,34 +565,47 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private void StartPublishing()
+        void StartPublishing()
         {
             if (_publishingCancellationToken != null)
             {
                 StopPublishing();
             }
 
-            var cts = new CancellationTokenSource();
-            _publishingCancellationToken = cts;
+            var cancellationTokenSource = new CancellationTokenSource();
+            var cancellationToken = cancellationTokenSource.Token;
+            _publishingCancellationToken = cancellationTokenSource;
 
-            Task.Run(() => PublishQueuedMessagesAsync(cts.Token), cts.Token).Forget(_logger);
+            Task.Run(() => PublishQueuedMessagesAsync(cancellationToken), cancellationToken).Forget(_logger);
         }
 
-        private void StopPublishing()
+        void StopPublishing()
         {
-            _publishingCancellationToken?.Cancel(false);
-            _publishingCancellationToken?.Dispose();
-            _publishingCancellationToken = null;
+            try
+            {
+                _publishingCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _publishingCancellationToken?.Dispose();
+                _publishingCancellationToken = null;
+            }
         }
 
-        private void StopMaintainingConnection()
+        void StopMaintainingConnection()
         {
-            _connectionCancellationToken?.Cancel(false);
-            _connectionCancellationToken?.Dispose();
-            _connectionCancellationToken = null;
+            try
+            {
+                _connectionCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _connectionCancellationToken?.Dispose();
+                _connectionCancellationToken = null;
+            }
         }
 
-        private TimeSpan GetRemainingTime(DateTime endTime)
+        static TimeSpan GetRemainingTime(DateTime endTime)
         {
             var remainingTime = endTime - DateTime.UtcNow;
             return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs

@@ -1,5 +1,4 @@
 using System;
-using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
@@ -9,6 +8,8 @@ namespace MQTTnet.Extensions.ManagedClient
     {
         public IMqttClientOptions ClientOptions { get; set; }
 
+        public bool AutoReconnect { get; set; } = true;
+
         public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
 
         public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1);

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs

@@ -1,5 +1,4 @@
 using System;
-using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
@@ -7,8 +6,8 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClientOptionsBuilder
     {
-        private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
-        private MqttClientOptionsBuilder _clientOptionsBuilder;
+        readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
+        MqttClientOptionsBuilder _clientOptionsBuilder;
 
         public ManagedMqttClientOptionsBuilder WithMaxPendingMessages(int value)
         {
@@ -70,6 +69,18 @@ namespace MQTTnet.Extensions.ManagedClient
             return this;
         }
 
+        public ManagedMqttClientOptionsBuilder WithAutoReconnect()
+        {
+            _options.AutoReconnect = true;
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithoutAutoReconnect()
+        {
+            _options.AutoReconnect = false;
+            return this;
+        }
+        
         public ManagedMqttClientOptions Build()
         {
             if (_clientOptionsBuilder != null)

Source/MQTTnet/Client/MqttClient.cs

@@ -245,17 +245,22 @@ namespace MQTTnet.Client
 
         void Cleanup()
         {
-            _backgroundCancellationTokenSource?.Cancel(false);
-            _backgroundCancellationTokenSource?.Dispose();
-            _backgroundCancellationTokenSource = null;
+            try
+            {
+                _backgroundCancellationTokenSource?.Cancel(false);
+            }
+            finally
+            {
+                _backgroundCancellationTokenSource?.Dispose();
+                _backgroundCancellationTokenSource = null;
 
-            _publishPacketReceiverQueue?.Dispose();
-            _publishPacketReceiverQueue = null;
+                _publishPacketReceiverQueue?.Dispose();
+                _publishPacketReceiverQueue = null;
 
-            _adapter?.Dispose();
+                _adapter?.Dispose();
+            }
         }
 
-
         protected override void Dispose(bool disposing)
         {
             if (disposing)
@@ -302,7 +307,7 @@ namespace MQTTnet.Client
 
         async Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
         {
-            var clientWasConnected = IsConnected;
+            var clientWasConnected = _isConnected;
             var reasonCode = MqttClientDisconnectReason.NormalDisconnection;
 
             TryInitiateDisconnect();
@@ -388,7 +393,7 @@ namespace MQTTnet.Client
         async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
         {
             cancellationToken.ThrowIfCancellationRequested();
-            
+
             ushort identifier = 0;
             if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
             {
@@ -437,7 +442,7 @@ namespace MQTTnet.Client
                 {
                     // Values described here: [MQTT-3.1.2-24].
                     var waitTime = keepAlivePeriod - _sendTracker.Elapsed;
-                    
+
                     if (waitTime <= TimeSpan.Zero)
                     {
                         await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
@@ -562,12 +567,15 @@ namespace MQTTnet.Client
                 {
                     await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false);
                 }
-                else if (packet is MqttDisconnectPacket)
+                else if (packet is MqttDisconnectPacket disconnectPacket)
                 {
                     // Also dispatch disconnect to waiting threads to generate a proper exception.
                     _packetDispatcher.Dispatch(packet);
 
-                    await DisconnectAsync(null, cancellationToken).ConfigureAwait(false);
+                    await DisconnectAsync(new MqttClientDisconnectOptions()
+                    {
+                        ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError)
+                    }, cancellationToken).ConfigureAwait(false);
                 }
                 else if (packet is MqttAuthPacket authPacket)
                 {
@@ -629,7 +637,6 @@ namespace MQTTnet.Client
                 try
                 {
                     var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken);
-
                     if (!publishPacketDequeueResult.IsSuccess)
                     {
                         return;

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -280,7 +280,9 @@ namespace MQTTnet.Client.Options
 #pragma warning disable CS0618 // Type or member is obsolete
                         CertificateValidationCallback = _tlsParameters.CertificateValidationCallback,
 #pragma warning restore CS0618 // Type or member is obsolete
-
+#if NETCOREAPP3_1
+                        ApplicationProtocols = _tlsParameters.ApplicationProtocols,
+#endif
                         CertificateValidationHandler = _tlsParameters.CertificateValidationHandler,
                         IgnoreCertificateChainErrors = _tlsParameters.IgnoreCertificateChainErrors,
                         IgnoreCertificateRevocationErrors = _tlsParameters.IgnoreCertificateRevocationErrors

Source/MQTTnet/Client/Options/MqttClientOptionsBuilderTlsParameters.cs

@@ -27,7 +27,11 @@ namespace MQTTnet.Client.Options
         public IEnumerable<X509Certificate> Certificates { get; set; }
 #endif
 
-        public bool AllowUntrustedCertificates { get; set; }
+#if NETCOREAPP3_1
+	    public List<SslApplicationProtocol> ApplicationProtocols { get;set; }
+#endif 
+
+	    public bool AllowUntrustedCertificates { get; set; }
 
         public bool IgnoreCertificateChainErrors { get; set; }
 

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -22,6 +22,10 @@ namespace MQTTnet.Client.Options
         public List<X509Certificate> Certificates { get; set; }
 #endif
 
+#if NETCOREAPP3_1
+        public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
+#endif
+
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
 
         [Obsolete("This property will be removed soon. Use CertificateValidationHandler instead.")]

Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs

@@ -20,6 +20,8 @@
                 return "netstandard2.1";
 #elif WINDOWS_UWP
                 return "uap10.0";
+#elif NETCOREAPP3_1
+                return "netcoreapp3.1";
 #endif
             }
         }

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -81,11 +81,24 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
+#if NETCOREAPP3_1
+                        var sslOptions = new SslClientAuthenticationOptions
+                        {
+                            ApplicationProtocols = _options.TlsOptions.ApplicationProtocols,
+                            ClientCertificates = LoadCertificates(),
+                            EnabledSslProtocols = _options.TlsOptions.SslProtocol,
+                            CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.Online : X509RevocationMode.NoCheck,
+                            TargetHost = _options.Server
+                        };
+
+                        await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
+#else 
                         await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+#endif
                     }
                     catch
                     {
-#if NETSTANDARD2_1
+#if NETSTANDARD2_1 || NETCOREAPP3_1
                         await sslStream.DisposeAsync().ConfigureAwait(false);
 #else
                         sslStream.Dispose();
@@ -209,7 +222,7 @@ namespace MQTTnet.Implementations
 
         bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
         {
-#region OBSOLETE
+            #region OBSOLETE
 
 #pragma warning disable CS0618 // Type or member is obsolete
             var certificateValidationCallback = _options?.TlsOptions?.CertificateValidationCallback;
@@ -218,7 +231,7 @@ namespace MQTTnet.Implementations
             {
                 return certificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _clientOptions);
             }
-#endregion
+            #endregion
 
             var certificateValidationHandler = _options?.TlsOptions?.CertificateValidationHandler;
             if (certificateValidationHandler != null)

Source/MQTTnet/Internal/AsyncQueue.cs

@@ -7,16 +7,15 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncQueue<TItem> : IDisposable
     {
-        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
-
-        private ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
-
+        SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+        ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
+        
         public int Count => _queue.Count;
 
         public void Enqueue(TItem item)
         {
             _queue.Enqueue(item);
-            _semaphore.Release();
+            _semaphore?.Release();
         }
 
         public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -25,12 +24,17 @@ namespace MQTTnet.Internal
             {
                 try
                 {
+                    if (_semaphore == null)
+                    {
+                        return new AsyncQueueDequeueResult<TItem>(false, default);
+                    }
+
                     await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
                     cancellationToken.ThrowIfCancellationRequested();
                 }
                 catch (OperationCanceledException)
                 {
-                    return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+                    return new AsyncQueueDequeueResult<TItem>(false, default);
                 }
 
                 if (_queue.TryDequeue(out var item))
@@ -39,7 +43,7 @@ namespace MQTTnet.Internal
                 }
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+            return new AsyncQueueDequeueResult<TItem>(false, default);
         }
 
         public AsyncQueueDequeueResult<TItem> TryDequeue()
@@ -49,7 +53,7 @@ namespace MQTTnet.Internal
                 return new AsyncQueueDequeueResult<TItem>(true, item);
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+            return new AsyncQueueDequeueResult<TItem>(false, default);
         }
 
         public void Clear()
@@ -60,6 +64,7 @@ namespace MQTTnet.Internal
         public void Dispose()
         {
             _semaphore?.Dispose();
+            _semaphore = null;
         }
     }
 }

Source/MQTTnet/MQTTnet.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1</TargetFrameworks>
+    <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' AND '$(SolutionName)' != 'MQTTnet.noUWP' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
     <AssemblyName>MQTTnet</AssemblyName>

Source/MQTTnet/Server/IMqttServerOptions.cs

@@ -27,5 +27,7 @@ namespace MQTTnet.Server
         IMqttRetainedMessagesManager RetainedMessagesManager { get; }
 
         IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
+
+        IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -23,8 +23,6 @@ namespace MQTTnet.Server
         readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
 
         readonly IMqttRetainedMessagesManager _retainedMessagesManager;
-        readonly Func<Task> _onStart;
-        readonly Func<MqttClientDisconnectType, Task> _onStop;
         readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         readonly MqttClientSessionsManager _sessionsManager;
 
@@ -32,6 +30,7 @@ namespace MQTTnet.Server
         readonly IMqttServerOptions _serverOptions;
 
         readonly IMqttChannelAdapter _channelAdapter;
+        readonly MqttConnectionValidatorContext _connectionValidatorContext;
         readonly IMqttDataConverter _dataConverter;
         readonly string _endpoint;
         readonly DateTime _connectedTimestamp;
@@ -47,25 +46,22 @@ namespace MQTTnet.Server
 
         volatile bool _isTakeover;
 
-        public MqttClientConnection(
-            MqttConnectPacket connectPacket,
+        public MqttClientConnection(MqttConnectPacket connectPacket,
             IMqttChannelAdapter channelAdapter,
             MqttClientSession session,
+            MqttConnectionValidatorContext connectionValidatorContext,
             IMqttServerOptions serverOptions,
             MqttClientSessionsManager sessionsManager,
             IMqttRetainedMessagesManager retainedMessagesManager,
-            Func<Task> onStart,
-            Func<MqttClientDisconnectType, Task> onStop,
             IMqttNetLogger logger)
         {
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
-            _onStart = onStart ?? throw new ArgumentNullException(nameof(onStart));
-            _onStop = onStop ?? throw new ArgumentNullException(nameof(onStop));
 
             _channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
+            _connectionValidatorContext = connectionValidatorContext ?? throw new ArgumentNullException(nameof(connectionValidatorContext));
             _dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter;
             _endpoint = _channelAdapter.Endpoint;
             ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
@@ -131,18 +127,17 @@ namespace MQTTnet.Server
             _cancellationToken.Dispose();
         }
 
-        public Task RunAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        public Task RunAsync()
         {
-            _packageReceiverTask = RunInternalAsync(connectionValidatorContext);
+            _packageReceiverTask = RunInternalAsync();
             return _packageReceiverTask;
         }
 
-        async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        async Task RunInternalAsync()
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             try
             {
-                await _onStart();
                 _logger.Info("Client '{0}': Session started.", ClientId);
 
                 _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
@@ -155,9 +150,7 @@ namespace MQTTnet.Server
                 // TODO: Change to single thread in SessionManager. Or use SessionManager and stats from KeepAliveMonitor.
                 _keepAliveMonitor.Start(ConnectPacket.KeepAlivePeriod, _cancellationToken.Token);
 
-                await SendAsync(
-                    _channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext)
-                    ).ConfigureAwait(false);
+                await SendAsync(_channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(_connectionValidatorContext)).ConfigureAwait(false);
 
                 Session.IsCleanSession = false;
 
@@ -262,17 +255,33 @@ namespace MQTTnet.Server
                 _channelAdapter.ReadingPacketStartedCallback = null;
                 _channelAdapter.ReadingPacketCompletedCallback = null;
 
-                _logger.Info("Client '{0}': Connection stopped.", ClientId);
-
                 _packageReceiverTask = null;
 
+                if (_isTakeover)
+                {
+                    try
+                    {
+                        // Don't use SendAsync here _cancellationToken is already cancelled.
+                        await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket
+                        {
+                            ReasonCode = MqttDisconnectReasonCode.SessionTakenOver
+                        }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
+                    }
+                    catch (Exception exception)
+                    {
+                        _logger.Error(exception, "Client '{0}': Error while sending DISCONNECT packet after takeover.", ClientId);
+                    }
+                }
+
+                _logger.Info("Client '{0}': Connection stopped.", ClientId);
+                
                 try
                 {
-                    await _onStop(disconnectType);
+                    await _sessionsManager.CleanUpClient(ClientId, _channelAdapter, disconnectType);
                 }
                 catch (Exception e)
                 {
-                    _logger.Error(e, "client '{0}': Error while cleaning up", ClientId);
+                    _logger.Error(e, "Client '{0}': Error while cleaning up", ClientId);
                 }
             }
         }
@@ -392,22 +401,18 @@ namespace MQTTnet.Server
                     }
 
                     publishPacket = _dataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
-                    publishPacket.QualityOfServiceLevel = queuedApplicationMessage.QualityOfServiceLevel;
+                    publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel;
 
                     // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
                     publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage;
 
-                    if (publishPacket.QualityOfServiceLevel > 0)
-                    {
-                        publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-                    }
-
                     if (_serverOptions.ClientMessageQueueInterceptor != null)
                     {
                         var context = new MqttClientMessageQueueInterceptorContext(
                             queuedApplicationMessage.SenderClientId,
                             ClientId,
-                            queuedApplicationMessage.ApplicationMessage);
+                            queuedApplicationMessage.ApplicationMessage,
+                            queuedApplicationMessage.SubscriptionQualityOfServiceLevel);
 
                         if (_serverOptions.ClientMessageQueueInterceptor != null)
                         {
@@ -421,7 +426,12 @@ namespace MQTTnet.Server
 
                         publishPacket.Topic = context.ApplicationMessage.Topic;
                         publishPacket.Payload = context.ApplicationMessage.Payload;
-                        publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
+                        publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel;
+                    }
+
+                    if (publishPacket.QualityOfServiceLevel > 0)
+                    {
+                        publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
                     }
 
                     if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)

Source/MQTTnet/Server/MqttClientMessageQueueInterceptorContext.cs

@@ -1,12 +1,15 @@
-namespace MQTTnet.Server
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
 {
     public class MqttClientMessageQueueInterceptorContext
     {
-        public MqttClientMessageQueueInterceptorContext(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage)
+        public MqttClientMessageQueueInterceptorContext(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, MqttQualityOfServiceLevel subscriptionQualityOfServiceLevel)
         {
             SenderClientId = senderClientId;
             ReceiverClientId = receiverClientId;
             ApplicationMessage = applicationMessage;
+            SubscriptionQualityOfServiceLevel = subscriptionQualityOfServiceLevel;
         }
 
         public string SenderClientId { get; }
@@ -16,5 +19,7 @@
         public MqttApplicationMessage ApplicationMessage { get; set; }
 
         public bool AcceptEnqueue { get; set; } = true;
+
+        public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
     }
 }

Source/MQTTnet/Server/MqttClientMessageQueueInterceptorDelegate.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientMessageQueueInterceptorDelegate : IMqttServerClientMessageQueueInterceptor
+    {
+        private readonly Func<MqttClientMessageQueueInterceptorContext, Task> _callback;
+
+        public MqttClientMessageQueueInterceptorDelegate(Action<MqttClientMessageQueueInterceptorContext> callback)
+        {
+            if (callback == null) throw new ArgumentNullException(nameof(callback));
+
+            _callback = context =>
+            {
+                callback(context);
+                return Task.FromResult(0);
+            };
+        }
+
+        public MqttClientMessageQueueInterceptorDelegate(Func<MqttClientMessageQueueInterceptorContext, Task> callback)
+        {
+            _callback = callback ?? throw new ArgumentNullException(nameof(callback));
+        }
+
+        public Task InterceptClientMessageQueueEnqueueAsync(MqttClientMessageQueueInterceptorContext context)
+        {
+            return _callback(context);
+        }
+    }
+}

Source/MQTTnet/Server/MqttClientSessionApplicationMessagesQueue.cs

@@ -27,7 +27,7 @@ namespace MQTTnet.Server
             {
                 ApplicationMessage = applicationMessage,
                 SenderClientId = senderClientId,
-                QualityOfServiceLevel = qualityOfServiceLevel,
+                SubscriptionQualityOfServiceLevel = qualityOfServiceLevel,
                 IsRetainedMessage = isRetainedMessage
             });
         }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -21,6 +21,7 @@ namespace MQTTnet.Server
         readonly AsyncLock _createConnectionGate = new AsyncLock();
         readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
         readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
+
         readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
 
         readonly CancellationToken _cancellationToken;
@@ -183,7 +184,6 @@ namespace MQTTnet.Server
                 }
 
                 var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
-
                 if (!dequeueResult.IsSuccess)
                 {
                     return;
@@ -258,8 +258,6 @@ namespace MQTTnet.Server
 
         async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
-            string clientId = null;
-
             try
             {
                 MqttConnectPacket connectPacket;
@@ -291,17 +289,9 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                clientId = connectPacket.ClientId;
-
-                var connection = await CreateClientConnectionAsync(
-                    connectPacket,
-                    connectionValidatorContext,
-                    channelAdapter,
-                    async () => await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false),
-                    async disconnectType => await CleanUpClient(clientId, channelAdapter, disconnectType)
-                ).ConfigureAwait(false);
-
-                await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false);
+                var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
+                await connection.RunAsync().ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
@@ -312,15 +302,19 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
+        public async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
         {
             if (clientId != null)
             {
-                _connections.TryRemove(clientId, out _);
-
-                if (!_options.EnablePersistentSessions)
+                // in case it is a takeover _connections already contains the new connection
+                if (disconnectType != MqttClientDisconnectType.Takeover)
                 {
-                    await DeleteSessionAsync(clientId).ConfigureAwait(false);
+                    _connections.TryRemove(clientId, out _);
+
+                    if (!_options.EnablePersistentSessions)
+                    {
+                        await DeleteSessionAsync(clientId).ConfigureAwait(false);
+                    }
                 }
             }
 
@@ -360,45 +354,52 @@ namespace MQTTnet.Server
             return context;
         }
 
-        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter, Func<Task> onStart, Func<MqttClientDisconnectType, Task> onStop)
+        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
         {
+            MqttClientConnection connection;
+            MqttClientConnection existingConnection = null;
+
             using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false))
             {
-                var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session);
-
-                var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection);
-                if (isConnectionPresent)
-                {
-                    await existingConnection.StopAsync(true).ConfigureAwait(false);
-                }
-
-                if (isSessionPresent)
-                {
-                    if (connectPacket.CleanSession)
+                var session = _sessions.AddOrUpdate(
+                    connectPacket.ClientId,
+                    key =>
                     {
-                        session = null;
-
-                        _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
-                    }
-                    else
+                        _logger.Verbose("Created a new session for client '{0}'.", key);
+                        return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
+                    },
+                    (key, existingSession) =>
                     {
+                        if (connectPacket.CleanSession)
+                        {
+                            _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
+                            return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
+                        }
+
                         _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
-                    }
-                }
+                        return existingSession;
+                    });
 
-                if (session == null)
-                {
-                    session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
-                    _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
-                }
+                connection = new MqttClientConnection(connectPacket, channelAdapter, session, connectionValidatorContext, _options, this, _retainedMessagesManager, _rootLogger);
 
-                var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger);
-
-                _connections[connection.ClientId] = connection;
-                _sessions[session.ClientId] = session;
-
-                return connection;
+                _connections.AddOrUpdate(
+                    connectPacket.ClientId,
+                    key => connection,
+                    (key, tempExistingConnection) =>
+                    {
+                        existingConnection = tempExistingConnection;
+                        return connection;
+                    });
             }
+
+            // Disconnect the client outside of the lock so that new clients can still connect while a single
+            // one is being disconnected.
+            if (existingConnection != null)
+            {
+                await existingConnection.StopAsync(true).ConfigureAwait(false);
+            }
+
+            return connection;
         }
 
         async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)

Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs

@@ -1,4 +1,5 @@
-using MQTTnet.Protocol;
+using System;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -10,7 +11,14 @@ namespace MQTTnet.Server
 
         public bool IsRetainedMessage { get; set; }
 
-        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+        public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
+
+        [Obsolete("Use 'SubscriptionQualityOfServiceLevel' instead.")]
+        public MqttQualityOfServiceLevel QualityOfServiceLevel
+        {
+            get => SubscriptionQualityOfServiceLevel;
+            set => SubscriptionQualityOfServiceLevel = value;
+        }
 
         public bool IsDuplicate { get; set; }
     }

Source/MQTTnet/Server/MqttServerOptions.cs

@@ -33,5 +33,7 @@ namespace MQTTnet.Server
         public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager();
 
         public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
+
+        public IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -163,6 +163,18 @@ namespace MQTTnet.Server
             return this;
         }
 
+        public MqttServerOptionsBuilder WithDisconnectedInterceptor(IMqttServerClientDisconnectedHandler value)
+        {
+            _options.ClientDisconnectedInterceptor = value;
+            return this;
+        }
+
+        public MqttServerOptionsBuilder WithDisconnectedInterceptor(Action<MqttServerClientDisconnectedEventArgs> value)
+        {
+            _options.ClientDisconnectedInterceptor = new MqttServerClientDisconnectedHandlerDelegate(value);
+            return this;
+        }
+
         public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value)
         {
             _options.ApplicationMessageInterceptor = value;
@@ -175,6 +187,18 @@ namespace MQTTnet.Server
             return this;
         }
 
+        public MqttServerOptionsBuilder WithClientMessageQueueInterceptor(IMqttServerClientMessageQueueInterceptor value)
+        {
+            _options.ClientMessageQueueInterceptor = value;
+            return this;
+        }
+
+        public MqttServerOptionsBuilder WithClientMessageQueueInterceptor(Action<MqttClientMessageQueueInterceptorContext> value)
+        {
+            _options.ClientMessageQueueInterceptor = new MqttClientMessageQueueInterceptorDelegate(value);
+            return this;
+        }
+
         public MqttServerOptionsBuilder WithSubscriptionInterceptor(IMqttServerSubscriptionInterceptor value)
         {
             _options.SubscriptionInterceptor = value;

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -23,6 +23,38 @@ namespace MQTTnet.Tests
     {
         public TestContext TestContext { get; set; }
 
+        [TestMethod]
+        public async Task Set_ClientWasConnected_On_ServerDisconnect()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                var server = await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                Assert.IsTrue(client.IsConnected);
+                client.UseDisconnectedHandler(e => Assert.IsTrue(e.ClientWasConnected));
+
+                await server.StopAsync();
+                await Task.Delay(4000);
+            }
+        }
+
+        [TestMethod]
+        public async Task Set_ClientWasConnected_On_ClientDisconnect()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                var server = await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                Assert.IsTrue(client.IsConnected);
+                client.UseDisconnectedHandler(e => Assert.IsTrue(e.ClientWasConnected));
+
+                await client.DisconnectAsync();
+                await Task.Delay(200);
+            }
+        }
+
         [TestMethod]
         [ExpectedException(typeof(MqttCommunicationTimedOutException))]
         public async Task Connect_To_Invalid_Server_Wrong_IP()

Tests/MQTTnet.Core.Tests/Server_Tests.cs

@@ -1415,5 +1415,68 @@ namespace MQTTnet.Tests
                 Assert.AreEqual("a", undeliverd);
             }
         }
+
+        [TestMethod]
+        public async Task Intercept_ClientMessageQueue()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()
+                    .WithClientMessageQueueInterceptor(c => c.ApplicationMessage.Topic = "a"));
+
+                var topicAReceived = false;
+                var topicBReceived = false;
+
+                var client = await testEnvironment.ConnectClientAsync();
+                client.UseApplicationMessageReceivedHandler(c =>
+                {
+                    if (c.ApplicationMessage.Topic == "a")
+                    {
+                        topicAReceived = true;
+                    }
+                    else if (c.ApplicationMessage.Topic == "b")
+                    {
+                        topicBReceived = true;
+                    }
+                });
+
+                await client.SubscribeAsync("a");
+                await client.SubscribeAsync("b");
+
+                await client.PublishAsync("b");
+
+                await Task.Delay(500);
+
+                Assert.IsTrue(topicAReceived);
+                Assert.IsFalse(topicBReceived);
+            }
+        }
+
+        [TestMethod]
+        public async Task Intercept_ClientMessageQueue_Different_QoS_Of_Subscription_And_Message()
+        {
+            const string topic = "a";
+
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()
+                    .WithClientMessageQueueInterceptor(c => { })); // Interceptor does nothing but has to be present.
+
+                bool receivedMessage = false;
+                var client = await testEnvironment.ConnectClientAsync();
+                client.UseApplicationMessageReceivedHandler(c =>
+                {
+                    receivedMessage = true;
+                });
+
+                await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
+
+                await client.PublishAsync(new MqttApplicationMessage{ Topic = topic, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+
+                await Task.Delay(500);
+
+                Assert.IsTrue(receivedMessage);
+            }
+        }
     }
 }

Tests/MQTTnet.Core.Tests/Session_Tests.cs

@@ -1,5 +1,6 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -84,5 +85,40 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(true, session.Items["can_subscribe_x"]);
             }
         }
+
+
+        [TestMethod]
+        public async Task Manage_Session_MaxParallel()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                var serverOptions = new MqttServerOptionsBuilder();
+                await testEnvironment.StartServerAsync(serverOptions);
+
+                var options = new MqttClientOptionsBuilder()
+                    .WithClientId("1")
+                    ;
+
+                var clients = await Task.WhenAll(Enumerable.Range(0, 10)
+                    .Select(i => TryConnect(testEnvironment, options)));
+
+                var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
+
+                Assert.AreEqual(1, connectedClients.Count);
+            }
+        }
+
+        private async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
+        {
+            try
+            {
+                return await testEnvironment.ConnectClientAsync(options);
+            }
+            catch (System.Exception)
+            {
+                return null;
+            }
+        }
     }
 }