1889 no service for type mqttnetservermqttserver has been registered (#1892)

* Add missing service to service collection

* Update ReleaseNotes.md

* Restore service registrations

.github/workflows/ReleaseNotes.md

@@ -1,11 +1,3 @@
-* [Client] Fixed _PlatformNotSupportedException_ when using Blazor (#1755, thanks to @Nickztar).
-* [Client] Added hot reload of client certificates (#1781).
-* [Client] Added several new option builders and aligned usage (#1781, BREAKING CHANGE!).
-* [Client] Added support for _RemoteCertificateValidationCallback_ for .NET 4.5.2, 4.6.1 and 4.8 (#1806, thanks to @troky).
-* [Client] Fixed wrong logging of obsolete feature when connection was not successful (#1801, thanks to @ramonsmits).
-* [Client] Fixed _NullReferenceException_ when performing several actions when not connected (#1800, thanks to @ramonsmits).
-* [RpcClient] Added support for passing custom parameters to topic generation context (#1798, thanks to @Temppus).
-* [Server] Fixed _NullReferenceException_ in retained messages management (#1762, thanks to @logicaloud).
-* [Server] Exposed new option which allows disabling packet fragmentation (#1753).
-* [Server] Expired sessions will no longer be used when a client connects (#1756).
-* [Server] Fixed an issue in connection handling for ASP.NET connections (#1819, thanks to @CZEMacLeod).
+* [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
+* [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883).
+* [Server] Fixed "service not registered" exception in ASP.NET integration (#1889).

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.3.0.${{github.run_number}}"
+  VERSION: "4.3.3.${{github.run_number}}"
 
 jobs:
   build:
@@ -54,7 +54,7 @@ jobs:
         uses: darenm/Setup-VSTest@v1
 
       - name: Core Tests
-        run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net6.0\MQTTnet.Tests.dll
+        run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net7.0\MQTTnet.Tests.dll
 
       - name: ASP.NET Tests
         run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll

MQTTnet.sln.DotSettings

@@ -240,6 +240,7 @@ See the LICENSE file in the project root for more information.</s:String>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=catched/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=CONNACK/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>

Samples/Client/Client_Connection_Samples.cs

@@ -7,6 +7,7 @@
 // ReSharper disable InconsistentNaming
 
 using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
 using MQTTnet.Client;
 using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
@@ -430,6 +431,59 @@ public static class Client_Connection_Samples
                         }
                     }
                 });
+
+            Console.WriteLine("Press <Enter> to exit");
+            Console.ReadLine();
         }
     }
+
+    public static async Task ConnectTls_WithCaFile()
+    {
+        var mqttFactory = new MqttFactory();
+
+        X509Certificate2Collection caChain = new X509Certificate2Collection();
+        caChain.ImportFromPem(mosquitto_org); // from https://test.mosquitto.org/ssl/mosquitto.org.crt
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("test.mosquitto.org", 8883)
+                .WithTlsOptions(new MqttClientTlsOptionsBuilder()
+                    .WithTrustChain(caChain) 
+                    .Build())
+                .Build();
+
+            var connAck = await mqttClient.ConnectAsync(mqttClientOptions);
+            Console.WriteLine("Connected to test.moquitto.org:8883 with CaFile mosquitto.org.crt: " + connAck.ResultCode);
+        }
+
+
+    }
+    const string mosquitto_org = @"
+-----BEGIN CERTIFICATE-----
+MIIEAzCCAuugAwIBAgIUBY1hlCGvdj4NhBXkZ/uLUZNILAwwDQYJKoZIhvcNAQEL
+BQAwgZAxCzAJBgNVBAYTAkdCMRcwFQYDVQQIDA5Vbml0ZWQgS2luZ2RvbTEOMAwG
+A1UEBwwFRGVyYnkxEjAQBgNVBAoMCU1vc3F1aXR0bzELMAkGA1UECwwCQ0ExFjAU
+BgNVBAMMDW1vc3F1aXR0by5vcmcxHzAdBgkqhkiG9w0BCQEWEHJvZ2VyQGF0Y2hv
+by5vcmcwHhcNMjAwNjA5MTEwNjM5WhcNMzAwNjA3MTEwNjM5WjCBkDELMAkGA1UE
+BhMCR0IxFzAVBgNVBAgMDlVuaXRlZCBLaW5nZG9tMQ4wDAYDVQQHDAVEZXJieTES
+MBAGA1UECgwJTW9zcXVpdHRvMQswCQYDVQQLDAJDQTEWMBQGA1UEAwwNbW9zcXVp
+dHRvLm9yZzEfMB0GCSqGSIb3DQEJARYQcm9nZXJAYXRjaG9vLm9yZzCCASIwDQYJ
+KoZIhvcNAQEBBQADggEPADCCAQoCggEBAME0HKmIzfTOwkKLT3THHe+ObdizamPg
+UZmD64Tf3zJdNeYGYn4CEXbyP6fy3tWc8S2boW6dzrH8SdFf9uo320GJA9B7U1FW
+Te3xda/Lm3JFfaHjkWw7jBwcauQZjpGINHapHRlpiCZsquAthOgxW9SgDgYlGzEA
+s06pkEFiMw+qDfLo/sxFKB6vQlFekMeCymjLCbNwPJyqyhFmPWwio/PDMruBTzPH
+3cioBnrJWKXc3OjXdLGFJOfj7pP0j/dr2LH72eSvv3PQQFl90CZPFhrCUcRHSSxo
+E6yjGOdnz7f6PveLIB574kQORwt8ePn0yidrTC1ictikED3nHYhMUOUCAwEAAaNT
+MFEwHQYDVR0OBBYEFPVV6xBUFPiGKDyo5V3+Hbh4N9YSMB8GA1UdIwQYMBaAFPVV
+6xBUFPiGKDyo5V3+Hbh4N9YSMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEL
+BQADggEBAGa9kS21N70ThM6/Hj9D7mbVxKLBjVWe2TPsGfbl3rEDfZ+OKRZ2j6AC
+6r7jb4TZO3dzF2p6dgbrlU71Y/4K0TdzIjRj3cQ3KSm41JvUQ0hZ/c04iGDg/xWf
++pp58nfPAYwuerruPNWmlStWAXf0UTqRtg4hQDWBuUFDJTuWuuBvEXudz74eh/wK
+sMwfu1HFvjy5Z0iMDU8PUDepjVolOCue9ashlS4EB5IECdSR2TItnAIiIwimx839
+LdUdRudafMu5T5Xma182OC0/u/xRlEm+tvKGGmfFcN0piqVl8OrSPBgIlb+1IKJE
+m/XriWr/Cq4h/JfB7NTsezVslgkBaoU=
+-----END CERTIFICATE-----
+";
+
 }

Samples/Client/Client_Subscribe_Samples.cs

@@ -184,7 +184,7 @@ public static class Client_Subscribe_Samples
          * This sample shows how to achieve concurrent processing and not have message AutoAcknowledged
          * This to have a proper QoS1 (at-least-once) experience for what at least MQTT specification can provide
          */
-        mqttClient.ApplicationMessageReceivedAsync += async ea =>
+        mqttClient.ApplicationMessageReceivedAsync += ea =>
         {
             ea.AutoAcknowledge = false;
 
@@ -201,6 +201,8 @@ public static class Client_Subscribe_Samples
             }
 
             _ = Task.Run(ProcessAsync, shutdownToken);
+
+            return Task.CompletedTask;
         };
     }
 

Samples/MQTTnet.Samples.csproj

@@ -2,7 +2,7 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <TargetFramework>net6.0</TargetFramework>
+        <TargetFramework>net7.0</TargetFramework>
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>enable</Nullable>
         <IsPackable>false</IsPackable>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -83,8 +83,6 @@ namespace MQTTnet.AspNetCore
             }
         }
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection
         {
             get
@@ -164,11 +162,6 @@ namespace MQTTnet.AspNetCore
                                 BytesReceived += received;
                                 return packet;
                             }
-                            else
-                            {
-                                // we did receive something but the message is not yet complete
-                                IsReadingPacket = true;
-                            }
                         }
                         else if (readResult.IsCompleted)
                         {
@@ -189,12 +182,9 @@ namespace MQTTnet.AspNetCore
                 // completing the channel makes sure that there is no more data read after a protocol error
                 _input?.Complete(exception);
                 _output?.Complete(exception);
+
                 throw;
             }
-            finally
-            {
-                IsReadingPacket = false;
-            }
 
             cancellationToken.ThrowIfCancellationRequested();
             return null;

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -14,20 +15,27 @@ namespace MQTTnet.AspNetCore
 {
     public sealed class MqttHostedServer : MqttServer, IHostedService
     {
-        public MqttHostedServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) 
-            : base(options, adapters, logger)
+        readonly MqttFactory _mqttFactory;
+
+        public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
+            options,
+            adapters,
+            logger)
         {
+            _mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
         }
 
-        public Task StartAsync(CancellationToken cancellationToken)
+        public async Task StartAsync(CancellationToken cancellationToken)
         {
-            _ =  StartAsync();
-            return Task.CompletedTask;
+            // The yield makes sure that the hosted service is considered up and running.
+            await Task.Yield();
+
+            _ = StartAsync();
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            return StopAsync();
+            return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
         }
     }
 }

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -4,6 +4,7 @@
 
 using System;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
@@ -14,86 +15,74 @@ namespace MQTTnet.AspNetCore
 {
     public static class ServiceCollectionExtensions
     {
-        public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null)
-        {
-            if (serviceCollection is null)
-            {
-                throw new ArgumentNullException(nameof(serviceCollection));
-            }
-
-            serviceCollection.AddMqttConnectionHandler();
-            serviceCollection.AddHostedMqttServer(configure);
-            
-            return serviceCollection;
-        }
-
         public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
         {
-            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (services == null)
+            {
+                throw new ArgumentNullException(nameof(services));
+            }
+
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
 
             services.AddSingleton(options);
-
             services.AddHostedMqttServer();
 
             return services;
         }
 
-        public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure = null)
+        public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure)
         {
-            services.AddSingleton(s =>
+            if (services == null)
             {
-                var serverOptionsBuilder = new MqttServerOptionsBuilder();
-                configure?.Invoke(serverOptionsBuilder);
-                return serverOptionsBuilder.Build();
-            });
+                throw new ArgumentNullException(nameof(services));
+            }
 
-            services.AddHostedMqttServer();
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
 
