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,3 +1,3 @@
-* [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).
+* [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.2.${{github.run_number}}"
+  VERSION: "4.3.3.${{github.run_number}}"
 
 jobs:
   build:

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.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(
-                    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

@@ -55,8 +55,6 @@ namespace MQTTnet.Adapter
 
         public string Endpoint => _channel.Endpoint;
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection => _channel.IsSecureConnection;
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
@@ -389,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

@@ -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
-                // already being gone!
+                // 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);
@@ -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;
-            }
-            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/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/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)
@@ -369,7 +404,7 @@ namespace MQTTnet.Server
         {
             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);
         }
 
@@ -405,4 +445,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/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;
+        }
+    }
+}