Add README to nugets

* Move "Client is not connected" error to it's own exception

* Update MqttClientNotConnectedException.cs

* Update release notes

---------

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

.github/workflows/ReleaseNotes.md

@@ -1,7 +1,4 @@
-* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
+* [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo).
-* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
+* [Core] The package inspector is now fully async (#1941).
-* [Client] Fixed handling of unobserved tasks exceptions (#1871).
+* [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs).
-* [Client] Fixed not specified ReasonCode when using _SendExtendedAuthenticationExchangeDataAsync_ (#1882, thanks to @rido-min).
+* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene).
-* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
-* [Server] Added support for custom DISCONNECT packets when stopping the server or disconnect a client (BREAKING CHANGE!, #1846).
-* [Server] Added new property to stop the server from accepting new connections even if it is running (#1846).

.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.4.${{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>

Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs

@@ -0,0 +1,87 @@
+// 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.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
+using MQTTnet.Client;
+using MQTTnet.Extensions.ManagedClient;
+
+namespace MQTTnet.Samples.ManagedClient;
+
+public sealed class Managed_Client_Subscribe_Samples
+{
+    public static async Task Connect_Client()
+    {
+        /*
+         * This sample creates a simple managed MQTT client and connects to a public broker, subscribe to a topic and verifies subscription result.
+         *
+         * The managed client extends the existing _MqttClient_. It adds the following features.
+         * - Reconnecting when connection is lost.
+         * - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected.
+         */
+        
+        var mqttFactory = new MqttFactory();
+        var subscribed = false;
+
+        using (var managedMqttClient = mqttFactory.CreateManagedMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("broker.hivemq.com")
+                .Build();
+
+            var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
+                .WithClientOptions(mqttClientOptions)
+                .Build();
+
+            await managedMqttClient.StartAsync(managedMqttClientOptions);
+
+            // The application message is not sent. It is stored in an internal queue and
+            // will be sent when the client is connected.
+            await managedMqttClient.EnqueueAsync("Topic", "Payload");
+
+            Console.WriteLine("The managed MQTT client is connected.");
+            
+            // Wait until the queue is fully processed.
+            SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000);
+            
+            Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}");
+
+            managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, subscribed);
+            await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
+
+            SpinWait.SpinUntil(() => subscribed, 1000);
+            Console.WriteLine("Subscription properly done");
+        }
+    }
+
+    private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed)
+    {
+        foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
+        {
+            Console.WriteLine($"Subscription reason {mqttClientSubscribeResult.ReasonString}");
+            foreach (var item in mqttClientSubscribeResult.Items)
+            {
+                Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
+
+                if (item.TopicFilter.Topic == "Topic" && item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS0 && !subscribed)
+                {
+                    subscribed = true;
+                }
+            }
+        }
+
+        foreach (var mqttClientUnsubscribeResult in arg.UnsubscribeResult)
+        {
+            Console.WriteLine($"Unsubscription reason {mqttClientUnsubscribeResult.ReasonString}");
+            foreach (var item in mqttClientUnsubscribeResult.Items)
+            {
+                Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
+            }
+        }
+
+        return Task.CompletedTask;
+    }
+}

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 SocketConnection(tcpOptions.RemoteEndpoint);
-                        var tcpConnection = new TcpConnection(endpoint);
                         
                         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;
+        volatile bool _aborted;
-        private readonly EndPoint _endPoint;
+        readonly EndPoint _endPoint;
-        private SocketSender _sender;
+        SocketSender _sender;
-        private SocketReceiver _receiver;
+        SocketReceiver _receiver;
 
-        private Socket _socket;
+        Socket _socket;
-        private IDuplexPipe _application;
+        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/MQTTnet.AspNetCore.csproj

@@ -11,6 +11,7 @@
         <Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet.AspNetCore</PackageId>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
         <SignAssembly>false</SignAssembly>
         <DelaySign>false</DelaySign>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -37,12 +38,10 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
-            <Pack>True</Pack>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
-    
+
     <ItemGroup>
         <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
     </ItemGroup>
@@ -51,11 +50,11 @@
         <DefineConstants>RELEASE;NETSTANDARD2_0</DefineConstants>
     </PropertyGroup>
 
-    <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1' or '$(TargetFramework)' == 'net5.0' or '$(TargetFramework)' == 'net6.0' or '$(TargetFramework)' == 'net7.0' ">
+    <ItemGroup Condition="'$(TargetFramework)' != 'netstandard2.0' and '$(TargetFramework)' != 'netcoreapp2.1' ">
         <FrameworkReference Include="Microsoft.AspNetCore.App" />
     </ItemGroup>
 
-    <ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1' and '$(TargetFramework)' != 'net5.0' and '$(TargetFramework)' != 'net6.0' and '$(TargetFramework)' != 'net7.0' ">
+    <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0' or '$(TargetFramework)' == 'netcoreapp2.1' ">
         <PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
     </ItemGroup>
 

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/ServiceCollectionExtensions.cs

@@ -5,6 +5,7 @@
 using System;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
@@ -58,7 +59,8 @@ namespace MQTTnet.AspNetCore
             services.TryAddSingleton(new MqttFactory());
 
             services.AddSingleton<MqttHostedServer>();
-            services.AddHostedService<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)

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/IManagedMqttClient.cs

@@ -25,6 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
         
         event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
         
+        event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync;
+        
         IMqttClient InternalClient { get; }
         
         bool IsConnected { get; }

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/MQTTnet.Extensions.ManagedClient.csproj

@@ -13,6 +13,7 @@
         <Description>This is an extension library which provides a managed MQTT client with additional features using MQTTnet.</Description>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet.Extensions.ManagedClient</PackageId>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
         <SignAssembly>false</SignAssembly>
         <DelaySign>false</DelaySign>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -50,10 +51,8 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
-            <Pack>True</Pack>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
     
     <ItemGroup>

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -19,12 +19,14 @@ 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>();
         readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
         readonly AsyncEvent<ManagedProcessFailedEventArgs> _synchronizingSubscriptionsFailedEvent = new AsyncEvent<ManagedProcessFailedEventArgs>();
-        
+        readonly AsyncEvent<SubscriptionsChangedEventArgs> _subscriptionsChangedEvent = new AsyncEvent<SubscriptionsChangedEventArgs>();
+
         readonly MqttNetSourceLogger _logger;
         readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
 
@@ -88,6 +90,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;
@@ -124,6 +134,12 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value);
         }
 