-            return services;
+            var serverOptionsBuilder = new MqttServerOptionsBuilder();
+            configure.Invoke(serverOptionsBuilder);
+            var options = serverOptionsBuilder.Build();
+
+            return AddHostedMqttServer(services, options);
+        }
+
+        public static void AddHostedMqttServer(this IServiceCollection services)
+        {
+            // The user may have these services already registered.
+            services.TryAddSingleton<IMqttNetLogger>(MqttNetNullLogger.Instance);
+            services.TryAddSingleton(new MqttFactory());
+
+            services.AddSingleton<MqttHostedServer>();
+            services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
+            services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
         }
 
         public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
         {
-            services.AddSingleton(s =>
+            if (services == null)
             {
-                var builder = new AspNetMqttServerOptionsBuilder(s);
-                configure(builder);
-                return builder.Build();
-            });
+                throw new ArgumentNullException(nameof(services));
+            }
+
+            services.AddSingleton(
+                s =>
+                {
+                    var builder = new AspNetMqttServerOptionsBuilder(s);
+                    configure(builder);
+                    return builder.Build();
+                });
 
             services.AddHostedMqttServer();
 
             return services;
         }
 
-        static IServiceCollection AddHostedMqttServer(this IServiceCollection services)
-        {
-            var logger = new MqttNetEventLogger();
-
-            services.AddSingleton<IMqttNetLogger>(logger);
-            services.AddSingleton<MqttHostedServer>();
-            services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
-            services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
-
-            return services;
-        }
-
-        public static IServiceCollection AddMqttWebSocketServerAdapter(this IServiceCollection services)
-        {
-            services.AddSingleton<MqttWebSocketServerAdapter>();
-            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
-
-            return services;
-        }
-
-        public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
-        {
-            services.AddSingleton<MqttTcpServerAdapter>();
-            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
-
-            return services;
-        }
-
         public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services)
         {
             services.AddSingleton<MqttConnectionHandler>();
@@ -101,5 +90,44 @@ namespace MQTTnet.AspNetCore
 
             return services;
         }
+
+        public static void AddMqttLogger(this IServiceCollection services, IMqttNetLogger logger)
+        {
+            if (services == null)
+            {
+                throw new ArgumentNullException(nameof(services));
+            }
+
+            services.AddSingleton(logger);
+        }
+
+        public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null)
+        {
+            if (serviceCollection is null)
+            {
+                throw new ArgumentNullException(nameof(serviceCollection));
+            }
+
+            serviceCollection.AddMqttConnectionHandler();
+            serviceCollection.AddHostedMqttServer(configure);
+
+            return serviceCollection;
+        }
+
+        public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
+        {
+            services.AddSingleton<MqttTcpServerAdapter>();
+            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
+
+            return services;
+        }
+
+        public static IServiceCollection AddMqttWebSocketServerAdapter(this IServiceCollection services)
+        {
+            services.AddSingleton<MqttWebSocketServerAdapter>();
+            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
+
+            return services;
+        }
     }
-}
+}

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -47,13 +47,13 @@ namespace MQTTnet.Extensions.WebSocket4Net
             var uri = _webSocketOptions.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
             {
-                if (_webSocketOptions.TlsOptions?.UseTls == false)
+                if (_webSocketOptions.TlsOptions?.UseTls == true)
                 {
-                    uri = "ws://" + uri;
+                    uri = "wss://" + uri;
                 }
                 else
                 {
-                    uri = "wss://" + uri;
+                    uri = "ws://" + uri;
                 }
             }
             

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -25,7 +25,9 @@ namespace MQTTnet.TestApp
                 UseTls = true,
                 SslProtocol = SslProtocols.Tls13,
                 // Don't use this in production code. This handler simply allows any invalid certificate to work.
-                CertificateValidationHandler = w => true
+                AllowUntrustedCertificates = true,
+                IgnoreCertificateChainErrors = true,
+                CertificateValidationHandler = _ => true
             };
 #endif
             // Also defining TLS12 for servers that don't seem no to support TLS13.
@@ -34,7 +36,9 @@ namespace MQTTnet.TestApp
                 UseTls = true,
                 SslProtocol = SslProtocols.Tls12,
                 // Don't use this in production code. This handler simply allows any invalid certificate to work.
-                CertificateValidationHandler = w => true
+                AllowUntrustedCertificates = true,
+                IgnoreCertificateChainErrors = true,
+                CertificateValidationHandler = _ => true
             };
 
             // mqtt.eclipseprojects.io
@@ -97,10 +101,10 @@ namespace MQTTnet.TestApp
                 "test.mosquitto.org WS TLS12",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            // await ExecuteTestAsync(
-            //     "test.mosquitto.org WS TLS12 (WebSocket4Net)",
-            //     new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build(),
-            //     true);
+            await ExecuteTestAsync(
+                "test.mosquitto.org WS TLS12 (WebSocket4Net)",
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
+                true);
 
             // broker.emqx.io
             await ExecuteTestAsync(
@@ -150,7 +154,6 @@ namespace MQTTnet.TestApp
                 true);
 
             // mqtt.swifitch.cz: Does not seem to operate any more
-
             // cloudmqtt.com: Cannot test because it does not offer a free plan any more.
 
             Write("Finished.", ConsoleColor.White);
@@ -197,9 +200,10 @@ namespace MQTTnet.TestApp
 
                 Write("[OK]\n", ConsoleColor.Green);
             }
-            catch (Exception e)
+            catch (Exception exception)
             {
-                Write("[FAILED] " + e.Message + "\n", ConsoleColor.Red);
+                Write("[FAILED]" + Environment.NewLine, ConsoleColor.Red);
+                Write(exception + Environment.NewLine, ConsoleColor.Red);
             }
         }
 

Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs

@@ -13,6 +13,7 @@ using MQTTnet.Exceptions;
 using MQTTnet.LowLevelClient;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Tests.Clients.LowLevelMqttClient
 {

Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs

@@ -22,99 +22,6 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
     [TestClass]
     public sealed class ManagedMqttClient_Tests : BaseTestClass
     {
-        [TestMethod]
-        public async Task Expose_Custom_Connection_Error()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var server = await testEnvironment.StartServer();
-                
-                server.ValidatingConnectionAsync += args =>
-                {
-                    args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
-                    return CompletedTask.Instance;
-                };
-
-                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
-
-                MqttClientDisconnectedEventArgs disconnectedArgs = null;
-                managedClient.DisconnectedAsync += args =>
-                {
-                    disconnectedArgs = args;
-                    return CompletedTask.Instance;
-                };
-                
-                var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
-                await managedClient.StartAsync(clientOptions);
-
-                await LongTestDelay();
-
-                Assert.IsNotNull(disconnectedArgs);
-                Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Receive_While_Not_Cleanly_Disconnected()
-        {
-            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-            {
-                await testEnvironment.StartServer(o => o.WithPersistentSessions());
-
-                var senderClient = await testEnvironment.ConnectClient();
-
-                // Prepare managed client.
-                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
-                await managedClient.SubscribeAsync("#");
-                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
-                
-                var managedClientOptions = new ManagedMqttClientOptions
-                {
-                    ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
-                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                        .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
-                        .WithCleanSession(false)
-                        .Build()
-                };
-
-                await managedClient.StartAsync(managedClientOptions);
-                await LongTestDelay();
-                await LongTestDelay();
-                
-                // Send test data.
-                await senderClient.PublishStringAsync("topic1");
-                await LongTestDelay();
-                await LongTestDelay();
-                
-                receivedMessages.AssertReceivedCountEquals(1);
-
-                // Stop the managed client but keep session at server (not clean disconnect required).
-                await managedClient.StopAsync(false);
-                await LongTestDelay();
-                
-                // Send new messages in the meantime.
-                await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
-                await LongTestDelay();
-                
-                // Start the managed client, it should receive the new message.
-                await managedClient.StartAsync(managedClientOptions);
-                await LongTestDelay();
-                
-                receivedMessages.AssertReceivedCountEquals(2);
-                
-                // Stop and start again, no new message should be received.
-                for (var i = 0; i < 3; i++)
-                {
-                    await managedClient.StopAsync(false);
-                    await LongTestDelay();
-                    await managedClient.StartAsync(managedClientOptions);
-                    await LongTestDelay();
-                }
-
-                receivedMessages.AssertReceivedCountEquals(2);
-            }
-        }
-
         [TestMethod]
         public async Task Connect_To_Invalid_Server()
         {
@@ -181,6 +88,38 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+        [TestMethod]
+        public async Task Expose_Custom_Connection_Error()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                server.ValidatingConnectionAsync += args =>
+                {
+                    args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
+                    return CompletedTask.Instance;
+                };
+
+                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
+
+                MqttClientDisconnectedEventArgs disconnectedArgs = null;
+                managedClient.DisconnectedAsync += args =>
+                {
+                    disconnectedArgs = args;
+                    return CompletedTask.Instance;
+                };
+
+                var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
+                await managedClient.StartAsync(clientOptions);
+
+                await LongTestDelay();
+
+                Assert.IsNotNull(disconnectedArgs);
+                Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
+            }
+        }
+
         [TestMethod]
         public async Task ManagedClients_Will_Message_Send()
         {
@@ -224,6 +163,67 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+        [TestMethod]
+        public async Task Receive_While_Not_Cleanly_Disconnected()
+        {
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                var senderClient = await testEnvironment.ConnectClient();
+
+                // Prepare managed client.
+                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
+                await managedClient.SubscribeAsync("#");
+                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
+
+                var managedClientOptions = new ManagedMqttClientOptions
+                {
+                    ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
+                        .WithCleanSession(false)
+                        .Build()
+                };
+
+                await managedClient.StartAsync(managedClientOptions);
+                await LongTestDelay();
+                await LongTestDelay();
+
+                // Send test data.
+                await senderClient.PublishStringAsync("topic1");
+                await LongTestDelay();
+                await LongTestDelay();
+
+                receivedMessages.AssertReceivedCountEquals(1);
+
+                // Stop the managed client but keep session at server (not clean disconnect required).
+                await managedClient.StopAsync(false);
+                await LongTestDelay();
+
+                // Send new messages in the meantime.
+                await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
+                await LongTestDelay();
+
+                // Start the managed client, it should receive the new message.
+                await managedClient.StartAsync(managedClientOptions);
+                await LongTestDelay();
+
+                receivedMessages.AssertReceivedCountEquals(2);
+
+                // Stop and start again, no new message should be received.
+                for (var i = 0; i < 3; i++)
+                {
+                    await managedClient.StopAsync(false);
+                    await LongTestDelay();
+                    await managedClient.StartAsync(managedClientOptions);
+                    await LongTestDelay();
+                }
+
+                receivedMessages.AssertReceivedCountEquals(2);
+            }
+        }
+
         [TestMethod]
         public async Task Start_Stop()
         {
@@ -375,7 +375,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort);
 
                 var receivedManagedMessages = new List<MqttApplicationMessage>();
-                
+
                 var managedClient = testEnvironment.Factory.CreateManagedMqttClient(testEnvironment.CreateClient());
                 managedClient.ApplicationMessageReceivedAsync += e =>
                 {
@@ -403,7 +403,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 // Make sure that it gets received after subscribing again.
                 await managedClient.SubscribeAsync("topic");
                 await LongTestDelay();
-                
+
                 Assert.AreEqual(2, receivedManagedMessages.Count);
             }
         }
