Avoid using obsolete code

* implement intercept publish message event on ManagedMqttClient

* rename ShouldPublish to ACceptPublish per PR feedback

.github/workflows/ReleaseNotes.md

@@ -1,3 +1,5 @@
-* [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] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919).
+* [ManagedClient] Added support for intercepting (and cancelling) publish messages (#1915, thanks to @hannasm).
+* [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:

MQTTnet.sln.DotSettings

@@ -230,6 +230,7 @@
 	<s:String x:Key="/Default/CodeStyle/FileHeader/FileHeaderText/@EntryValue">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.</s:String>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EFeature_002EServices_002ECodeCleanup_002EFileHeader_002EFileHeaderSettingsMigrate/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>

Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -48,7 +48,6 @@ namespace MQTTnet.AspNetCore.Tests
         {
             var mockup = new ConnectionHandlerMockup();
 
-
             using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
                        .UseStartup<Startup>()
                        .ConfigureServices(

Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -6,7 +6,6 @@ using MQTTnet.Adapter;
 using MQTTnet.AspNetCore.Client.Tcp;
 using MQTTnet.Formatter;
 using System;
-using System.Net;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 
@@ -22,8 +21,7 @@ namespace MQTTnet.AspNetCore.Client
             {
                 case MqttClientTcpOptions tcpOptions:
                     {
-                        var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
-                        var tcpConnection = new TcpConnection(endpoint);
+                        var tcpConnection = new SocketConnection(tcpOptions.RemoteEndpoint);
                         
                         var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535));
                         return new MqttConnectionContext(formatter, tcpConnection);

Source/MQTTnet.AspnetCore/Client/Tcp/SocketConnection.cs

@@ -15,15 +15,15 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore.Client.Tcp
 {
-    public class TcpConnection : ConnectionContext
+    public sealed class SocketConnection : ConnectionContext
     {
-        private volatile bool _aborted;
-        private readonly EndPoint _endPoint;
-        private SocketSender _sender;
-        private SocketReceiver _receiver;
+        volatile bool _aborted;
+        readonly EndPoint _endPoint;
+        SocketSender _sender;
+        SocketReceiver _receiver;
 
-        private Socket _socket;
-        private IDuplexPipe _application;
+        Socket _socket;
+        IDuplexPipe _application;
 
         public bool IsConnected { get; private set; }
         public override string ConnectionId { get; set; }
@@ -31,12 +31,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
         public override IDictionary<object, object> Items { get; set; }
         public override IDuplexPipe Transport { get; set; }
 
-        public TcpConnection(EndPoint endPoint)
+        public SocketConnection(EndPoint endPoint)
         {
             _endPoint = endPoint;
         }
 
-        public TcpConnection(Socket socket)
+        public SocketConnection(Socket socket)
         {
             _socket = socket;
             _endPoint = socket.RemoteEndPoint;
@@ -87,7 +87,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             IsConnected = true;
         }
 
-        private async Task ExecuteAsync()
+        async Task ExecuteAsync()
         {
             Exception sendError = null;
             try
@@ -114,7 +114,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
             catch (Exception ex)
             {
-                Console.WriteLine($"Unexpected exception in {nameof(TcpConnection)}.{nameof(StartAsync)}: " + ex);
+                Console.WriteLine($"Unexpected exception in {nameof(SocketConnection)}.{nameof(StartAsync)}: " + ex);
             }
             finally
             {
@@ -122,7 +122,8 @@ namespace MQTTnet.AspNetCore.Client.Tcp
                 _application.Input.Complete(sendError);
             }
         }
-        private async Task DoReceive()
+
+        async Task DoReceive()
         {
             Exception error = null;
 
@@ -171,7 +172,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
         }
 
-        private async Task ProcessReceives()
+        async Task ProcessReceives()
         {
             while (true)
             {
@@ -204,12 +205,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
         }
 
-        private Exception ConnectionAborted()
+        Exception ConnectionAborted()
         {
             return new MqttCommunicationException("Connection Aborted");
         }
 
-        private async Task<Exception> DoSend()
+        async Task<Exception> DoSend()
         {
             Exception error = null;
 
@@ -240,7 +241,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             return error;
         }
 
-        private async Task ProcessSends()
+        async Task ProcessSends()
         {
             while (true)
             {

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -33,7 +33,7 @@ namespace MQTTnet.AspNetCore
             PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
             _connection = connection ?? throw new ArgumentNullException(nameof(connection));
 
-            if (!(_connection is TcpConnection tcp) || tcp.IsConnected)
+            if (!(_connection is SocketConnection tcp) || tcp.IsConnected)
             {
                 _input = connection.Transport.Input;
                 _output = connection.Transport.Output;
@@ -83,8 +83,6 @@ namespace MQTTnet.AspNetCore
             }
         }
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection
         {
             get
@@ -111,7 +109,7 @@ namespace MQTTnet.AspNetCore
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
-            if (_connection is TcpConnection tcp && !tcp.IsConnected)
+            if (_connection is SocketConnection tcp && !tcp.IsConnected)
             {
                 await tcp.StartAsync().ConfigureAwait(false);
             }
@@ -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.Benchmarks/TcpPipesBenchmark.cs

@@ -29,12 +29,12 @@ namespace MQTTnet.Benchmarks
 
             var task = Task.Run(() => server.AcceptSocket());
 
-            var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883));
+            var clientConnection = new SocketConnection(new IPEndPoint(IPAddress.Loopback, 1883));
 
             clientConnection.StartAsync().GetAwaiter().GetResult();
             _client = clientConnection.Transport;
 
-            var serverConnection = new TcpConnection(task.GetAwaiter().GetResult());
+            var serverConnection = new SocketConnection(task.GetAwaiter().GetResult());
             serverConnection.StartAsync().GetAwaiter().GetResult();
             _server = serverConnection.Transport;
         }

Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs

@@ -0,0 +1,20 @@
+// 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.Extensions.ManagedClient
+{
+    public sealed class InterceptingPublishMessageEventArgs : EventArgs
+    {
+        public InterceptingPublishMessageEventArgs(ManagedMqttApplicationMessage applicationMessage)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+        }
+
+        public ManagedMqttApplicationMessage ApplicationMessage { get; }
+
+        public bool AcceptPublish { get; set; } = true;
+    }
+}

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -19,6 +19,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
     {
+        readonly AsyncEvent<InterceptingPublishMessageEventArgs> _interceptingPublishMessageEvent = new AsyncEvent<InterceptingPublishMessageEventArgs>();
         readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
         readonly AsyncEvent<ApplicationMessageSkippedEventArgs> _applicationMessageSkippedEvent = new AsyncEvent<ApplicationMessageSkippedEventArgs>();
         readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
@@ -88,6 +89,14 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => _applicationMessageProcessedEvent.RemoveHandler(value);
         }
 
+
+        public event Func<InterceptingPublishMessageEventArgs, Task> InterceptPublishMessageAsync
+        {
+            add => _interceptingPublishMessageEvent.AddHandler(value);
+            remove => _interceptingPublishMessageEvent.RemoveHandler(value);
+        }
+
+
         public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
         {
             add => InternalClient.ApplicationMessageReceivedAsync += value;
@@ -700,9 +709,20 @@ namespace MQTTnet.Extensions.ManagedClient
         async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
+            bool acceptPublish = true;
             try
             {
-                await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                if (_interceptingPublishMessageEvent.HasHandlers)
+                {
+                    var interceptEventArgs = new InterceptingPublishMessageEventArgs(message);
+                    await _interceptingPublishMessageEvent.InvokeAsync(interceptEventArgs).ConfigureAwait(false);
+                    acceptPublish = interceptEventArgs.AcceptPublish;
+                }
+
+                if (acceptPublish)
+                {
+                    await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                }
 
                 using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
                 {

Source/MQTTnet.TestApp/ClientTest.cs

@@ -4,6 +4,7 @@
 
 using MQTTnet.Client;
 using System;
+using System.Net;
 using System.Text;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
@@ -27,7 +28,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     }
                 };
 

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -10,6 +10,7 @@
         <EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <NoWarn>1591;NETSDK1138</NoWarn>
+        <LangVersion>7.3</LangVersion>
     </PropertyGroup>
 
     <ItemGroup>

Source/MQTTnet.TestApp/ManagedClientTest.cs

@@ -7,6 +7,7 @@ using System.Threading.Tasks;
 using System.IO;
 using Newtonsoft.Json;
 using System.Collections.Generic;
+using System.Net;
 using MQTTnet.Client;
 using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Internal;
@@ -28,7 +29,7 @@ namespace MQTTnet.TestApp
                     Credentials = new RandomPassword(),
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "broker.hivemq.com"
+                        RemoteEndpoint = new DnsEndPoint("broker.hivemq.com", 0)
                     }
                 },
 

Source/MQTTnet.TestApp/PerformanceTest.cs

@@ -5,13 +5,13 @@
 using System;
 using System.Diagnostics;
 using System.Linq;
+using System.Net;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
-using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet.TestApp
 {
@@ -25,7 +25,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -71,7 +71,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     }
                 };
 
@@ -114,7 +114,10 @@ namespace MQTTnet.TestApp
             {
                 var options = new MqttClientOptions
                 {
-                    ChannelOptions = new MqttClientTcpOptions { Server = "localhost" },
+                    ChannelOptions = new MqttClientTcpOptions
+                    {
+                        RemoteEndpoint = new DnsEndPoint("localhost", 0)
+                    },
                     ClientId = "Client1",
                     CleanSession = true
                 };
@@ -218,7 +221,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -269,7 +272,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -320,7 +323,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -42,45 +42,45 @@ namespace MQTTnet.TestApp
             };
 
             // mqtt.eclipseprojects.io
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "mqtt.eclipseprojects.io TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("mqtt.eclipseprojects.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "mqtt.eclipseprojects.io WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:80/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13",
+            await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
             
-            await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
+            await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build(),
                     true);
 #endif
 
             // test.mosquitto.org
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP - Authenticated",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1884).WithCredentials("rw", "readwrite").WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP TLS12",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("test.mosquitto.org TCP TLS13",
+            await ExecuteTestsAsync("test.mosquitto.org TCP TLS13",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP TLS12 - Authenticated",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8885)
                     .WithCredentials("rw", "readwrite")
@@ -88,67 +88,67 @@ namespace MQTTnet.TestApp
                     .WithTlsOptions(unsafeTls12)
                     .Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS TLS12",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS TLS12 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
                 true);
 
             // broker.emqx.io
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io TCP TLS12",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("broker.emqx.io TCP TLS13",
+            await ExecuteTestsAsync("broker.emqx.io TCP TLS13",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8083/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS TLS12",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS TLS12 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
                 true);
 
             // broker.hivemq.com
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
@@ -160,6 +160,15 @@ namespace MQTTnet.TestApp
             Console.ReadLine();
         }
 
+        static async Task ExecuteTestsAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
+        {
+            options.ProtocolVersion = MqttProtocolVersion.V311;
+            await ExecuteTestAsync(name + " V3.1.1", options, useWebSocket4Net);
+
+            options.ProtocolVersion = MqttProtocolVersion.V500;
+            await ExecuteTestAsync(name + " V5.0.0", options, useWebSocket4Net);
+        }
+        
         static async Task ExecuteTestAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
         {
             try

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

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

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

@@ -437,6 +437,109 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_PreventingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = false;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_AllowingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = true;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
         // This case also serves as a regression test for the previous behavior which re-published
         // each and every existing subscriptions with every new subscription that was made
         // (causing performance problems and having the visible symptom of retained messages being received again)

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/Client/Options/MqttClientOptionsBuilder.cs

@@ -7,6 +7,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Net;
 using System.Text;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
@@ -17,8 +18,10 @@ namespace MQTTnet.Client
     public sealed class MqttClientOptionsBuilder
     {
         readonly MqttClientOptions _options = new MqttClientOptions();
+        int? _port;
 
         [Obsolete] MqttClientWebSocketProxyOptions _proxyOptions;
+        EndPoint _remoteEndPoint;
 
         MqttClientTcpOptions _tcpOptions;
         MqttClientTlsOptions _tlsOptions;
@@ -63,6 +66,31 @@ namespace MQTTnet.Client
             if (_tcpOptions != null)
             {
                 _tcpOptions.TlsOptions = tlsOptions;
+
+                if (_remoteEndPoint == null)
+                {
+                    throw new ArgumentException("No endpoint is set.");
+                }
+
+                if (_remoteEndPoint is DnsEndPoint dns)
+                {
+                    if (dns.Port == 0)
+                    {
+                        if (_port.HasValue)
+                        {
+                            _remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value);
+                        }
+                        else
+                        {
+                            _remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
+                        }
+                    }
+                }
+
+                if (_tcpOptions.RemoteEndpoint == null)
+                {
+                    _tcpOptions.RemoteEndpoint = _remoteEndPoint;
+                }
             }
             else if (_webSocketOptions != null)
             {
@@ -117,7 +145,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
         {
             if (uri == null)
@@ -130,24 +157,33 @@ namespace MQTTnet.Client
             {
                 case "tcp":
                 case "mqtt":
+                {
                     WithTcpServer(uri.Host, port);
                     break;
+                }
 
                 case "mqtts":
+                {
                     WithTcpServer(uri.Host, port)
                         .WithTlsOptions(
                             o =>
                             {
+                                o.UseTls();
                             });
                     break;
+                }
 
                 case "ws":
                 case "wss":
+                {
                     WithWebSocketServer(o => o.WithUri(uri.ToString()));
                     break;
+                }
 
                 default:
-                    throw new ArgumentException("Unexpected scheme in uri.");
+                {
+                    throw new ArgumentException("Unexpected scheme in URI.");
+                }
             }
 
             if (!string.IsNullOrEmpty(uri.UserInfo))
@@ -161,7 +197,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(string uri)
         {
             return WithConnectionUri(new Uri(uri, UriKind.Absolute));
@@ -190,6 +225,14 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint)
+        {
+            _remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
+            _tcpOptions = new MqttClientTcpOptions();
+
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithExtendedAuthenticationExchangeHandler(IMqttExtendedAuthenticationExchangeHandler handler)
         {
             _options.ExtendedAuthenticationExchangeHandler = handler;
@@ -307,11 +350,12 @@ namespace MQTTnet.Client
 
         public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
         {
-            _tcpOptions = new MqttClientTcpOptions
-            {
-                Server = server,
-                Port = port
-            };
+            _tcpOptions = new MqttClientTcpOptions();
+
+            // The value 0 will be updated when building the options.
+            // This a backward compatibility feature.
+            _remoteEndPoint = new DnsEndPoint(server, port ?? 0);
+            _port = port;
 
             return this;
         }

Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs

@@ -11,34 +11,32 @@ namespace MQTTnet.Client
     {
         public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
 
-        /// <summary>
-        /// Gets the local endpoint (network card) which is used by the client.
-        /// Set it to _null_ to let the OS select the network card.
-        /// </summary>
-        public EndPoint LocalEndpoint { get; set; }
-        
         public int BufferSize { get; set; } = 8192;
 
         /// <summary>
-        /// Gets or sets whether the underlying socket should run in dual mode.
-        /// Leaving this _null_ will avoid setting this value at socket level.
-        /// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
+        ///     Gets or sets whether the underlying socket should run in dual mode.
+        ///     Leaving this _null_ will avoid setting this value at socket level.
+        ///     Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
         /// </summary>
         public bool? DualMode { get; set; }
 
         public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
 
+        /// <summary>
+        ///     Gets the local endpoint (network card) which is used by the client.
+        ///     Set it to _null_ to let the OS select the network card.
+        /// </summary>
+        public EndPoint LocalEndpoint { get; set; }
+
         public bool NoDelay { get; set; } = true;
 
-        public int? Port { get; set; }
-        
-        public string Server { get; set; }
+        public EndPoint RemoteEndpoint { get; set; }
 
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
 
         public override string ToString()
         {
-            return Server + ":" + this.GetPort();
+            return RemoteEndpoint?.ToString() ?? string.Empty;
         }
     }
 }

Source/MQTTnet/Client/Options/MqttClientTcpOptionsExtensions.cs

@@ -1,23 +0,0 @@
-// 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.Client
-{
-    public static class MqttClientTcpOptionsExtensions
-    {
-        public static int GetPort(this MqttClientTcpOptions options)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            if (options.Port.HasValue)
-            {
-                return options.Port.Value;
-            }
-
-            return !(options.TlsOptions?.UseTls ?? false) ? 1883 : 8883;
-        }
-    }
-}

Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs

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

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -170,11 +170,16 @@ namespace MQTTnet.Implementations
             _socket.Bind(localEndPoint);
         }
 
-        public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
+        public Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
         {
-            if (host is null)
+            return ConnectAsync(new DnsEndPoint(host, port), cancellationToken);
+        }
+
+        public async Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            if (endPoint is null)
             {
-                throw new ArgumentNullException(nameof(host));
+                throw new ArgumentNullException(nameof(endPoint));
             }
 
             cancellationToken.ThrowIfCancellationRequested();
@@ -191,15 +196,15 @@ namespace MQTTnet.Implementations
 #endif
 
 #if NET5_0_OR_GREATER
-                await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false);
+                await _socket.ConnectAsync(endPoint, cancellationToken).ConfigureAwait(false);
 #else
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
                 using (cancellationToken.Register(_socketDisposeAction))
                 {
 #if NET452 || NET461
-                    await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, host, port, null).ConfigureAwait(false);
+                    await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
 #else
-                    await _socket.ConnectAsync(host, port).ConfigureAwait(false);
+                    await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
 #endif
                 }
 #endif
@@ -217,7 +222,7 @@ namespace MQTTnet.Implementations
                     throw new MqttCommunicationTimedOutException();
                 }
 
-                throw new MqttCommunicationException($"Error while connecting with host '{host}:{port}'.", socketException);
+                throw new MqttCommunicationException($"Error while connecting host '{endPoint}'.", socketException);
             }
             catch (ObjectDisposedException)
             {

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -17,6 +17,7 @@ using MQTTnet.Channel;
 using MQTTnet.Client;
 using MQTTnet.Server;
 using System.Runtime.InteropServices.WindowsRuntime;
+using System.Net;
 
 namespace MQTTnet.Implementations
 {
@@ -65,9 +66,22 @@ namespace MQTTnet.Implementations
                 _socket.Control.KeepAlive = true;
             }
 
+            string hostName;
+            string serviceName;
+
+            if (_options.RemoteEndpoint is DnsEndPoint dns)
+            {
+                hostName = dns.Host;
+                serviceName = dns.Port.ToString();
+            }
+            else
+            {
+                throw new NotSupportedException("UWP only supports a DNS endpoint.");
+            }
+
             if (_options.TlsOptions?.UseTls != true)
             {
-                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString()).AsTask().ConfigureAwait(false);
+                await _socket.ConnectAsync(new HostName(hostName), serviceName).AsTask().ConfigureAwait(false);
             }
             else
             {
@@ -88,7 +102,7 @@ namespace MQTTnet.Implementations
                     socketProtectionLevel = SocketProtectionLevel.Tls10;
                 }
 