+        public event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync
+        {
+            add => _subscriptionsChangedEvent.AddHandler(value);
+            remove => _subscriptionsChangedEvent.RemoveHandler(value);
+        }
+
         public IMqttClient InternalClient { get; }
 
         public bool IsConnected => InternalClient.IsConnected;
@@ -363,6 +379,14 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
+        async Task HandleSubscriptionsResultAsync(SendSubscribeUnsubscribeResult subscribeUnsubscribeResult)
+        {
+            if (_subscriptionsChangedEvent.HasHandlers)
+            {
+                await _subscriptionsChangedEvent.InvokeAsync(new SubscriptionsChangedEventArgs(subscribeUnsubscribeResult.SubscribeResults, subscribeUnsubscribeResult.UnsubscribeResults)).ConfigureAwait(false);
+            }
+        }
+
         async Task MaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
@@ -464,19 +488,22 @@ namespace MQTTnet.Extensions.ManagedClient
                 if (_reconnectSubscriptions.Any())
                 {
                     topicFilters = new List<MqttTopicFilter>();
-
+                    SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
+                    
                     foreach (var sub in _reconnectSubscriptions)
                     {
                         topicFilters.Add(sub.Value);
 
                         if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                         {
-                            await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                            subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
                             topicFilters.Clear();
+                            await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                         }
                     }
 
-                    await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                    subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                    await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                 }
             }
             catch (Exception exception)