@@ -421,7 +421,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 var receivingClient = await CreateManagedClientAsync(testEnvironment, null, connectionCheckInterval);
                 var sendingClient = await testEnvironment.ConnectClient();
 
-                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>( new byte[] { 1 }), Retain = true });
+                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>(new byte[] { 1 }), Retain = true });
 
                 var subscribeTime = DateTime.UtcNow;
 

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs

@@ -167,6 +167,38 @@ namespace MQTTnet.Tests.Clients.MqttClient
             }
         }
 
+        [TestMethod]
+        public async Task No_Unobserved_Exception()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                var client = testEnvironment.CreateClient();
+                var options = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithTimeout(TimeSpan.FromSeconds(2)).Build();
+
+                try
+                {
+                    using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(0.5)))
+                    {
+                        await client.ConnectAsync(options, timeout.Token);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                }
+
+                client.Dispose();
+
+                // These delays and GC calls are required in order to make calling the finalizer reproducible.
+                GC.Collect();
+                GC.WaitForPendingFinalizers();
+                await LongTestDelay();
+                await LongTestDelay();
+                await LongTestDelay();
+            }
+        }
+
         [TestMethod]
         public async Task Return_Non_Success()
         {

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs

@@ -17,6 +17,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
 
 // ReSharper disable InconsistentNaming

Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs

@@ -52,7 +52,7 @@ namespace MQTTnet.Tests.Extensions
             var paramValue = "123";
             var parameters = new Dictionary<string, object>
             {
-                { TestParametersTopicGenerationStrategy.ExpectedParamName, "123" },
+                { TestParametersTopicGenerationStrategy.ExpectedParamName, "123" }
             };
 
             using (var testEnvironment = CreateTestEnvironment())
@@ -164,7 +164,7 @@ namespace MQTTnet.Tests.Extensions
 
                 using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
                 {
-                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+                    await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
                 }
             }
         }

Source/MQTTnet.Tests/MQTTnet.Tests.csproj

@@ -1,7 +1,8 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
-        <TargetFrameworks>net452;net461;net48;netcoreapp3.1;net6.0;net7.0</TargetFrameworks>
+        <TargetFrameworks>netcoreapp3.1;net7.0</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net48</TargetFrameworks>
         <IsPackable>false</IsPackable>
         <LangVersion>7.3</LangVersion>
         <EnableNETAnalyzers>false</EnableNETAnalyzers>

Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs

@@ -273,7 +273,7 @@ namespace MQTTnet.Tests.MQTTv5
                 Assert.AreEqual(applicationMessage.ResponseTopic, receivedMessage.ResponseTopic);
                 Assert.AreEqual(applicationMessage.MessageExpiryInterval, receivedMessage.MessageExpiryInterval);
                 CollectionAssert.AreEqual(applicationMessage.CorrelationData, receivedMessage.CorrelationData);
-                CollectionAssert.AreEqual(applicationMessage.Payload, receivedMessage.Payload);
+                CollectionAssert.AreEqual(applicationMessage.PayloadSegment.ToArray(), receivedMessage.PayloadSegment.ToArray());
                 CollectionAssert.AreEqual(applicationMessage.UserProperties, receivedMessage.UserProperties);
             }
         }

Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs

@@ -10,6 +10,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Tests.MQTTv5
 {

Source/MQTTnet.Tests/Mockups/TestEnvironment.cs

@@ -3,7 +3,6 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -25,9 +24,9 @@ namespace MQTTnet.Tests.Mockups
     public sealed class TestEnvironment : IDisposable
     {
         readonly List<string> _clientErrors = new List<string>();
-        readonly ConcurrentBag<IMqttClient> _clients = new ConcurrentBag<IMqttClient>();
+        readonly List<IMqttClient> _clients = new List<IMqttClient>();
         readonly List<Exception> _exceptions = new List<Exception>();
-        readonly ConcurrentBag<ILowLevelMqttClient> _lowLevelClients = new ConcurrentBag<ILowLevelMqttClient>();
+        readonly List<ILowLevelMqttClient> _lowLevelClients = new List<ILowLevelMqttClient>();
         readonly MqttProtocolVersion _protocolVersion;
         readonly List<string> _serverErrors = new List<string>();
 
@@ -40,6 +39,8 @@ namespace MQTTnet.Tests.Mockups
             _protocolVersion = protocolVersion;
             TestContext = testContext;
 
+            TaskScheduler.UnobservedTaskException += TrackUnobservedTaskException;
+
             ServerLogger.LogMessagePublished += (s, e) =>
             {
                 if (Debugger.IsAttached)
@@ -216,8 +217,9 @@ namespace MQTTnet.Tests.Mockups
         public IMqttClient CreateClient()
         {
             var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
-            
+
             var client = Factory.CreateMqttClient(logger);
+
             client.ConnectingAsync += e =>
             {
                 if (TestContext != null)
@@ -232,9 +234,12 @@ namespace MQTTnet.Tests.Mockups
 
                 return CompletedTask.Instance;
             };
-            
-            _clients.Add(client);
-            
+
+            lock (_clients)
+            {
+                _clients.Add(client);
+            }
+
             return client;
         }
 
@@ -245,13 +250,20 @@ namespace MQTTnet.Tests.Mockups
 
         public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder()
         {
-            return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort).WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
+            return Factory.CreateClientOptionsBuilder()
+                .WithProtocolVersion(_protocolVersion)
+                .WithTcpServer("127.0.0.1", ServerPort)
+                .WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
         }
 
         public ILowLevelMqttClient CreateLowLevelClient()
         {
             var client = Factory.CreateLowLevelMqttClient(ClientLogger);
-            _lowLevelClients.Add(client);
+
+            lock (_lowLevelClients)
+            {
+                _lowLevelClients.Add(client);
+            }
 
             return client;
         }
@@ -287,45 +299,68 @@ namespace MQTTnet.Tests.Mockups
 
         public void Dispose()
         {
-            foreach (var mqttClient in _clients)
+            try
             {
+                lock (_clients)
+                {
+                    foreach (var mqttClient in _clients)
+                    {
+                        try
+                        {
+                            //mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+                        }
+                        catch
+                        {
+                            // This can happen when the test already disconnected the client.
+                        }
+                        finally
+                        {
+                            mqttClient?.Dispose();
+                        }
+                    }
+
+                    _clients.Clear();
+                }
+
+                lock (_lowLevelClients)
+                {
+                    foreach (var lowLevelMqttClient in _lowLevelClients)
+                    {
+                        lowLevelMqttClient.Dispose();
+                    }
+
+                    _lowLevelClients.Clear();
+                }
+
                 try
                 {
-                    //mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+                    Server?.StopAsync().GetAwaiter().GetResult();
                 }
                 catch
                 {
-                    // This can happen when the test already disconnected the client.
+                    // This can happen when the test already stopped the server.
                 }
                 finally
                 {
-                    mqttClient?.Dispose();
+                    Server?.Dispose();
                 }
-            }
 
-            foreach (var lowLevelMqttClient in _lowLevelClients)
-            {
-                lowLevelMqttClient.Dispose();
-            }
+                Server = null;
 
-            try
-            {
-                Server?.StopAsync().GetAwaiter().GetResult();
-            }
-            catch
-            {
-                // This can happen when the test already stopped the server.
+                ThrowIfLogErrors();
+
+                GC.Collect();
+                GC.WaitForFullGCComplete();
+                GC.WaitForPendingFinalizers();
+                
+                if (_exceptions.Any())
+                {
+                    throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
+                }
             }
             finally
             {
-                Server?.Dispose();
-            }
-
-            ThrowIfLogErrors();
-
-            if (_exceptions.Any())
-            {
-                throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
+                TaskScheduler.UnobservedTaskException -= TrackUnobservedTaskException;
             }
         }
 
@@ -342,7 +377,7 @@ namespace MQTTnet.Tests.Mockups
 
             var options = optionsBuilder.Build();
             var server = CreateServer(options);
-            await server.StartAsync();
+            await server.StartAsync().ConfigureAwait(false);
 
             // The OS has chosen the port to we have to properly expose it to the tests.
             ServerPort = options.DefaultEndpointOptions.Port;
@@ -370,13 +405,16 @@ namespace MQTTnet.Tests.Mockups
 
         public void ThrowIfLogErrors()
         {
-            lock (_serverErrors)
+            if (!IgnoreServerLogErrors)
             {
-                if (!IgnoreServerLogErrors && _serverErrors.Count > 0)
+                lock (_serverErrors)
                 {
-                    var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
-                    Console.WriteLine(message);
-                    throw new Exception(message);
+                    if (_serverErrors.Count > 0)
+                    {
+                        var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
+                        Console.WriteLine(message);
+                        throw new Exception(message);
+                    }
                 }
             }
 
@@ -396,10 +434,20 @@ namespace MQTTnet.Tests.Mockups
 
         public void TrackException(Exception exception)
         {
+            if (exception == null)
+            {
+                return;
+            }
+
             lock (_exceptions)
             {
                 _exceptions.Add(exception);
             }
         }
+
+        void TrackUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e)
+        {
+            TrackException(e.Exception);
+        }
     }
 }

Source/MQTTnet.Tests/MqttApplicationMessageTest.cs

@@ -1,3 +1,5 @@
+#pragma warning disable CS0618 // Type or member is obsolete
+
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using System;
 using System.Collections.Generic;
@@ -8,7 +10,7 @@ using System.Threading.Tasks;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttApplicationMessageTest
+    public sealed class MqttApplicationMessageTest
     {
         [TestMethod]
         public void PayloadSegment()
@@ -17,32 +19,27 @@ namespace MQTTnet.Tests
             Assert.AreEqual(0, message.Payload.Length);
             Assert.AreEqual(0, message.PayloadSegment.Count);
 
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.Payload = new byte[] { 1, 2 };
             Assert.AreEqual(2, message.Payload.Length);
             Assert.AreEqual(2, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.Payload = new byte[] { 1, 2, 3 };
             Assert.AreEqual(3, message.Payload.Length);
             Assert.AreEqual(3, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 });
             Assert.AreEqual(3, message.Payload.Length);
             Assert.AreEqual(3, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 }, 1, 1);
             Assert.AreEqual(1, message.Payload.Length);
             Assert.AreEqual(1, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsFalse(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsFalse(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
         }
     }
 }

Source/MQTTnet.Tests/Server/General.cs

@@ -303,7 +303,7 @@ namespace MQTTnet.Tests.Server
                 var server = await testEnvironment.StartServer();
                 server.InterceptingPublishAsync += e =>
                 {
-                    e.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended");
+                    e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.ASCII.GetBytes("extended"));
                     return CompletedTask.Instance;
                 };
 
@@ -314,7 +314,7 @@ namespace MQTTnet.Tests.Server
                 var isIntercepted = false;
                 c2.ApplicationMessageReceivedAsync += e =>
                 {
-                    isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0;
+                    isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment.ToArray()), StringComparison.Ordinal) == 0;
                     return CompletedTask.Instance;
                 };
 