-                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel).AsTask().ConfigureAwait(false);
+                await _socket.ConnectAsync(new HostName(hostName), serviceName, socketProtectionLevel).AsTask().ConfigureAwait(false);
             }
 
             Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -5,6 +5,7 @@
 #if !WINDOWS_UWP
 using System;
 using System.IO;
+using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
 using System.Runtime.ExceptionServices;
@@ -92,7 +93,7 @@ namespace MQTTnet.Implementations
                     socket.DualMode = _tcpOptions.DualMode.Value;
                 }
 
-                await socket.ConnectAsync(_tcpOptions.Server, _tcpOptions.GetPort(), cancellationToken).ConfigureAwait(false);
+                await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
 
                 cancellationToken.ThrowIfCancellationRequested();
 
@@ -103,9 +104,11 @@ namespace MQTTnet.Implementations
                     var targetHost = _tcpOptions.TlsOptions.TargetHost;
                     if (string.IsNullOrEmpty(targetHost))
                     {
-                        targetHost = _tcpOptions.Server;
+                        if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
+                        {
+                            targetHost = dns.Host;    
+                        }
                     }
-
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {

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/Protocol/MqttPorts.cs

@@ -0,0 +1,13 @@
+// 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.Protocol
+{
+    public static class MqttPorts
+    {
+        public const int Default = 1883;
+
+        public const int Secure = 8883;
+    }
+}

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,12 +52,12 @@ namespace MQTTnet.Server
         {
             try
             {
-                _logger.Info("Starting keep alive monitor.");
+                _logger.Info("Starting keep alive monitor");
 
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     TryProcessClients();
-                    Sleep(_options.KeepAliveMonitorInterval);
+                    Sleep(_options.KeepAliveOptions.MonitorInterval);
                 }
             }
             catch (OperationCanceledException)
@@ -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;
+        }
+    }
+}