1846 server mqttserverstopasync doesnt indicate correct reason (#1872)

* add reason code to ext auth

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>

.github/workflows/ReleaseNotes.md

@@ -1,3 +1,7 @@
-* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
 * [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
-* [Client] Fixed handling of unobserved tasks exceptions (#1871).
+* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
+* [Client] Fixed handling of unobserved tasks exceptions (#1871).
+* [Client] Fixed not specified ReasonCode when using _SendExtendedAuthenticationExchangeDataAsync_ (#1882, thanks to @rido-min).
+* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
+* [Server] Added support for custom DISCONNECT packets when stopping the server or disconnect a client (BREAKING CHANGE!, #1846).
+* [Server] Added new property to stop the server from accepting new connections even if it is running (#1846).

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) 
+        readonly MqttFactory _mqttFactory;
-            : base(options, adapters, logger)
+
+        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();
+            // The yield makes sure that the hosted service is considered up and running.
-            return Task.CompletedTask;
+            await Task.Yield();
+
+            _ = StartAsync();
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            return StopAsync();
+            return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
         }
     }
 }

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -4,7 +4,7 @@
 
 using System;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
@@ -14,86 +14,73 @@ 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();
+                throw new ArgumentNullException(nameof(services));
-                configure?.Invoke(serverOptionsBuilder);
+            }
-                return serverOptionsBuilder.Build();
-            });
 
-            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.AddHostedService<MqttHostedServer>();
         }
 
         public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
         {
-            services.AddSingleton(s =>
+            if (services == null)
             {
-                var builder = new AspNetMqttServerOptionsBuilder(s);
+                throw new ArgumentNullException(nameof(services));
-                configure(builder);
+            }
-                return builder.Build();
+
-            });
+            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 +88,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.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/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/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/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/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(
+                .WithTlsOptions(
-                    tls =>
+                    o =>
                     {
-                        tls.UseTls = true;
+                        o.WithSslProtocols(SslProtocols.Tls12).WithCertificateValidationHandler(certValidator);
-                        tls.SslProtocol = SslProtocols.Tls12;
-                        tls.CertificateValidationHandler = certValidator;
                     });
 
             var clientOptions = clientOptionsBuilder.Build();

Source/MQTTnet/Client/MqttClient.cs

@@ -309,6 +309,7 @@ namespace MQTTnet.Client
                 AuthenticationMethod = Options.AuthenticationMethod,
                 AuthenticationData = data.AuthenticationData,
                 ReasonString = data.ReasonString,
+                ReasonCode = data.ReasonCode,
                 UserProperties = data.UserProperties
             };
 
@@ -716,7 +717,7 @@ namespace MQTTnet.Client
                 }
                 catch (Exception exception)
                 {
-                    _logger.Error(exception, "Error while handling application message.");
+                    _logger.Error(exception, "Error while handling application message");
                 }
             }
         }
@@ -835,8 +836,7 @@ namespace MQTTnet.Client
                     _logger.Error(exception, "Error while receiving packets");
                 }
 
-                // The packet dispatcher is set to null when the client is being disposed so it may
+                // The packet dispatcher is set to null when the client is being disposed so it may already being gone!
-                // already being gone!
                 _packetDispatcher?.FailAll(exception);
 
                 await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
@@ -865,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);
                 }
 
@@ -877,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;
@@ -943,7 +943,7 @@ namespace MQTTnet.Client
                 }
                 catch (Exception exception)
                 {
-                    _logger.Warning(exception, "Error while initiating disconnect.");
+                    _logger.Warning(exception, "Error while initiating disconnect");
                 }
             }
         }
@@ -978,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
                 {
@@ -1000,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);
             }
@@ -1017,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;
 
@@ -1055,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/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;
+                return DefaultNormalDisconnection;
-            }
-            else
-            {
-                packet.ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason;
-                packet.UserProperties = clientDisconnectOptions.UserProperties;
-                packet.SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval;
             }
 
-            return packet;
+            return new MqttDisconnectPacket
+            {
+                ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason,
+                UserProperties = clientDisconnectOptions.UserProperties,
+                SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval
+            };
         }
     }
 }

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/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.
+                    // 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. Maybe adding
+                    // This library does not sent a DISCONNECT packet for a normal disconnection.
-                    // a configuration option is requested in the future.
+                    // TODO: Maybe adding a configuration option is requested in the future.
-                    if (reason != MqttDisconnectReasonCode.NormalDisconnection)
+                    if (disconnectOptions != null)
                     {
-                        // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
+                        if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
-                        // token because the entire connection is closed (disposed) as soon as the cancellation
+                            !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
-                        // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+                        {
-                        // at the client!
+                            // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
-                        await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+                            // 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);
@@ -366,7 +372,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 +602,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 +638,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 +742,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/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");
             }
         }
 
@@ -117,18 +126,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/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)
         {
@@ -169,6 +171,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 +193,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 +217,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 +229,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 +261,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 +277,8 @@ namespace MQTTnet.Server
         {
             ThrowIfStarted();
 
+            _isStopping = false;
+            
             _cancellationTokenSource = new CancellationTokenSource();
             var cancellationToken = _cancellationTokenSource.Token;
 
@@ -278,11 +294,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 +311,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 +331,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)
@@ -369,7 +392,7 @@ namespace MQTTnet.Server
         {
             if (disposing)
             {
-                StopAsync().GetAwaiter().GetResult();
+                StopAsync(new MqttServerStopOptions()).GetAwaiter().GetResult();
 
                 foreach (var adapter in _adapters)
                 {
@@ -382,6 +405,11 @@ namespace MQTTnet.Server
 
         Task OnHandleClient(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
+            if (_isStopping || !AcceptNewConnections)
+            {
+                return CompletedTask.Instance;
+            }
+
             return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken);
         }
 
@@ -405,4 +433,4 @@ namespace MQTTnet.Server
             }
         }
     }
-}
+}

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/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>
+        public long BytesReceived => _client.ChannelAdapter.BytesReceived;
-        /// 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.
+        public long BytesSent => _client.ChannelAdapter.BytesSent;
-        /// </summary>
+
-        public string Id => _client.Id;
+        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 Task DisconnectAsync(MqttServerClientDisconnectOptions options)
-
-        public long BytesReceived => _client.ChannelAdapter.BytesReceived;
-        
-        public Task DisconnectAsync()
         {
-            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;
+        }
+    }
+}