@@ -425,7 +425,7 @@ namespace MQTTnet.Tests.Server
                             new MqttApplicationMessage
                             {
                                 Topic = "/test/1",
-                                Payload = Encoding.UTF8.GetBytes("true"),
+                                PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes("true")),
                                 QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
                             })
                         {
@@ -780,7 +780,7 @@ namespace MQTTnet.Tests.Server
                 var client1 = await testEnvironment.ConnectClient();
                 client1.ApplicationMessageReceivedAsync += e =>
                 {
-                    receivedBody = e.ApplicationMessage.Payload;
+                    receivedBody = e.ApplicationMessage.PayloadSegment.ToArray();
                     return CompletedTask.Instance;
                 };
 

Source/MQTTnet.Tests/Server/HotSwapCerts_Tests.cs

@@ -1,4 +1,4 @@
-#if !(NET452 || NET461)
+#if !(NET452 || NET461 || NET48)
 using System;
 using System.Collections.Concurrent;
 using System.Diagnostics;
@@ -149,18 +149,14 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        class ClientTestHarness : IDisposable
+        sealed class ClientTestHarness : IDisposable
         {
             readonly HotSwappableClientCertProvider _hotSwapClient = new HotSwappableClientCertProvider();
+
             IMqttClient _client;
 
             public string ClientId => _client.Options.ClientId;
 
-            public void ClearServerCerts()
-            {
-                _hotSwapClient.ClearServerCerts();
-            }
-
             public Task Connect()
             {
                 return Run_Client_Connection();
@@ -169,6 +165,7 @@ namespace MQTTnet.Tests.Server
             public void Dispose()
             {
                 _client.Dispose();
+                _hotSwapClient.Dispose();
             }
 
             public X509Certificate2 GetCurrentClientCert()
@@ -189,6 +186,8 @@ namespace MQTTnet.Tests.Server
 
             public void WaitForConnectOrFail(TimeSpan timeout)
             {
+                Thread.Sleep(100);
+
                 if (!_client.IsConnected)
                 {
                     _client.ReconnectAsync().Wait(timeout);
@@ -210,12 +209,12 @@ namespace MQTTnet.Tests.Server
                 Assert.IsFalse(_client.IsConnected, "Client connection success but test wanted fail");
             }
 
-            public void WaitForDisconnect(TimeSpan timeout)
+            void WaitForDisconnect(TimeSpan timeout)
             {
                 var timer = Stopwatch.StartNew();
                 while ((_client == null || _client.IsConnected) && timer.Elapsed < timeout)
                 {
-                    Thread.Sleep(5);
+                    Thread.Sleep(100);
                 }
             }
 
@@ -252,22 +251,17 @@ namespace MQTTnet.Tests.Server
                 var timer = Stopwatch.StartNew();
                 while ((_client == null || !_client.IsConnected) && timer.Elapsed < timeout)
                 {
-                    Thread.Sleep(5);
+                    Thread.Sleep(100);
                 }
             }
         }
 
-        class ServerTestHarness : IDisposable
+        sealed class ServerTestHarness : IDisposable
         {
             readonly HotSwappableServerCertProvider _hotSwapServer = new HotSwappableServerCertProvider();
 
             MqttServer _server;
 
-            public void ClearClientCerts()
-            {
-                _hotSwapServer.ClearClientCerts();
-            }
-
             public void Dispose()
             {
                 if (_server != null)
@@ -276,15 +270,12 @@ namespace MQTTnet.Tests.Server
                     _server.Dispose();
                 }
 
-                if (_hotSwapServer != null)
-                {
-                    _hotSwapServer.Dispose();
-                }
+                _hotSwapServer?.Dispose();
             }
 
-            public async Task ForceDisconnectAsync(ClientTestHarness client)
+            public Task ForceDisconnectAsync(ClientTestHarness client)
             {
-                await _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
+                return _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
             }
 
             public X509Certificate2 GetCurrentServerCert()
@@ -302,7 +293,7 @@ namespace MQTTnet.Tests.Server
                 _hotSwapServer.InstallNewClientCert(serverCert);
             }
 
-            public async Task StartServer()
+            public Task StartServer()
             {
                 var mqttFactory = new MqttFactory();
 
@@ -312,24 +303,33 @@ namespace MQTTnet.Tests.Server
                     .Build();
 
                 mqttServerOptions.TlsEndpointOptions.ClientCertificateRequired = true;
+
                 _server = mqttFactory.CreateMqttServer(mqttServerOptions);
-                await _server.StartAsync();
+                return _server.StartAsync();
             }
         }
 
-        class HotSwappableClientCertProvider : IMqttClientCertificatesProvider
+        class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable
         {
             X509Certificate2Collection _certificates;
-            ConcurrentBag<X509Certificate2> ServerCerts = new ConcurrentBag<X509Certificate2>();
+            ConcurrentBag<X509Certificate2> _serverCerts = new ConcurrentBag<X509Certificate2>();
 
             public HotSwappableClientCertProvider()
             {
                 _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
             }
-
-            public void ClearServerCerts()
+            
+            public void Dispose()
             {
-                ServerCerts = new ConcurrentBag<X509Certificate2>();
+                if (_certificates != null)
+                {
+                    foreach (var certs in _certificates)
+                    {
+#if !NET452
+                        certs.Dispose();
+#endif
+                    }
+                }
             }
 
             public X509CertificateCollection GetCertificates()
@@ -339,18 +339,17 @@ namespace MQTTnet.Tests.Server
 
             public void HotSwapCert()
             {
-                var newCert = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
-                var oldCerts = Interlocked.Exchange(ref _certificates, newCert);
+                _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
             }
 
             public void InstallNewServerCert(X509Certificate2 serverCert)
             {
-                ServerCerts.Add(serverCert);
+                _serverCerts.Add(serverCert);
             }
 
             public bool OnCertificateValidation(MqttClientCertificateValidationEventArgs certContext)
             {
-                var serverCerts = ServerCerts.ToArray();
+                var serverCerts = _serverCerts.ToArray();
 
                 var providedCert = certContext.Certificate.GetRawCertData();
                 for (int i = 0, n = serverCerts.Length; i < n; i++)
@@ -365,36 +364,18 @@ namespace MQTTnet.Tests.Server
 
                 return false;
             }
-
-            void Dispose()
-            {
-                if (_certificates != null)
-                {
-                    foreach (var certs in _certificates)
-                    {
-#if !NET452
-                        certs.Dispose();
-#endif
-                    }
-                }
-            }
         }
 
-        class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
+        sealed class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
         {
+            readonly ConcurrentBag<X509Certificate2> _clientCerts = new ConcurrentBag<X509Certificate2>();
             X509Certificate2 _certificate;
-            ConcurrentBag<X509Certificate2> ClientCerts = new ConcurrentBag<X509Certificate2>();
 
             public HotSwappableServerCertProvider()
             {
                 _certificate = CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.1");
             }
 
-            public void ClearClientCerts()
-            {
-                ClientCerts = new ConcurrentBag<X509Certificate2>();
-            }
-
             public void Dispose()
             {
 #if !NET452
@@ -418,12 +399,12 @@ namespace MQTTnet.Tests.Server
 
             public void InstallNewClientCert(X509Certificate2 certificate)
             {
-                ClientCerts.Add(certificate);
+                _clientCerts.Add(certificate);
             }
 
             public bool RemoteCertificateValidationCallback(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
             {
-                var serverCerts = ClientCerts.ToArray();
+                var serverCerts = _clientCerts.ToArray();
 
                 var providedCert = certificate.GetRawCertData();
                 for (int i = 0, n = serverCerts.Length; i < n; i++)

Source/MQTTnet.Tests/Server/Publishing_Tests.cs

@@ -8,6 +8,7 @@ using MQTTnet.Client;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Tests.Server
 {

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -283,7 +283,7 @@ namespace MQTTnet.Tests.Server
 
                 server.InterceptingPublishAsync += e =>
                 {
-                    e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty);
+                    e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty));
                     return CompletedTask.Instance;
                 };
 

Source/MQTTnet.Tests/Server/Tls_Tests.cs

@@ -101,7 +101,7 @@ namespace MQTTnet.Tests.Server
                 new MqttApplicationMessage
                 {
                     Topic = "TestTopic1",
-                    Payload = new byte[] { 1, 2, 3, 4 }
+                    PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                 });
 
             await testEnvironment.Server.InjectApplicationMessage(
@@ -109,7 +109,7 @@ namespace MQTTnet.Tests.Server
                     new MqttApplicationMessage
                     {
                         Topic = "TestTopic1",
-                        Payload = new byte[] { 1, 2, 3, 4 }
+                        PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                     }));
 
             certificateProvider.CurrentCertificate = CreateCertificate(secondOid);
@@ -137,7 +137,7 @@ namespace MQTTnet.Tests.Server
                 new MqttApplicationMessage
                 {
                     Topic = "TestTopic2",
-                    Payload = new byte[] { 1, 2, 3, 4 }
+                    PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                 });
 
             await testEnvironment.Server.InjectApplicationMessage(
@@ -145,7 +145,7 @@ namespace MQTTnet.Tests.Server
                     new MqttApplicationMessage
                     {
                         Topic = "TestTopic2",
-                        Payload = new byte[] { 1, 2, 3, 4 }
+                        PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                     }));
 
             // Ensure first client still works
@@ -153,7 +153,7 @@ namespace MQTTnet.Tests.Server
                 new MqttApplicationMessage
                 {
                     Topic = "TestTopic1",
-                    Payload = new byte[] { 1, 2, 3, 4 }
+                    PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                 });
 
             await testEnvironment.Server.InjectApplicationMessage(
@@ -161,7 +161,7 @@ namespace MQTTnet.Tests.Server
                     new MqttApplicationMessage
                     {
                         Topic = "TestTopic1",
-                        Payload = new byte[] { 1, 2, 3, 4 }
+                        PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
                     }));
 
             await Task.Delay(1000);
@@ -178,12 +178,10 @@ namespace MQTTnet.Tests.Server
             var clientOptionsBuilder = testEnvironment.Factory.CreateClientOptionsBuilder();
             clientOptionsBuilder.WithClientId(Guid.NewGuid().ToString())
                 .WithTcpServer("localhost", 8883)