@@ -492,6 +519,7 @@ namespace MQTTnet.Extensions.ManagedClient
             while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
             {
                 List<MqttTopicFilter> subscriptions;
+                SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
                 HashSet<string> unsubscriptions;
 
                 lock (_subscriptions)
@@ -527,12 +555,14 @@ namespace MQTTnet.Extensions.ManagedClient
 
                     if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                     {
-                        await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                        subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
                         addedTopicFilters.Clear();
+                        await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                     }
                 }
-
+                
-                await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
 
                 var removedTopicFilters = new List<string>();
                 foreach (var unSub in unsubscriptions)
@@ -541,12 +571,14 @@ namespace MQTTnet.Extensions.ManagedClient
 
                     if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                     {
-                        await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                        subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
                         removedTopicFilters.Clear();
+                        await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                     }
                 }
 
-                await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
             }
         }
 
@@ -579,8 +611,10 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
+        async Task<SendSubscribeUnsubscribeResult> SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
         {
+            var subscribeResults = new List<MqttClientSubscribeResult>();
+            var unsubscribeResults = new List<MqttClientUnsubscribeResult>();
             try
             {
                 if (removedSubscriptions != null && removedSubscriptions.Any())
@@ -592,7 +626,8 @@ namespace MQTTnet.Extensions.ManagedClient
                         unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription);
                     }
 
-                    await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    unsubscribeResults.Add(unsubscribeResult);
 
                     //clear because these worked, maybe the subscribe below will fail, only report those
                     removedSubscriptions.Clear();
@@ -607,13 +642,16 @@ namespace MQTTnet.Extensions.ManagedClient
                         subscribeOptionsBuilder.WithTopicFilter(addedSubscription);
                     }
 
-                    await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    subscribeResults.Add(subscribeResult);
                 }
             }
             catch (Exception exception)
             {
                 await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false);
             }
+
+            return new SendSubscribeUnsubscribeResult(subscribeResults, unsubscribeResults);
         }
 
         void StartPublishing()
@@ -700,9 +738,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.Extensions.ManagedClient/SendSubscribeUnsubscribeResult.cs

@@ -0,0 +1,23 @@
+// 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.Collections.Generic;
+using MQTTnet.Client;
+
+namespace MQTTnet.Extensions.ManagedClient
+{
+    public sealed class SendSubscribeUnsubscribeResult
+    {
+        public SendSubscribeUnsubscribeResult(List<MqttClientSubscribeResult> subscribeResults, List<MqttClientUnsubscribeResult> unsubscribeResults)
+        {
+            SubscribeResults = subscribeResults ?? throw new ArgumentNullException(nameof(subscribeResults));
+            UnsubscribeResults = unsubscribeResults ?? throw new ArgumentNullException(nameof(unsubscribeResults));
+        }
+        
+        public List<MqttClientSubscribeResult> SubscribeResults { get; private set; }
+        
+        public List<MqttClientUnsubscribeResult> UnsubscribeResults { get; private set; }
+    }
+}

Source/MQTTnet.Extensions.ManagedClient/SubscriptionsChangedEventArgs.cs

@@ -0,0 +1,23 @@
+// 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.Client;
+using System;
+using System.Collections.Generic;
+
+namespace MQTTnet.Extensions.ManagedClient
+{
+    public sealed class SubscriptionsChangedEventArgs : EventArgs
+    {
+        public SubscriptionsChangedEventArgs(List<MqttClientSubscribeResult> subscribeResult, List<MqttClientUnsubscribeResult> unsubscribeResult)
+        {
+            SubscribeResult = subscribeResult ?? throw new ArgumentNullException(nameof(subscribeResult));
+            UnsubscribeResult = unsubscribeResult ?? throw new ArgumentNullException(nameof(unsubscribeResult));
+        }
+
+        public List<MqttClientSubscribeResult> SubscribeResult { get; }
+
+        public List<MqttClientUnsubscribeResult> UnsubscribeResult { get; }
+    }
+}

Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj

@@ -13,6 +13,7 @@
         <Description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</Description>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet.Extensions.Rpc</PackageId>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
         <SignAssembly>false</SignAssembly>
         <DelaySign>false</DelaySign>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
-            <Pack>True</Pack>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
     
     <ItemGroup>

Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj

@@ -13,6 +13,7 @@
         <Description>This is an extension library which allows using _WebSocket4Net_ as transport for MQTTnet clients.</Description>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet.Extensions.WebSocket4Net</PackageId>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
         <SignAssembly>false</SignAssembly>
         <DelaySign>false</DelaySign>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
-            <Pack>True</Pack>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
     
     <ItemGroup>

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -76,7 +76,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
             List<KeyValuePair<string, string>> customHeaders = null;
             if (_webSocketOptions.RequestHeaders != null)
             {
-                customHeaders = _webSocketOptions.RequestHeaders.Select(i => new KeyValuePair<string, string>(i.Key, i.Value)).ToList();
+                customHeaders = _webSocketOptions.RequestHeaders.ToList();
             }
 
             EndPoint proxy = null;

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/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_Connection_Tests.cs

@@ -65,7 +65,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
                         return CompletedTask.Instance;
                     };
 
-                    await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build());
+                    await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").Build());
 
                     Assert.Fail("Must fail!");
                 }
@@ -175,7 +175,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 testEnvironment.IgnoreClientLogErrors = true;
 
                 var client = testEnvironment.CreateClient();
-                var options = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithTimeout(TimeSpan.FromSeconds(2)).Build();
+                var options = new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").WithTimeout(TimeSpan.FromSeconds(2)).Build();
 
                 try
                 {
@@ -240,9 +240,9 @@ namespace MQTTnet.Tests.Clients.MqttClient
                     await mqttClient.SubscribeAsync("test", MqttQualityOfServiceLevel.AtLeastOnce);
                 }
             }
-            catch (MqttCommunicationException exception)
+            catch (MqttClientNotConnectedException exception)
             {
-                if (exception.Message == "The client is not connected.")
+                if (exception.Message == "The MQTT client is not connected.")
                 {
                     return;
                 }

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; }
@@ -143,7 +141,8 @@ namespace MQTTnet.Adapter
 
             try
             {
-                PacketInspector?.BeginReceivePacket();
+                var localPacketInspector = PacketInspector;
+                localPacketInspector?.BeginReceivePacket();
 
                 ReceivedMqttPacket receivedPacket;
                 var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -161,7 +160,10 @@ namespace MQTTnet.Adapter
                     return null;
                 }
 
-                PacketInspector?.EndReceivePacket();
+                if (localPacketInspector != null)
+                {
+                    await localPacketInspector.EndReceivePacket().ConfigureAwait(false);
+                }
 
                 Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
 
@@ -217,8 +219,13 @@ namespace MQTTnet.Adapter
                 try
                 {
                     var packetBuffer = PacketFormatterAdapter.Encode(packet);
-                    PacketInspector?.BeginSendPacket(packetBuffer);
 
+                    var localPacketInspector = PacketInspector;
+                    if (localPacketInspector != null)
+                    {
+                        await localPacketInspector.BeginSendPacket(packetBuffer).ConfigureAwait(false);    
+                    }
+                    
                     _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
 
                     if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
@@ -389,53 +396,45 @@ namespace MQTTnet.Adapter
                 return ReceivedMqttPacket.Empty;
             }
 