-                .WithTls(
-                    tls =>
+                .WithTlsOptions(
+                    o =>
                     {
-                        tls.UseTls = true;
-                        tls.SslProtocol = SslProtocols.Tls12;
-                        tls.CertificateValidationHandler = certValidator;
+                        o.WithSslProtocols(SslProtocols.Tls12).WithCertificateValidationHandler(certValidator);
                     });
 
             var clientOptions = clientOptionsBuilder.Build();

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -25,8 +25,6 @@ namespace MQTTnet.Adapter
 
         long BytesReceived { get; }
 
-        bool IsReadingPacket { get; }
-
         Task ConnectAsync(CancellationToken cancellationToken);
 
         Task DisconnectAsync(CancellationToken cancellationToken);

Source/MQTTnet/Adapter/IMqttServerAdapter.cs

@@ -14,6 +14,7 @@ namespace MQTTnet.Adapter
         Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
 
         Task StartAsync(MqttServerOptions options, IMqttNetLogger logger);
+
         Task StopAsync();
     }
 }

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -26,9 +26,7 @@ namespace MQTTnet.Adapter
         readonly IMqttChannel _channel;
         readonly byte[] _fixedHeaderBuffer = new byte[2];
         readonly MqttNetSourceLogger _logger;
-
         readonly byte[] _singleByteBuffer = new byte[1];
-
         readonly AsyncLock _syncRoot = new AsyncLock();
 
         Statistics _statistics; // mutable struct, don't make readonly!
@@ -47,6 +45,8 @@ namespace MQTTnet.Adapter
             _logger = logger.WithSource(nameof(MqttChannelAdapter));
         }
 
+        public bool AllowPacketFragmentation { get; set; } = true;
+
         public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived);
 
         public long BytesSent => Volatile.Read(ref _statistics._bytesSent);
@@ -55,16 +55,12 @@ namespace MQTTnet.Adapter
 
         public string Endpoint => _channel.Endpoint;
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection => _channel.IsSecureConnection;
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
 
         public MqttPacketInspector PacketInspector { get; set; }
 
-        public bool AllowPacketFragmentation { get; set; } = true;
-
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
@@ -78,12 +74,30 @@ namespace MQTTnet.Adapter
                  * block forever. Even a cancellation token is not supported properly.
                  */
 
-                var connectTask = _channel.ConnectAsync(cancellationToken);
-
                 var timeout = new TaskCompletionSource<object>();
                 using (cancellationToken.Register(() => timeout.TrySetResult(null)))
                 {
+                    var connectTask = Task.Run(
+                        async () =>
+                        {
+                            try
+                            {
+                                await _channel.ConnectAsync(cancellationToken).ConfigureAwait(false);
+                            }
+                            catch
+                            {
+                                // If the timeout is already reached the exception is no longer of interest and
+                                // must be catched. Otherwise it will arrive at the TaskScheduler.UnobservedTaskException.
+                                if (!timeout.Task.IsCompleted)
+                                {
+                                    throw;
+                                }
+                            }
+                        },
+                        CancellationToken.None);
+
                     await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
+
                     if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
                     {
                         throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
@@ -204,7 +218,7 @@ namespace MQTTnet.Adapter
                     PacketInspector?.BeginSendPacket(packetBuffer);
 
                     _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
-                    
+
                     if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
                     {
                         await _channel.WriteAsync(packetBuffer.Join(), true, cancellationToken).ConfigureAwait(false);
@@ -373,53 +387,45 @@ namespace MQTTnet.Adapter
                 return ReceivedMqttPacket.Empty;
             }
 
-            IsReadingPacket = true;
-            try
+            var fixedHeader = readFixedHeaderResult.FixedHeader;
+            if (fixedHeader.RemainingLength == 0)
             {
-                var fixedHeader = readFixedHeaderResult.FixedHeader;
-                if (fixedHeader.RemainingLength == 0)
+                return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
+            }
+
+            var bodyLength = fixedHeader.RemainingLength;
+            var body = new byte[bodyLength];
+
+            var bodyOffset = 0;
+            var chunkSize = Math.Min(ReadBufferSize, bodyLength);
+
+            do
+            {
+                var bytesLeft = body.Length - bodyOffset;
+                if (chunkSize > bytesLeft)
                 {
-                    return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
+                    chunkSize = bytesLeft;
                 }
 
-                var bodyLength = fixedHeader.RemainingLength;
-                var body = new byte[bodyLength];
+                var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
 
-                var bodyOffset = 0;
-                var chunkSize = Math.Min(ReadBufferSize, bodyLength);
-
-                do
+                if (cancellationToken.IsCancellationRequested)
                 {
-                    var bytesLeft = body.Length - bodyOffset;
-                    if (chunkSize > bytesLeft)
-                    {
-                        chunkSize = bytesLeft;
-                    }
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
+                if (readBytes == 0)
+                {
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+                bodyOffset += readBytes;
+            } while (bodyOffset < bodyLength);
 
-                    if (readBytes == 0)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+            PacketInspector?.FillReceiveBuffer(body);
 
-                    bodyOffset += readBytes;
-                } while (bodyOffset < bodyLength);
-
-                PacketInspector?.FillReceiveBuffer(body);
-
-                var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
-                return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
-            }
-            finally
-            {
-                IsReadingPacket = false;
-            }
+            var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
+            return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
         }
 
         static bool WrapAndThrowException(Exception exception)

Source/MQTTnet/Client/MqttClient.cs

@@ -238,6 +238,11 @@ namespace MQTTnet.Client
 
         public async Task PingAsync(CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+            
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
             if (cancellationToken.CanBeCanceled)
             {
                 await Request<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false);
@@ -304,6 +309,7 @@ namespace MQTTnet.Client
                 AuthenticationMethod = Options.AuthenticationMethod,
                 AuthenticationData = data.AuthenticationData,
                 ReasonString = data.ReasonString,
+                ReasonCode = data.ReasonCode,
                 UserProperties = data.UserProperties
             };
 
@@ -441,7 +447,7 @@ namespace MQTTnet.Client
 
                 if (receivedPacket is MqttConnAckPacket connAckPacket)
                 {
-                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
+                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
                 }
                 else if (receivedPacket is MqttAuthPacket)
                 {
@@ -513,7 +519,7 @@ namespace MQTTnet.Client
             using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
             {
                 _logger.Verbose("Trying to connect with server '{0}'", Options.ChannelOptions);
-                await _adapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
+                await channelAdapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
                 _logger.Verbose("Connection with server established");
 
                 _publishPacketReceiverQueue?.Dispose();
@@ -703,12 +709,15 @@ namespace MQTTnet.Client
                         await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
                     }
                 }
+                catch (ObjectDisposedException)
+                {
+                }
                 catch (OperationCanceledException)
                 {
                 }
                 catch (Exception exception)
                 {
-                    _logger.Error(exception, "Error while handling application message.");
+                    _logger.Error(exception, "Error while handling application message");
                 }
             }
         }
@@ -782,7 +791,7 @@ namespace MQTTnet.Client
         {
             try
             {
-                _logger.Verbose("Start receiving packets.");
+                _logger.Verbose("Start receiving packets");
 
                 while (!cancellationToken.IsCancellationRequested)
                 {
@@ -820,20 +829,21 @@ namespace MQTTnet.Client
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Communication error while receiving packets.");
+                    _logger.Warning(exception, "Communication error while receiving packets");
                 }
                 else
                 {
-                    _logger.Error(exception, "Error while receiving packets.");
+                    _logger.Error(exception, "Error while receiving packets");
                 }
 
-                _packetDispatcher.FailAll(exception);
+                // The packet dispatcher is set to null when the client is being disposed so it may already being gone!
+                _packetDispatcher?.FailAll(exception);
 
                 await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
             }
             finally
             {
-                _logger.Verbose("Stopped receiving packets.");
+                _logger.Verbose("Stopped receiving packets");
             }
         }
 
@@ -855,7 +865,7 @@ namespace MQTTnet.Client
                 }
                 catch (Exception exception)
                 {
-                    _logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
+                    _logger.Warning(exception, "Error when sending {0} request packet", requestPacket.GetType().Name);
                     packetAwaitable.Fail(exception);
                 }
 
@@ -867,7 +877,7 @@ namespace MQTTnet.Client
                 {
                     if (exception is MqttCommunicationTimedOutException)
                     {
-                        _logger.Warning("Timeout while waiting for response packet ({0}).", typeof(TResponsePacket).Name);
+                        _logger.Warning("Timeout while waiting for {0} response packet", typeof(TResponsePacket).Name);
                     }
 
                     throw;
@@ -933,7 +943,7 @@ namespace MQTTnet.Client
                 }
                 catch (Exception exception)
                 {
-                    _logger.Warning(exception, "Error while initiating disconnect.");
+                    _logger.Warning(exception, "Error while initiating disconnect");
                 }
             }
         }
@@ -968,7 +978,7 @@ namespace MQTTnet.Client
                 }
                 else if (packet is MqttPingReqPacket)
                 {
-                    throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a Client to the Server only.");
+                    throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a client to the server only.");
                 }
                 else
                 {
@@ -990,14 +1000,15 @@ namespace MQTTnet.Client
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Communication error while receiving packets.");
+                    _logger.Warning(exception, "Communication error while receiving packets");
                 }
                 else
                 {
-                    _logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name}).");
+                    _logger.Error(exception, "Error while processing received {0} packet", packet.GetType().Name);
                 }
 
-                _packetDispatcher.FailAll(exception);
+                // The packet dispatcher may already be gone due to disconnect etc!
+                _packetDispatcher?.FailAll(exception);
 
                 await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
             }
@@ -1007,7 +1018,7 @@ namespace MQTTnet.Client
         {
             try
             {
-                _logger.Verbose("Start sending keep alive packets.");
+                _logger.Verbose("Start sending keep alive packets");
 
                 var keepAlivePeriod = Options.KeepAlivePeriod;
 
@@ -1045,18 +1056,18 @@ namespace MQTTnet.Client
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Communication error while sending/receiving keep alive packets.");
+                    _logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
                 }
                 else
                 {
-                    _logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
+                    _logger.Error(exception, "Error exception while sending/receiving keep alive packets");
                 }
 
                 await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
             }
             finally
             {
-                _logger.Verbose("Stopped sending keep alive packets.");
+                _logger.Verbose("Stopped sending keep alive packets");
             }
         }
     }

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -2,6 +2,8 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+#pragma warning disable CS0612 // Type or member is obsolete
+
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -32,7 +34,12 @@ namespace MQTTnet.Client
                 throw new InvalidOperationException("A channel must be set.");
             }
 
-            var tlsOptions = _tlsOptions;
+            // The user can specify the TCP options with already configured TLS options
+            // or start with TLS settings not knowing which transport will be used (depending
+            // on the order of called methods from the builder).
+            // The builder prefers the explicitly set TLS options!
+            var tlsOptions = _tlsOptions ?? _tcpOptions?.TlsOptions;
+
             if (_tlsParameters != null)
             {
                 if (_tlsParameters?.UseTls == true)

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -50,8 +50,13 @@ namespace MQTTnet.Client
 
 #if NET48 || NETCOREAPP3_1_OR_GREATER
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
+
 #else
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | (SslProtocols)0x00003000 /*Tls13*/;
 #endif
+
+#if NET7_0_OR_GREATER
+        public X509Certificate2Collection TrustChain { get; set; }
+#endif
     }
 }

Source/MQTTnet/Client/Options/MqttClientTlsOptionsBuilder.cs

@@ -131,6 +131,15 @@ namespace MQTTnet.Client
             _tlsOptions.EncryptionPolicy = encryptionPolicy;
             return this;
         }
+#endif
+
+#if NET7_0_OR_GREATER
+        public MqttClientTlsOptionsBuilder WithTrustChain(X509Certificate2Collection chain)
+        {
+            _tlsOptions.TrustChain = chain;
+            return this;
+        }
+
 #endif
     }
 }

Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs

@@ -5,45 +5,107 @@
 using MQTTnet.Client;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Formatter
 {
     public sealed class MqttDisconnectPacketFactory
     {
-        readonly MqttDisconnectPacket _normalDisconnection = new MqttDisconnectPacket
+        static readonly MqttDisconnectPacket DefaultNormalDisconnection = new MqttDisconnectPacket
         {
-            ReasonCode = MqttDisconnectReasonCode.NormalDisconnection
+            ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
+            UserProperties = null,
+            ReasonString = null,
+            ServerReference = null,
+            SessionExpiryInterval = 0
+        };
+
+        static readonly MqttDisconnectPacket DefaultServerShuttingDown = new MqttDisconnectPacket
+        {
+            ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
+            UserProperties = null,
+            ReasonString = null,
+            ServerReference = null,
+            SessionExpiryInterval = 0
+        };
+
+        static readonly MqttDisconnectPacket DefaultUnspecifiedError = new MqttDisconnectPacket
+        {
+            ReasonCode = MqttDisconnectReasonCode.UnspecifiedError,
+            UserProperties = null,
+            ReasonString = null,
+            ServerReference = null,
+            SessionExpiryInterval = 0
         };
 
         public MqttDisconnectPacket Create(MqttDisconnectReasonCode reasonCode)
         {
             if (reasonCode == MqttDisconnectReasonCode.NormalDisconnection)
             {
-                return _normalDisconnection;
+                return DefaultNormalDisconnection;
+            }
+
+            if (reasonCode == MqttDisconnectReasonCode.ServerShuttingDown)
+            {
+                return DefaultServerShuttingDown;
+            }
+
+            if (reasonCode == MqttDisconnectReasonCode.UnspecifiedError)
+            {
+                return DefaultUnspecifiedError;
             }
 
             return new MqttDisconnectPacket
             {
-                ReasonCode = reasonCode
+                ReasonCode = reasonCode,
+                UserProperties = null,
+                ReasonString = null,
+                ServerReference = null,
+                SessionExpiryInterval = 0
+            };
+        }
+
+        public MqttDisconnectPacket Create(MqttServerStopOptions serverStopOptions)
+        {
+            if (serverStopOptions == null)
+            {
+                return DefaultServerShuttingDown;
+            }
+
+            return Create(serverStopOptions.DefaultClientDisconnectOptions);
+        }
+
+        public MqttDisconnectPacket Create(MqttServerClientDisconnectOptions clientDisconnectOptions)
+        {
+            if (clientDisconnectOptions == null)
+            {
+                return DefaultNormalDisconnection;
+            }
+
+            return new MqttDisconnectPacket
+            {
+                ReasonCode = clientDisconnectOptions.ReasonCode,
+                UserProperties = clientDisconnectOptions.UserProperties,
+                ReasonString = clientDisconnectOptions.ReasonString,
+                ServerReference = clientDisconnectOptions.ServerReference,
+                SessionExpiryInterval = 0 // TODO: Not yet supported!
             };
         }
 
         public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions)
         {
-            var packet = new MqttDisconnectPacket();
-
             if (clientDisconnectOptions == null)
             {
-                packet.ReasonCode = MqttDisconnectReasonCode.NormalDisconnection;
-            }
-            else
-            {
-                packet.ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason;
-                packet.UserProperties = clientDisconnectOptions.UserProperties;
-                packet.SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval;
+                return DefaultNormalDisconnection;
             }
 
-            return packet;
+            return new MqttDisconnectPacket
+            {
+                ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason,
+                UserProperties = clientDisconnectOptions.UserProperties,
+                SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval
+            };
         }
     }
 }

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -115,14 +115,31 @@ namespace MQTTnet.Implementations
                             ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
                             ClientCertificates = LoadCertificates(),
                             EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
-                            CertificateRevocationCheckMode =
- _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
+                            CertificateRevocationCheckMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
+                                ? X509RevocationMode.NoCheck
+                                : _tcpOptions.TlsOptions.RevocationMode,
                             TargetHost = targetHost,
                             CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
                             EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
                             AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
                         };
 
+#if NET7_0_OR_GREATER
+                        if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
+                        {
+                            sslOptions.CertificateChainPolicy = new X509ChainPolicy
+                            {
+                                TrustMode = X509ChainTrustMode.CustomRootTrust,
+                                VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
+                                RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
+                                    ? X509RevocationMode.NoCheck
+                                    : _tcpOptions.TlsOptions.RevocationMode
+                            };
+
+                            sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
+                        }
+#endif
+
                         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
 #else
                         await sslStream.AuthenticateAsClientAsync(

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -46,13 +46,13 @@ namespace MQTTnet.Implementations
             var uri = _options.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
             {
-                if (_options.TlsOptions?.UseTls == false)
+                if (_options.TlsOptions?.UseTls == true)
                 {
-                    uri = "ws://" + uri;
+                    uri = "wss://" + uri;
                 }
                 else
                 {
-                    uri = "wss://" + uri;
+                    uri = "ws://" + uri;
                 }
             }
 

Source/MQTTnet/Internal/MqttPacketBus.cs

@@ -104,17 +104,22 @@ namespace MQTTnet.Internal
             _signal.Dispose();
         }
 
-        public void DropFirstItem(MqttPacketBusPartition partition)
+        public MqttPacketBusItem DropFirstItem(MqttPacketBusPartition partition)
         {
             lock (_syncRoot)
             {
                 var partitionInstance = _partitions[(int)partition];
 
-                if (partitionInstance.Any())
+                if (partitionInstance.Count > 0)
                 {
+                    var firstItem = partitionInstance.First.Value;
                     partitionInstance.RemoveFirst();
+
+                    return firstItem;
                 }
             }
+
+            return null;
         }
 
         public void EnqueueItem(MqttPacketBusItem item, MqttPacketBusPartition partition)

Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs

@@ -19,7 +19,6 @@ namespace MQTTnet.LowLevelClient
         readonly IMqttClientAdapterFactory _clientAdapterFactory;
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
         readonly MqttNetSourceLogger _logger;
-
         readonly IMqttNetLogger _rootLogger;
 
         IMqttChannelAdapter _adapter;

Source/MQTTnet/MQTTnet.csproj.DotSettings

@@ -17,4 +17,5 @@
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cmanagement/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstopping/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

Source/MQTTnet/MqttFactory.cs

@@ -11,6 +11,7 @@ using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
 using MQTTnet.LowLevelClient;
 using MQTTnet.Server;
+using MQTTnet.Server.Disconnecting;
 using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet
@@ -175,6 +176,16 @@ namespace MQTTnet
             return new MqttServer(options, serverAdapters, DefaultLogger);
         }
 
+        public MqttServerClientDisconnectOptionsBuilder CreateMqttServerClientDisconnectOptionsBuilder()
+        {
+            return new MqttServerClientDisconnectOptionsBuilder();
+        }
+
+        public MqttServerStopOptionsBuilder CreateMqttServerStopOptionsBuilder()
+        {
+            return new MqttServerStopOptionsBuilder();
+        }
+
         public MqttServerOptionsBuilder CreateServerOptionsBuilder()
         {
             return new MqttServerOptionsBuilder();

Source/MQTTnet/Server/Disconnecting/MqttServerClientDisconnectOptions.cs

@@ -0,0 +1,33 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server.Disconnecting
+{
+    public sealed class MqttServerClientDisconnectOptions
+    {
+        public MqttDisconnectReasonCode ReasonCode { get; set; } = MqttDisconnectReasonCode.NormalDisconnection;
+
+        /// <summary>
+        ///     The reason string is sent to every client via a DISCONNECT packet.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public string ReasonString { get; set; }
+
+        /// <summary>
+        ///     The server reference is sent to every client via a DISCONNECT packet.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public string ServerReference { get; set; }
+
+        /// <summary>
+        ///     These user properties are sent to every client via a DISCONNECT packet.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public List<MqttUserProperty> UserProperties { get; set; }
+    }
+}

Source/MQTTnet/Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs

@@ -0,0 +1,55 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server.Disconnecting
+{
+    public sealed class MqttServerClientDisconnectOptionsBuilder
+    {
+        readonly MqttServerClientDisconnectOptions _options = new MqttServerClientDisconnectOptions();
+
+        public MqttServerClientDisconnectOptions Build()
+        {
+            return _options;
+        }
+
+        public MqttServerClientDisconnectOptionsBuilder WithReasonCode(MqttDisconnectReasonCode value)
+        {
+            _options.ReasonCode = value;
+            return this;
+        }
+
+        public MqttServerClientDisconnectOptionsBuilder WithReasonString(string value)
+        {
+            _options.ReasonString = value;
+            return this;
+        }
+
+        public MqttServerClientDisconnectOptionsBuilder WithServerReference(string value)
+        {
+            _options.ServerReference = value;
+            return this;
+        }
+
+        public MqttServerClientDisconnectOptionsBuilder WithUserProperties(List<MqttUserProperty> value)
+        {
+            _options.UserProperties = value;
+            return this;
+        }
+
+        public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, string value)
+        {
+            if (_options.UserProperties == null)
+            {
+                _options.UserProperties = new List<MqttUserProperty>();
+            }
+
+            _options.UserProperties.Add(new MqttUserProperty(name, value));
+            return this;
+        }
+    }
+}

Source/MQTTnet/Server/Events/ApplicationMessageEnqueuedEventArgs.cs

@@ -0,0 +1,27 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Server
+{
+    public sealed class ApplicationMessageEnqueuedEventArgs : EventArgs
+    {
+        public ApplicationMessageEnqueuedEventArgs(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, bool isDropped)
+        {
+            SenderClientId = senderClientId ?? throw new ArgumentNullException( nameof(senderClientId));
+            ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            IsDropped = isDropped;
+        }
+
+        public string SenderClientId { get; }
+
+        public string ReceiverClientId { get; }
+
+        public bool IsDropped { get; }
+
+        public MqttApplicationMessage ApplicationMessage { get; }
+    }
+}

Source/MQTTnet/Server/Events/QueueMessageOverwrittenEventArgs.cs

@@ -0,0 +1,22 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class QueueMessageOverwrittenEventArgs : EventArgs
+    {
+        public QueueMessageOverwrittenEventArgs(string receiverClientId, MqttPacket packet)
+        {
+            ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
+            Packet = packet ?? throw new ArgumentNullException(nameof(packet));
+        }
+
+        public MqttPacket Packet { get; }
+
+        public string ReceiverClientId { get; }
+    }
+}