-            IsReadingPacket = true;
+            var fixedHeader = readFixedHeaderResult.FixedHeader;
-            try
+            if (fixedHeader.RemainingLength == 0)
             {
-                var fixedHeader = readFixedHeaderResult.FixedHeader;
+                return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
-                if (fixedHeader.RemainingLength == 0)
+            }
+
+            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 readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
-                var body = new byte[bodyLength];
 
-                var bodyOffset = 0;
+                if (cancellationToken.IsCancellationRequested)
-                var chunkSize = Math.Min(ReadBufferSize, bodyLength);
-
-                do
                 {
-                    var bytesLeft = body.Length - bodyOffset;
+                    return ReceivedMqttPacket.Empty;
-                    if (chunkSize > bytesLeft)
+                }
-                    {
-                        chunkSize = bytesLeft;
-                    }
 
-                    var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
+                if (readBytes == 0)
+                {
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    if (cancellationToken.IsCancellationRequested)
+                bodyOffset += readBytes;
-                    {
+            } while (bodyOffset < bodyLength);
-                        return ReceivedMqttPacket.Empty;
-                    }
 
-                    if (readBytes == 0)
+            PacketInspector?.FillReceiveBuffer(body);
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
 
-                    bodyOffset += readBytes;
+            var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
-                } while (bodyOffset < bodyLength);
+            return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
-
-                PacketInspector?.FillReceiveBuffer(body);
-
-                var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
-                return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
-            }
-            finally
-            {
-                IsReadingPacket = false;
-            }
         }
 
         static bool WrapAndThrowException(Exception exception)

Source/MQTTnet/Adapter/MqttPacketInspector.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.IO;
+using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
@@ -12,16 +13,20 @@ namespace MQTTnet.Adapter
 {
     public sealed class MqttPacketInspector
     {
-        readonly MqttNetSourceLogger _logger;
         readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent;
-        
+        readonly MqttNetSourceLogger _logger;
+
         MemoryStream _receivedPacketBuffer;
-        
+
         public MqttPacketInspector(AsyncEvent<InspectMqttPacketEventArgs> asyncEvent, IMqttNetLogger logger)
         {
             _asyncEvent = asyncEvent ?? throw new ArgumentNullException(nameof(asyncEvent));
-            
+
-            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (logger == null)
+            {
+                throw new ArgumentNullException(nameof(logger));
+            }
+
             _logger = logger.WithSource(nameof(MqttPacketInspector));
         }
 
@@ -36,28 +41,15 @@ namespace MQTTnet.Adapter
             {
                 _receivedPacketBuffer = new MemoryStream();
             }
-            
+
             _receivedPacketBuffer?.SetLength(0);
         }
 
-        public void EndReceivePacket()
+        public Task BeginSendPacket(MqttPacketBuffer buffer)
         {
             if (!_asyncEvent.HasHandlers)
             {
-                return;
+                return CompletedTask.Instance;
-            }
-
-            var buffer = _receivedPacketBuffer.ToArray();
-            _receivedPacketBuffer.SetLength(0);
-
-            InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
-        }
-
-        public void BeginSendPacket(MqttPacketBuffer buffer)
-        {
-            if (!_asyncEvent.HasHandlers)
-            {
-                return;
             }
 
             // Create a copy of the actual packet so that the inspector gets no access
@@ -65,7 +57,20 @@ namespace MQTTnet.Adapter
             // intended for debugging etc. so that this is OK.
             var bufferCopy = buffer.ToArray();
 
-            InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
+            return InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
+        }
+
+        public Task EndReceivePacket()
+        {
+            if (!_asyncEvent.HasHandlers)
+            {
+                return CompletedTask.Instance;
+            }
+
+            var buffer = _receivedPacketBuffer.ToArray();
+            _receivedPacketBuffer.SetLength(0);
+
+            return InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
         }
 
         public void FillReceiveBuffer(byte[] buffer)
@@ -74,16 +79,16 @@ namespace MQTTnet.Adapter
             {
                 return;
             }
-            
+
             _receivedPacketBuffer?.Write(buffer, 0, buffer.Length);
         }
 