Source/MQTTnet/Server/Internal/EnqueueDataPacketResult.cs

@@ -0,0 +1,12 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Server
+{
+    public enum EnqueueDataPacketResult
+    {
+        Enqueued,
+        Dropped
+    }
+}

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
@@ -14,6 +15,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -95,15 +97,16 @@ namespace MQTTnet.Server
             {
                 var cancellationToken = _cancellationToken.Token;
                 IsRunning = true;
-                
-                _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
+
+                _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default)
+                    .ConfigureAwait(false);
 
                 await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
             }
             finally
             {
                 IsRunning = false;
-                
+
                 Session.DisconnectedTimestamp = DateTime.UtcNow;
 
                 _cancellationToken?.TryCancel();
@@ -141,7 +144,7 @@ namespace MQTTnet.Server
             Statistics.HandleSentPacket(packet);
         }
 
-        public async Task StopAsync(MqttDisconnectReasonCode reason)
+        public async Task StopAsync(MqttServerClientDisconnectOptions disconnectOptions)
         {
             IsRunning = false;
 
@@ -150,16 +153,20 @@ namespace MQTTnet.Server
                 // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
                 if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
                 {
-                    // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
-                    // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
-                    // a configuration option is requested in the future.
-                    if (reason != MqttDisconnectReasonCode.NormalDisconnection)
+                    // From RFC: The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
+                    // This library does not sent a DISCONNECT packet for a normal disconnection.
+                    // TODO: Maybe adding a configuration option is requested in the future.
+                    if (disconnectOptions != null)
                     {
-                        // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
-                        // token because the entire connection is closed (disposed) as soon as the cancellation
-                        // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
-                        // at the client!
-                        await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+                        if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
+                            !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
+                        {
+                            // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
+                            // token because the entire connection is closed (disposed) as soon as the cancellation
+                            // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+                            // at the client!
+                            await TrySendDisconnectPacket(disconnectOptions).ConfigureAwait(false);
+                        }
                     }
                 }
             }
@@ -219,7 +226,7 @@ namespace MQTTnet.Server
 
             if (dispatchApplicationMessageResult.CloseConnection)
             {
-                await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
+                await StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.UnspecifiedError });
                 return;
             }
 
@@ -550,14 +557,14 @@ namespace MQTTnet.Server
             _cancellationToken?.TryCancel();
         }
 
-        async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode)
+        async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options)
         {
             try
             {
                 // This also indicates that it was tried at least!
                 _disconnectPacketSent = true;
 
-                var disconnectPacket = MqttPacketFactories.Disconnect.Create(reasonCode);
+                var disconnectPacket = MqttPacketFactories.Disconnect.Create(options);
 
                 using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
                 {
@@ -566,7 +573,7 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, reasonCode);
+                _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
             }
         }
     }

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -16,6 +16,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -32,11 +33,11 @@ namespace MQTTnet.Server
         readonly MqttRetainedMessagesManager _retainedMessagesManager;
         readonly IMqttNetLogger _rootLogger;
 
+        readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
+
         // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
         // See the MqttSubscription object for a detailed explanation.
         readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
-
-        readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
         readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
 
         public MqttClientSessionsManager(
@@ -58,8 +59,13 @@ namespace MQTTnet.Server
             _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
         }
 
-        public async Task CloseAllConnections()
+        public async Task CloseAllConnections(MqttServerClientDisconnectOptions options)
         {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
             List<MqttClient> connections;
             lock (_clients)
             {
@@ -69,7 +75,7 @@ namespace MQTTnet.Server
 
             foreach (var connection in connections)
             {
-                await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                await connection.StopAsync(options).ConfigureAwait(false);
             }
         }
 
@@ -101,7 +107,7 @@ namespace MQTTnet.Server
             {
                 if (connection != null)
                 {
-                    await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                    await connection.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.NormalDisconnection }).ConfigureAwait(false);
                 }
             }
             catch (Exception exception)
@@ -204,7 +210,7 @@ namespace MQTTnet.Server
                         {
                             continue;
                         }
-                        
+
                         if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
                         {
                             var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
@@ -238,7 +244,14 @@ namespace MQTTnet.Server
 
                         matchingSubscribersCount++;
 
-                        session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
+                        var result = session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
+
+                        if (_eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.HasHandlers)
+                        {
+                            var eventArgs = new ApplicationMessageEnqueuedEventArgs(senderId, session.Id, applicationMessage, result == EnqueueDataPacketResult.Dropped);
+                            await _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                        }
+
                         _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'", session.Id, applicationMessage.Topic);
                     }
 
@@ -366,7 +379,8 @@ namespace MQTTnet.Server
 
                 if (_eventContainer.ClientConnectedEvent.HasHandlers)
                 {
-                    var eventArgs = new ClientConnectedEventArgs(connectPacket,
+                    var eventArgs = new ClientConnectedEventArgs(
+                        connectPacket,
                         channelAdapter.PacketFormatterAdapter.ProtocolVersion,
                         channelAdapter.Endpoint,
                         client.Session.Items);
@@ -595,13 +609,13 @@ namespace MQTTnet.Server
                             session.IsPersistent = sessionShouldPersist;
                             session.DisconnectedTimestamp = null;
                             session.Recover();
-                            
+
                             connAckPacket.IsSessionPresent = true;
                         }
                     }
 
                     _sessionsStorage.UpdateSession(connectPacket.ClientId, session);
-                    
+
                     // Create a new client (always required).
                     lock (_clients)
                     {
@@ -631,7 +645,8 @@ namespace MQTTnet.Server
 
                 if (oldClient != null)
                 {
-                    await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
+                    // TODO: Consider event here for session takeover to allow manipulation of user properties etc.
+                    await oldClient.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.SessionTakenOver }).ConfigureAwait(false);
 
                     if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
                     {
@@ -734,4 +749,4 @@ namespace MQTTnet.Server
             return eventArgs;
         }
     }
-}
+}

Source/MQTTnet/Server/Internal/MqttClientStatistics.cs

@@ -92,7 +92,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private struct Statistics
+        struct Statistics
         {
             public long _receivedPacketsCount;
             public long _sentPacketsCount;

Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs

@@ -22,6 +22,10 @@ namespace MQTTnet.Server
         public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
 
         public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs>();
+        
+        public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new AsyncEvent<ApplicationMessageEnqueuedEventArgs>();
+
+        public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new AsyncEvent<QueueMessageOverwrittenEventArgs>();
 
         public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
 

Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -8,11 +8,20 @@ using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
     public sealed class MqttServerKeepAliveMonitor
     {
+        static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
+        {
+            ReasonCode = MqttDisconnectReasonCode.KeepAliveTimeout,
+            UserProperties = null,
+            ReasonString = null,
+            ServerReference = null
+        };
+
         readonly MqttNetSourceLogger _logger;
         readonly MqttServerOptions _options;
         readonly MqttClientSessionsManager _sessionsManager;
@@ -43,7 +52,7 @@ namespace MQTTnet.Server
         {
             try
             {
-                _logger.Info("Starting keep alive monitor.");
+                _logger.Info("Starting keep alive monitor");
 
                 while (!cancellationToken.IsCancellationRequested)
                 {
@@ -56,11 +65,11 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Unhandled exception while checking keep alive timeouts.");
+                _logger.Error(exception, "Unhandled exception while checking keep alive timeouts");
             }
             finally
             {
-                _logger.Verbose("Stopped checking keep alive timeout.");
+                _logger.Verbose("Stopped checking keep alive timeout");
             }
         }
 
@@ -98,13 +107,6 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                if (connection.ChannelAdapter.IsReadingPacket)
-                {
-                    // The connection is currently reading a (large) packet. So it is obviously 
-                    // doing something and thus "connected".
-                    return;
-                }
-
                 // Values described here: [MQTT-3.1.2-24].
                 // If the client sends 5 sec. the server will allow up to 7.5 seconds.
                 // If the client sends 1 sec. the server will allow up to 1.5 seconds.
@@ -117,18 +119,18 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.Id);
+                _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal", connection.Id);
 
                 // Execute the disconnection in background so that the keep alive monitor can continue
                 // with checking other connections.
                 // We do not need to wait for the task so no await is needed.
                 // Also the internal state of the connection must be swapped to "Finalizing" because the
                 // next iteration of the keep alive timer happens.
-                var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout);
+                _ = connection.StopAsync(DefaultDisconnectOptions);
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id);
+                _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts", connection.Id);
             }
         }
 

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -18,6 +18,7 @@ namespace MQTTnet.Server
     public sealed class MqttSession : IDisposable
     {
         readonly MqttClientSessionsManager _clientSessionsManager;
+        readonly MqttServerEventContainer _eventContainer;
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
@@ -44,6 +45,7 @@ namespace MQTTnet.Server
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
+            _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
 
             _subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
         }
@@ -117,20 +119,25 @@ namespace MQTTnet.Server
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
         }
 
-        public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
+        public EnqueueDataPacketResult EnqueueDataPacket(MqttPacketBusItem packetBusItem)
         {
             if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
             {
                 if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
                 {
-                    return;
+                    return EnqueueDataPacketResult.Dropped;
                 }
 
                 if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
                 {
                     // Only drop from the data partition. Dropping from control partition might break the connection
                     // because the client does not receive PINGREQ packets etc. any longer.
-                    _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
+                    var firstItem = _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
+                    if (firstItem != null && _eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)
+                    {
+                        var eventArgs = new QueueMessageOverwrittenEventArgs(Id, firstItem.Packet);
+                        _eventContainer.QueuedApplicationMessageOverwrittenEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                    }
                 }
             }
 
@@ -147,6 +154,7 @@ namespace MQTTnet.Server
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
+            return EnqueueDataPacketResult.Enqueued;
         }
 
         public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)

Source/MQTTnet/Server/MqttServer.cs

@@ -14,6 +14,7 @@ using MQTTnet.Diagnostics;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -29,6 +30,7 @@ namespace MQTTnet.Server
         readonly IMqttNetLogger _rootLogger;
 
         CancellationTokenSource _cancellationTokenSource;
+        bool _isStopping;
 
         public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
@@ -49,6 +51,12 @@ namespace MQTTnet.Server
             _keepAliveMonitor = new MqttServerKeepAliveMonitor(options, _clientSessionsManager, _rootLogger);
         }
 
+        public event Func<ApplicationMessageEnqueuedEventArgs, Task> ApplicationMessageEnqueuedOrDroppedAsync
+        {
+            add => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.AddHandler(value);
+            remove => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.RemoveHandler(value);
+        }
+
         public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
         {
             add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value);
@@ -133,6 +141,12 @@ namespace MQTTnet.Server
             remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
         }
 
+        public event Func<QueueMessageOverwrittenEventArgs, Task> QueuedApplicationMessageOverwrittenAsync
+        {
+            add => _eventContainer.QueuedApplicationMessageOverwrittenEvent.AddHandler(value);
+            remove => _eventContainer.QueuedApplicationMessageOverwrittenEvent.RemoveHandler(value);
+        }
+
         public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
         {
             add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);
@@ -169,6 +183,13 @@ namespace MQTTnet.Server
             remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
         }
 
+        /// <summary>
+        ///     Gets or sets whether the server will accept new connections.
+        ///     If not, the server will close the connection without any notification (DISCONNECT packet).
+        ///     This feature can be used when the server is shutting down.
+        /// </summary>
+        public bool AcceptNewConnections { get; set; } = true;
+
         public bool IsStarted => _cancellationTokenSource != null;
 
         /// <summary>
@@ -184,16 +205,21 @@ namespace MQTTnet.Server
             return _retainedMessagesManager?.ClearMessages() ?? CompletedTask.Instance;
         }
 
-        public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode)
+        public Task DisconnectClientAsync(string id, MqttServerClientDisconnectOptions options)
         {
             if (id == null)
             {
                 throw new ArgumentNullException(nameof(id));
             }
 
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
             ThrowIfNotStarted();
 
-            return _clientSessionsManager.GetClient(id).StopAsync(reasonCode);
+            return _clientSessionsManager.GetClient(id).StopAsync(options);
         }
 
         public Task<IList<MqttClientStatus>> GetClientsAsync()
@@ -203,13 +229,6 @@ namespace MQTTnet.Server
             return _clientSessionsManager.GetClientsStatus();
         }
 
-        public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
-        {
-            ThrowIfNotStarted();
-
-            return _retainedMessagesManager.GetMessages();
-        }
-
         public Task<MqttApplicationMessage> GetRetainedMessageAsync(string topic)
         {
             if (topic == null)
@@ -222,6 +241,13 @@ namespace MQTTnet.Server
             return _retainedMessagesManager.GetMessage(topic);
         }
 
+        public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
+        {
+            ThrowIfNotStarted();
+
+            return _retainedMessagesManager.GetMessages();
+        }
+
         public Task<IList<MqttSessionStatus>> GetSessionsAsync()
         {
             ThrowIfNotStarted();
@@ -247,7 +273,7 @@ namespace MQTTnet.Server
 
             if (string.IsNullOrEmpty(injectedApplicationMessage.ApplicationMessage.Topic))
             {
-                throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported.");
+                throw new NotSupportedException("Injected application messages must contain a topic (topic alias is not supported)");
             }
 
             var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems;
@@ -263,6 +289,8 @@ namespace MQTTnet.Server
         {
             ThrowIfStarted();
 
+            _isStopping = false;
+            
             _cancellationTokenSource = new CancellationTokenSource();
             var cancellationToken = _cancellationTokenSource.Token;
 
@@ -278,11 +306,16 @@ namespace MQTTnet.Server
 
             await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
 
-            _logger.Info("Started.");
+            _logger.Info("Started");
         }
 
-        public async Task StopAsync()
+        public async Task StopAsync(MqttServerStopOptions options)
         {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
             try
             {
                 if (_cancellationTokenSource == null)
@@ -290,9 +323,11 @@ namespace MQTTnet.Server
                     return;
                 }
 
+                _isStopping = true;
+
                 _cancellationTokenSource.Cancel(false);
 
-                await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false);
+                await _clientSessionsManager.CloseAllConnections(options.DefaultClientDisconnectOptions).ConfigureAwait(false);
 
                 foreach (var adapter in _adapters)
                 {
@@ -308,7 +343,7 @@ namespace MQTTnet.Server
 
             await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
 
-            _logger.Info("Stopped.");
+            _logger.Info("Stopped");
         }
 
         public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -362,14 +397,14 @@ namespace MQTTnet.Server
             ThrowIfDisposed();
             ThrowIfNotStarted();
 
-            return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
+            return _retainedMessagesManager?.UpdateMessage(string.Empty, retainedMessage);
         }
 
         protected override void Dispose(bool disposing)
         {
             if (disposing)
             {
-                StopAsync().GetAwaiter().GetResult();
+                StopAsync(new MqttServerStopOptions()).GetAwaiter().GetResult();
 
                 foreach (var adapter in _adapters)
                 {
@@ -382,6 +417,11 @@ namespace MQTTnet.Server
 
         Task OnHandleClient(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
+            if (_isStopping || !AcceptNewConnections)
+            {
+                return CompletedTask.Instance;
+            }
+
             return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken);
         }
 

Source/MQTTnet/Server/MqttServerExtensions.cs

@@ -8,11 +8,22 @@ using System.Threading.Tasks;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
     public static class MqttServerExtensions
     {
+        public static Task DisconnectClientAsync(this MqttServer server, string id, MqttDisconnectReasonCode reasonCode = MqttDisconnectReasonCode.NormalDisconnection)
+        {
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
+
+            return server.DisconnectClientAsync(id, new MqttServerClientDisconnectOptions { ReasonCode = reasonCode });
+        }
+
         public static Task InjectApplicationMessage(
             this MqttServer server,
             string topic,
@@ -47,6 +58,16 @@ namespace MQTTnet.Server
                     }));
         }
 
+        public static Task StopAsync(this MqttServer server)
+        {
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
+
+            return server.StopAsync(new MqttServerStopOptions());
+        }
+
         public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
         {
             if (server == null)

Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs

@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttServerKeepAliveOptions
+    {
+        /// <summary>
+        ///     When this mode is enabled the MQTT server will not close a connection when the
+        ///     client is currently sending a (large) payload. This may lead to "dead" connections
+        ///     When this mode is disabled the MQTT server will disconnect a client when the keep
+        ///     alive timeout is reached even if the client is currently sending a (large) payload.
+        /// </summary>
+        public bool DisconnectClientWhenReadingPayload { get; set; }
+
+        public TimeSpan MonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+    }
+}

Source/MQTTnet/Server/Options/MqttServerOptions.cs

@@ -9,12 +9,19 @@ namespace MQTTnet.Server
     public sealed class MqttServerOptions
     {
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(100);
-        
+
         public MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } = new MqttServerTcpEndpointOptions();
 
         public bool EnablePersistentSessions { get; set; }
 
-        public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+        [Obsolete("Use KeepAliveOptions instead.")]
+        public TimeSpan KeepAliveMonitorInterval
+        {
+            get => KeepAliveOptions.MonitorInterval;
+            set => KeepAliveOptions.MonitorInterval = value;
+        }
+
+        public MqttServerKeepAliveOptions KeepAliveOptions { get; } = new MqttServerKeepAliveOptions();
 
         public int MaxPendingMessagesPerClient { get; set; } = 250;
 

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -5,7 +5,7 @@
 using System;
 using System.Threading.Tasks;
 using MQTTnet.Formatter;
-using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -18,41 +18,46 @@ namespace MQTTnet.Server
             _client = client ?? throw new ArgumentNullException(nameof(client));
         }
 
-        /// <summary>
-        /// Gets or sets the client identifier.
-        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
-        /// </summary>
-        public string Id => _client.Id;
+        public long BytesReceived => _client.ChannelAdapter.BytesReceived;
+
+        public long BytesSent => _client.ChannelAdapter.BytesSent;
+
+        public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp;
 
         public string Endpoint => _client.Endpoint;
 
-        public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
+        /// <summary>
+        ///     Gets or sets the client identifier.
+        ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
+        /// </summary>
+        public string Id => _client.Id;
 
-        public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp;
+        public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
 
         public DateTime LastPacketReceivedTimestamp => _client.Statistics.LastPacketReceivedTimestamp;
 
         public DateTime LastPacketSentTimestamp => _client.Statistics.LastPacketSentTimestamp;
 
-        public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
+        public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
 
         public long ReceivedApplicationMessagesCount => _client.Statistics.ReceivedApplicationMessagesCount;
 
-        public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
-
         public long ReceivedPacketsCount => _client.Statistics.ReceivedPacketsCount;
 
+        public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
+
         public long SentPacketsCount => _client.Statistics.SentPacketsCount;
 
         public MqttSessionStatus Session { get; set; }
 
-        public long BytesSent => _client.ChannelAdapter.BytesSent;
-
-        public long BytesReceived => _client.ChannelAdapter.BytesReceived;
-        
-        public Task DisconnectAsync()
+        public Task DisconnectAsync(MqttServerClientDisconnectOptions options)
         {
-            return _client.StopAsync(MqttDisconnectReasonCode.NormalDisconnection);
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            return _client.StopAsync(options);
         }
 
         public void ResetStatistics()
@@ -60,4 +65,4 @@ namespace MQTTnet.Server
             _client.ResetStatistics();
         }
     }
-}
+}

Source/MQTTnet/Server/Status/MqttClientStatusExtensions.cs

@@ -0,0 +1,32 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading.Tasks;
+using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
+
+namespace MQTTnet.Server
+{
+    public static class MqttClientStatusExtensions
+    {
+        static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
+        {
+            ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
+            ReasonString = null,
+            UserProperties = null,
+            ServerReference = null
+        };
+
+        public static Task DisconnectAsync(this MqttClientStatus clientStatus)
+        {
+            if (clientStatus == null)
+            {
+                throw new ArgumentNullException(nameof(clientStatus));
+            }
+
+            return clientStatus.DisconnectAsync(DefaultDisconnectOptions);
+        }
+    }
+}

Source/MQTTnet/Server/Stopping/MqttServerStopOptions.cs

@@ -0,0 +1,24 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Protocol;
+using MQTTnet.Server.Disconnecting;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttServerStopOptions
+    {
+        /// <summary>
+        ///     These disconnect options are sent to every connected client via a DISCONNECT packet.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public MqttServerClientDisconnectOptions DefaultClientDisconnectOptions { get; set; } = new MqttServerClientDisconnectOptions
+        {
+            ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
+            UserProperties = null,
+            ReasonString = null,
+            ServerReference = null
+        };
+    }
+}

Source/MQTTnet/Server/Stopping/MqttServerStopOptionsBuilder.cs

@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using MQTTnet.Server.Disconnecting;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttServerStopOptionsBuilder
+    {
+        readonly MqttServerStopOptions _options = new MqttServerStopOptions();
+
+        public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(MqttServerClientDisconnectOptions value)
+        {
+            _options.DefaultClientDisconnectOptions = value;
+            return this;
+        }
+        
+        public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(Action<MqttServerClientDisconnectOptionsBuilder> builder)
+        {
+            if (builder == null)
+            {
+                throw new ArgumentNullException(nameof(builder));
+            }
+
+            var optionsBuilder = new MqttServerClientDisconnectOptionsBuilder();
+            builder.Invoke(optionsBuilder);
+            
+            _options.DefaultClientDisconnectOptions = optionsBuilder.Build();
+            return this;
+        }
+        
+        public MqttServerStopOptions Build()
+        {
+            return _options;
+        }
+    }
+}