-        void InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
+        async Task InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
         {
             try
             {
                 var eventArgs = new InspectMqttPacketEventArgs(direction, buffer);
-                _asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult();
+                await _asyncEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Client/MqttClient.cs

@@ -930,7 +930,7 @@ namespace MQTTnet.Client
 
         static void ThrowNotConnected()
         {
-            throw new MqttCommunicationException("The client is not connected.");
+            throw new MqttClientNotConnectedException();
         }
 
         void TryInitiateDisconnect()

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;
@@ -22,7 +23,9 @@ namespace MQTTnet.Client
 
         MqttClientTcpOptions _tcpOptions;
         MqttClientTlsOptions _tlsOptions;
-
+        EndPoint _remoteEndPoint;
+        int? _port;
+        
         [Obsolete] MqttClientOptionsBuilderTlsParameters _tlsParameters;
 
         MqttClientWebSocketOptions _webSocketOptions;
@@ -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)
             {
@@ -307,11 +335,20 @@ namespace MQTTnet.Client
 
         public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
         {
-            _tcpOptions = new MqttClientTcpOptions
+            _tcpOptions = new MqttClientTcpOptions();
-            {
+
-                Server = server,
+            // The value 0 will be updated when building the options.
-                Port = port
+            // This a backward compatibility feature.
-            };
+            _remoteEndPoint = new DnsEndPoint(server, port ?? 0);
+            _port = port;
+
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint)
+        {
+            _remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
+            _tcpOptions = new MqttClientTcpOptions();
 
             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.
+        ///     Gets or sets whether the underlying socket should run in dual mode.
-        /// Leaving this _null_ will avoid setting this value at socket level.
+        ///     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.
+        ///     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 EndPoint RemoteEndpoint { get; set; }
-        
-        public string Server { 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/Exceptions/MqttClientNotConnectedException.cs

@@ -0,0 +1,19 @@
+// 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.Exceptions
+{
+    public class MqttClientNotConnectedException : MqttCommunicationException
+    {
+        public MqttClientNotConnectedException() : base("The MQTT client is not connected.")
+        {
+        }
+
+        public MqttClientNotConnectedException(Exception innerException) : base("The MQTT client is not connected.", innerException)
+        {
+        }
+    }
+}

Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs

@@ -279,7 +279,7 @@ namespace MQTTnet.Formatter.V5
             _propertiesWriter.WriteReasonString(packet.ReasonString);
             _propertiesWriter.WriteUserProperties(packet.UserProperties);
 
-            if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
+            if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
             {
                 _bufferWriter.WriteByte((byte)packet.ReasonCode);
                 _propertiesWriter.WriteTo(_bufferWriter);
@@ -375,7 +375,7 @@ namespace MQTTnet.Formatter.V5
 
             _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier);
 
-            if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
+            if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
             {
                 _bufferWriter.WriteByte((byte)packet.ReasonCode);
                 _propertiesWriter.WriteTo(_bufferWriter);
@@ -534,4 +534,4 @@ namespace MQTTnet.Formatter.V5
             }
         }
     }
-}
+}

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

@@ -23,6 +23,7 @@
         <Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet</PackageId>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
         <DelaySign>false</DelaySign>
         <SignAssembly>false</SignAssembly>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -69,10 +70,8 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
-            <Pack>True</Pack>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
 
     <ItemGroup>

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/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/MqttClientSessionsManager.cs

@@ -244,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);
                     }
 

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

@@ -57,7 +57,7 @@ namespace MQTTnet.Server
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     TryProcessClients();
-                    Sleep(_options.KeepAliveMonitorInterval);
+                    Sleep(_options.KeepAliveOptions.MonitorInterval);
                 }
             }
             catch (OperationCanceledException)
@@ -107,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.

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

@@ -51,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);
@@ -135,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);

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;