Restore old signing to get out hotfix release (#2004)

* Fix endpoint handling

* Fix different behavior for Mono framework.

* Restore old properties to avoid breaking changes

.editorconfig

@@ -40,7 +40,6 @@ dotnet_style_qualification_for_method = false:suggestion
 dotnet_style_qualification_for_property = false:suggestion
 
 # ReSharper properties
-resharper_autodetect_indent_settings = true
 resharper_default_private_modifier = implicit
 resharper_place_accessorholder_attribute_on_same_line = false
 resharper_use_indent_from_vs = false
@@ -61,7 +60,7 @@ resharper_web_config_wrong_module_highlighting = warning
 [*]
 charset = utf-8
 end_of_line = crlf
-trim_trailing_whitespace = false
+trim_trailing_whitespace = true
 insert_final_newline = false
 indent_style = space
 indent_size = 4

.github/workflows/ReleaseNotes.md

@@ -1,3 +1,2 @@
-* [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
+* [Client] Fix _None of the discovered or specified addresses match the socket address family._ (#1997).
-* [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).
+* [Client] Remove the obsolete attribute from the _WithConnectionUri_ methods (#1979).
-* [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.3.${{github.run_number}}"
+  VERSION: "4.3.5.${{github.run_number}}"
 
 jobs:
   build:

MQTTnet.sln

@@ -7,9 +7,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet", "Source\MQTTnet\M
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
 	ProjectSection(SolutionItems) = preProject
+		CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
 		LICENSE = LICENSE
 		README.md = README.md
-		CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
@@ -20,17 +20,19 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.ManagedC
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.WebSocket4Net", "Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj", "{2BD01D53-4CA5-4142-BE8D-313876395E3E}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Samples", "Samples\MQTTnet.Samples.csproj", "{71CF35F5-3327-4A91-AAF4-5340F6701771}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Samples", "Samples\MQTTnet.Samples.csproj", "{71CF35F5-3327-4A91-AAF4-5340F6701771}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Tests", "Source\MQTTnet.Tests\MQTTnet.Tests.csproj", "{B270F32A-9F3E-42EE-A989-813E35E29ADB}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Source\MQTTnet.Tests\MQTTnet.Tests.csproj", "{B270F32A-9F3E-42EE-A989-813E35E29ADB}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.AspNetCore.Tests", "Source\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj", "{A238BBBF-C75F-482D-9CC3-BB34ABA9B675}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore.Tests", "Source\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj", "{A238BBBF-C75F-482D-9CC3-BB34ABA9B675}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Benchmarks", "Source\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Benchmarks", "Source\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.TestApp", "Source\MQTTnet.TestApp\MQTTnet.TestApp.csproj", "{175D5340-CC5B-4542-939D-4E7D15A0BC8D}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp", "Source\MQTTnet.TestApp\MQTTnet.TestApp.csproj", "{175D5340-CC5B-4542-939D-4E7D15A0BC8D}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.AspTestApp", "Source\MQTTnet.AspTestApp\MQTTnet.AspTestApp.csproj", "{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspTestApp", "Source\MQTTnet.AspTestApp\MQTTnet.AspTestApp.csproj", "{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.TopicTemplate", "Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj", "{374D5861-F7A6-42A2-8CBE-040846000EF9}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -82,12 +84,14 @@ Global
 		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.Build.0 = Release|Any CPU
+		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
-	GlobalSection(NestedProjects) = preSolution
-	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
 	EndGlobalSection

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

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk.Web">
 
     <PropertyGroup>
-        <TargetFramework>net6.0</TargetFramework>
+        <TargetFramework>net7.0</TargetFramework>
         <Nullable>enable</Nullable>
         <ImplicitUsings>enable</ImplicitUsings>
         <IsPackable>false</IsPackable>

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

@@ -51,11 +51,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;
@@ -109,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);
             }

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -40,13 +40,10 @@ namespace MQTTnet.AspNetCore
                 throw new ArgumentNullException(nameof(services));
             }
 
-            if (configure == null)
-            {
-                throw new ArgumentNullException(nameof(configure));
-            }
-
             var serverOptionsBuilder = new MqttServerOptionsBuilder();
-            configure.Invoke(serverOptionsBuilder);
+
+            configure?.Invoke(serverOptionsBuilder);
+
             var options = serverOptionsBuilder.Build();
 
             return AddHostedMqttServer(services, options);

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

@@ -19,15 +19,17 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
     {
+        readonly MqttNetSourceLogger _logger;
+
+        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>();
-
         readonly AsyncLock _messageQueueLock = new AsyncLock();
 
         /// <summary>
@@ -88,6 +90,12 @@ 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 +132,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;
@@ -353,6 +367,13 @@ namespace MQTTnet.Extensions.ManagedClient
             return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;
         }
 
+        CancellationTokenSource NewTimeoutToken(CancellationToken linkedToken)
+        {
+            var newTimeoutToken = CancellationTokenSource.CreateLinkedTokenSource(linkedToken);
+            newTimeoutToken.CancelAfter(Options.ClientOptions.Timeout);
+            return newTimeoutToken;
+        }
+
         async Task HandleSubscriptionExceptionAsync(Exception exception, List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
         {
             _logger.Warning(exception, "Synchronizing subscriptions failed.");
@@ -363,6 +384,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
@@ -387,7 +416,7 @@ namespace MQTTnet.Extensions.ManagedClient
                     {
                         if (_isCleanDisconnect)
                         {
-                            using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
+                            using (var disconnectTimeout = NewTimeoutToken(CancellationToken.None))
                             {
                                 await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
                             }
@@ -437,7 +466,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
                     cancellationToken.ThrowIfCancellationRequested();
 
-                    await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
+                    await TryPublishQueuedMessageAsync(message, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -453,7 +482,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        async Task PublishReconnectSubscriptionsAsync()
+        async Task PublishReconnectSubscriptionsAsync(CancellationToken cancellationToken)
         {
             _logger.Info("Publishing subscriptions at reconnect");
 
@@ -464,6 +493,7 @@ namespace MQTTnet.Extensions.ManagedClient
                 if (_reconnectSubscriptions.Any())
                 {
                     topicFilters = new List<MqttTopicFilter>();
+                    SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
 
                     foreach (var sub in _reconnectSubscriptions)
                     {
@@ -471,12 +501,14 @@ namespace MQTTnet.Extensions.ManagedClient
 
                         if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                         {
-                            await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                            subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null, cancellationToken).ConfigureAwait(false);
                             topicFilters.Clear();
+                            await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                         }
                     }
 
-                    await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                    subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null, cancellationToken).ConfigureAwait(false);
+                    await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                 }
             }
             catch (Exception exception)
@@ -492,6 +524,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 +560,14 @@ namespace MQTTnet.Extensions.ManagedClient
 
                     if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                     {
-                        await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                        subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null, cancellationToken).ConfigureAwait(false);
                         addedTopicFilters.Clear();
+                        await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                     }
                 }
 
-                await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null, cancellationToken).ConfigureAwait(false);
+                await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
 
                 var removedTopicFilters = new List<string>();
                 foreach (var unSub in unsubscriptions)
@@ -541,12 +576,14 @@ namespace MQTTnet.Extensions.ManagedClient
 
                     if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                     {
-                        await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                        subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters, cancellationToken).ConfigureAwait(false);
                         removedTopicFilters.Clear();
+                        await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
                     }
                 }
 
-                await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters, cancellationToken).ConfigureAwait(false);
+                await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
             }
         }
 
@@ -560,7 +597,7 @@ namespace MQTTnet.Extensions.ManagedClient
             MqttClientConnectResult connectResult = null;
             try
             {
-                using (var connectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
+                using (var connectTimeout = NewTimeoutToken(cancellationToken))
                 {
                     connectResult = await InternalClient.ConnectAsync(Options.ClientOptions, connectTimeout.Token).ConfigureAwait(false);
                 }
@@ -579,8 +616,10 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
+        async Task<SendSubscribeUnsubscribeResult> SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions, CancellationToken cancellationToken)
         {
+            var subscribeResults = new List<MqttClientSubscribeResult>();
+            var unsubscribeResults = new List<MqttClientUnsubscribeResult>();
             try
             {
                 if (removedSubscriptions != null && removedSubscriptions.Any())
@@ -592,7 +631,11 @@ namespace MQTTnet.Extensions.ManagedClient
                         unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription);
                     }
 
-                    await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    using (var unsubscribeTimeout = NewTimeoutToken(cancellationToken))
+                    {
+                        var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build(), unsubscribeTimeout.Token).ConfigureAwait(false);
+                        unsubscribeResults.Add(unsubscribeResult);
+                    }
 
                     //clear because these worked, maybe the subscribe below will fail, only report those
                     removedSubscriptions.Clear();
@@ -607,13 +650,19 @@ namespace MQTTnet.Extensions.ManagedClient
                         subscribeOptionsBuilder.WithTopicFilter(addedSubscription);
                     }
 
-                    await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                    using (var subscribeTimeout = NewTimeoutToken(cancellationToken))
+                    {
+                        var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build(), subscribeTimeout.Token).ConfigureAwait(false);
+                        subscribeResults.Add(subscribeResult);
+                    }
                 }
             }
             catch (Exception exception)
             {
                 await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false);
             }
+
+            return new SendSubscribeUnsubscribeResult(subscribeResults, unsubscribeResults);
         }
 
         void StartPublishing()
@@ -667,7 +716,7 @@ namespace MQTTnet.Extensions.ManagedClient
                 }
                 else if (connectionState == ReconnectionResult.Reconnected)
                 {
-                    await PublishReconnectSubscriptionsAsync().ConfigureAwait(false);
+                    await PublishReconnectSubscriptionsAsync(cancellationToken).ConfigureAwait(false);
                     StartPublishing();
                 }
                 else if (connectionState == ReconnectionResult.Recovered)
@@ -697,12 +746,26 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
+        async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message, CancellationToken cancellationToken)
         {
             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)
+                {
+                    using (var publishTimeout = NewTimeoutToken(cancellationToken))
+                    {
+                        await InternalClient.PublishAsync(message.ApplicationMessage, publishTimeout.Token).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/IMqttRpcClient.cs

@@ -1,3 +1,7 @@
+// 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 System.Threading;

Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj

@@ -0,0 +1,69 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
+        
+        <AssemblyName>MQTTnet.Extensions.TopicTemplate</AssemblyName>
+        <RootNamespace>MQTTnet.Extensions.TopicTemplate</RootNamespace>
+        <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
+        <Company>The contributors of MQTTnet</Company>
+        <Product>MQTTnet</Product>
+        <Description>
+           This is an extension library which provides mqtt topic templating logic to support dispatch,
+           routing and similar functionality based on the well known moustache syntax, aka
+           AsyncAPI dynamic channel address.
+        </Description>
+        <Authors>The contributors of MQTTnet</Authors>
+        <PackageId>MQTTnet.Extensions.TopicTemplate</PackageId>
+        <SignAssembly>false</SignAssembly>
+        <DelaySign>false</DelaySign>
+        <PublishRepositoryUrl>true</PublishRepositoryUrl>
+        <IncludeSymbols>true</IncludeSymbols>
+        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
+        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
+        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
+        <RepositoryType>git</RepositoryType>
+        <PackageTags>MQTT Message Queue Telemetry MQTTClient Messaging Routing AsyncAPI Template</PackageTags>
+        <NeutralLanguage>en-US</NeutralLanguage>
+        <PackageIcon>nuget.png</PackageIcon>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
+        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
+        <GenerateDocumentationFile>true</GenerateDocumentationFile>
+        <NoWarn>1591;NETSDK1138</NoWarn>
+    </PropertyGroup>
+
+    <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
+        <CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
+        <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
+        <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
+        <TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
+        <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
+        <TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
+        <TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
+        <DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
+        <DefaultLanguage>en</DefaultLanguage>
+        <LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <None Include="..\..\Images\nuget.png">
+            <Pack>True</Pack>
+            <PackagePath>\</PackagePath>
+        </None>
+    </ItemGroup>
+    
+    <ItemGroup>
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
+        <PackageReference Include="System.ValueTuple" Version="4.5.0" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
+    </ItemGroup>
+
+</Project>

Source/MQTTnet.Extensions.TopicTemplate/MqttTopicTemplate.cs

@@ -0,0 +1,345 @@
+// 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 System.Linq;
+using System.Text.RegularExpressions;
+using System.Threading;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Extensions.TopicTemplate
+{
+    /// <summary>
+    ///     A topic template is an MQTT topic filter string that may contain
+    ///     segments in curly braces called parameters. This well-known
+    ///     'moustache' syntax also matches AsyncAPI Channel Address Expressions.
+    ///     The topic template is designed to support dynamic subscription/publication,
+    ///     message-topic matching and routing. It is intended to be more convenient
+    ///     than String.Format() for aforementioned purposes.
+    /// </summary>
+    /// <example>
+    ///     topic/subtopic/{parameter}/{otherParameter}
+    /// </example>
+    public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
+    {
+        static readonly Regex MoustacheRegex = new Regex("{([^/]+?)}", RegexOptions.Compiled);
+        
+        readonly string[] _parameterSegments;
+        
+        string _topicFilter;
+        
+        /// <summary>
+        ///     Create a topic template from an mqtt topic filter with moustache placeholders.
+        /// </summary>
+        /// <param
+        ///     name="topicTemplate">
+        /// </param>
+        /// <exception
+        ///     cref="ArgumentNullException">
+        /// </exception>
+        public MqttTopicTemplate(string topicTemplate)
+        {
+            if (topicTemplate == null)
+            {
+                throw new ArgumentNullException(nameof(topicTemplate));
+            }
+            
+            MqttTopicValidator.ThrowIfInvalidSubscribe(topicTemplate);
+            
+            Template = topicTemplate;
+            _parameterSegments = topicTemplate.Split(MqttTopicFilterComparer.LevelSeparator)
+                .Select(segment => MoustacheRegex.Match(segment).Groups[1].Value)
+                .Select(s => s.Length > 0 ? s : null)
+                .ToArray();
+        }
+        
+        /// <summary>
+        ///     Yield the template parameter names.
+        /// </summary>
+        public IEnumerable<string> Parameters => _parameterSegments.Where(s => s != null);
+        
+        /// <summary>
+        ///     The topic template string representation, e.g. A/B/{foo}/D.
+        /// </summary>
+        public string Template { get; }
+        
+        /// <summary>
+        ///     The topic template as an MQTT topic filter (+ substituted for all parameters). If the template
+        ///     ends with a multi-level wildcard (hash), this will be reflected here.
+        /// </summary>
+        public string TopicFilter
+        {
+            get
+            {
+                LazyInitializer.EnsureInitialized(ref _topicFilter, () => MoustacheRegex.Replace(Template, MqttTopicFilterComparer.SingleLevelWildcard.ToString()));
+                return _topicFilter;
+            }
+        }
+        
+        /// <summary>
+        ///     Return the topic filter of this template, ending with a multi-level wildcard (hash).
+        /// </summary>
+        public string TopicTreeRootFilter
+        {
+            get
+            {
+                var filter = TopicFilter;
+                // append slash if neccessary
+                if (filter.Length > 0 && !filter.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) &&
+                    !filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
+                {
+                    filter += MqttTopicFilterComparer.LevelSeparator;
+                }
+                
+                // append hash if neccessary
+                if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
+                {
+                    filter += MqttTopicFilterComparer.MultiLevelWildcard;
+                }
+                
+                return filter;
+            }
+        }
+        
+        public bool Equals(MqttTopicTemplate other)
+        {
+            return other != null && Template == other.Template;
+        }
+        
+        public override bool Equals(object obj)
+        {
+            if (ReferenceEquals(null, obj))
+            {
+                return false;
+            }
+            
+            if (ReferenceEquals(this, obj))
+            {
+                return true;
+            }
+            
+            if (obj.GetType() != GetType())
+            {
+                return false;
+            }
+            
+            return Equals((MqttTopicTemplate)obj);
+        }
+        
+        /// <summary>
+        ///     Determine the shortest common prefix of the given templates. Partial segments
+        ///     are not returned.
+        /// </summary>
+        /// <param
+        ///     name="templates">
+        ///     topic templates
+        /// </param>
+        /// <returns></returns>
+        public static MqttTopicTemplate FindCanonicalPrefix(IEnumerable<MqttTopicTemplate> templates)
+        {
+            string root = null;
+            
+            string CommonPrefix(string a, string b)
+            {
+                var maxIndex = Math.Min(a.Length, b.Length) - 1;
+                for (var i = 0; i <= maxIndex; i++)
+                {
+                    if (a[i] != b[i])
+                    {
+                        return a.Substring(0, i);
+                    }
+                }
+                
+                return a.Substring(0, maxIndex+1);
+            }
+            
+            foreach (string topic in from template in templates select template.Template)
+            {
+                root = root == null ? topic : CommonPrefix(root, topic);
+            }
+            
+            if (string.IsNullOrEmpty(root))
+                return new MqttTopicTemplate(MqttTopicFilterComparer.MultiLevelWildcard.ToString());
+            
+            if (root.Contains(MqttTopicFilterComparer.LevelSeparator) &&
+                !root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) &&
+                !root.EndsWith("}"))
+            {
+                root = root.Substring(0, root.LastIndexOf(MqttTopicFilterComparer.LevelSeparator)+1);
+            }
+            
+            if (root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()))
+                root += MqttTopicFilterComparer.SingleLevelWildcard;
+
+            return new MqttTopicTemplate(root);
+        }
+        
+        public override int GetHashCode()
+        {
+            return Template.GetHashCode();
+        }
+        
+        /// <summary>
+        ///     Test if this topic template matches a given topic.
+        /// </summary>
+        /// <param
+        ///     name="topic">
+        ///     a fully specified topic
+        /// </param>
+        /// <param
+        ///     name="subtree">
+        ///     true to match including the subtree (multi-level wildcard)
+        /// </param>
+        /// <returns>true iff the topic matches the template's filter</returns>
+        /// <exception
+        ///     cref="InvalidOperationException">
+        /// </exception>
+        /// <exception
+        ///     cref="ArgumentException">
+        ///     if the topic is invalid
+        /// </exception>
+        public bool MatchesTopic(string topic, bool subtree = false)
+        {
+            var comparison = MqttTopicFilterComparer.Compare(topic, subtree ? TopicTreeRootFilter : TopicFilter);
+            if (comparison == MqttTopicFilterCompareResult.FilterInvalid)
+            {
+                throw new InvalidOperationException("Invalid filter");
+            }
+            
+            if (comparison == MqttTopicFilterCompareResult.TopicInvalid)
+            {
+                throw new ArgumentException("Invalid topic", nameof(topic));
+            }
+            
+            return comparison == MqttTopicFilterCompareResult.IsMatch;
+        }
+        
+        /// <summary>
+        ///     Extract the parameter values from a topic corresponding to the template
+        ///     parameters. The topic has to match this template.
+        /// </summary>
+        /// <param
+        ///     name="topic">
+        ///     the topic
+        /// </param>
+        /// <returns>an enumeration of (parameter, index, value)</returns>
+        public IEnumerable<(string parameter, int index, string value)> ParseParameterValues(string topic)
+        {
+            if (!MatchesTopic(topic))
+            {
+                throw new ArgumentException("the topic has to match this template", nameof(topic));
+            }
+            
+            return parseParameterValuesInternal(topic);
+        }
+        
+        /// <summary>
+        ///     Extract the parameter values from the message topic corresponding to the template
+        ///     parameters. The message topic has to match this topic template.
+        /// </summary>
+        /// <param
+        ///     name="message">
+        ///     the message
+        /// </param>
+        /// <returns>an enumeration of (parameter, index, value)</returns>
+        public IEnumerable<(string parameter, int index, string value)> ParseParameterValues(MqttApplicationMessage message)
+        {
+            return ParseParameterValues(message.Topic);
+        }
+        
+        /// <summary>
+        ///     Try to set a parameter to a given value. If the parameter is not present,
+        ///     this is returned. The value must not contain slashes.
+        /// </summary>
+        /// <param
+        ///     name="parameter">
+        ///     a template parameter
+        /// </param>
+        /// <param
+        ///     name="value">
+        ///     a string
+        /// </param>
+        /// <returns></returns>
+        public MqttTopicTemplate TrySetParameter(string parameter, string value)
+        {
+            if (parameter != null && _parameterSegments.Contains(parameter))
+            {
+                return WithParameter(parameter, value);
+            }
+            
+            return this;
+        }
+        
+        /// <summary>
+        ///     Replace the given parameter with a single-level wildcard (plus sign).
+        /// </summary>
+        /// <param
+        ///     name="parameter">
+        ///     parameter name
+        /// </param>
+        /// <returns>the topic template (without the parameter)</returns>
+        public MqttTopicTemplate WithoutParameter(string parameter)
+        {
+            return WithParameter(parameter, MqttTopicFilterComparer.SingleLevelWildcard.ToString());
+        }
+        
+        /// <summary>
+        ///     Substitute a parameter with a given value, thus removing the parameter. If the parameter is not present,
+        ///     the method trows. The value must not contain slashes.
+        /// </summary>
+        /// <param
+        ///     name="parameter">
+        ///     a template parameter
+        /// </param>
+        /// <param
+        ///     name="value">
+        ///     a string
+        /// </param>
+        /// <exception
+        ///     cref="ArgumentException">
+        ///     when the parameter is not present
+        /// </exception>
+        /// <returns>the topic template (without the parameter)</returns>
+        public MqttTopicTemplate WithParameter(string parameter, string value)
+        {
+            if (value == null || string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter) || value.Contains(MqttTopicFilterComparer.LevelSeparator) ||
+                value.Contains(MqttTopicFilterComparer.MultiLevelWildcard))
+            {
+                throw new ArgumentException("parameter must exist and value must not contain slashes.");
+            }
+            
+            var moustache = "{" + parameter + "}";
+            return new MqttTopicTemplate(Template.Replace(moustache, value));
+        }
+        
+        /// <summary>
+        ///     Reuse parameters as they are extracted using another topic template on this template
+        ///     when the parameter name matches. Useful
+        ///     for compatibility routing.
+        /// </summary>
+        /// <param
+        ///     name="parameters">
+        /// </param>
+        /// <returns></returns>
+        public MqttTopicTemplate WithParameterValuesFrom(IEnumerable<(string parameter, int index, string value)> parameters)
+        {
+            return parameters.Aggregate(this, (t, p) => t.TrySetParameter(p.parameter, p.value));
+        }
+        
+        IEnumerable<(string parameter, int index, string value)> parseParameterValuesInternal(string topic)
+        {
+            // because we have a match, we know the segment array is at least the template's length
+            var segments = topic.Split(MqttTopicFilterComparer.LevelSeparator);
+            for (var i = 0; i < _parameterSegments.Length; i++)
+            {
+                var name = _parameterSegments[i];
+                if (name != null)
+                {
+                    yield return (name, i, segments[i]);
+                }
+            }
+        }
+    }
+}

Source/MQTTnet.Extensions.TopicTemplate/TopicTemplateExtensions.cs

@@ -0,0 +1,147 @@
+// 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.Linq;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Extensions.TopicTemplate
+{
+    public static class TopicTemplateExtensions
+    {
+        /// <summary>
+        ///     Modify this message builder to respond to a given message. The
+        ///     message's response topic and correlation data are included
+        ///     in the message builder.
+        /// </summary>
+        /// <param
+        ///     name="builder">
+        ///     a message builder
+        /// </param>
+        /// <param
+        ///     name="message">
+        ///     a message with a response topic
+        /// </param>
+        /// <returns>a message builder</returns>
+        /// <see
+        ///     cref="BuildResponse" />
+        public static MqttApplicationMessageBuilder AsResponseTo(this MqttApplicationMessageBuilder builder, MqttApplicationMessage message)
+        {
+            if (!string.IsNullOrEmpty(message.ResponseTopic))
+            {
+                throw new ArgumentException("message does not have a response topic");
+            }
+
+            return builder.WithTopic(message.ResponseTopic).WithCorrelationData(message.CorrelationData);
+        }
+
+        /// <summary>
+        ///     Set the filter topic according to the template, with
+        ///     remaining template parameters substituted by single-level
+        ///     wildcard.
+        /// </summary>
+        /// <param
+        ///     name="topicTemplate">
+        ///     a topic template
+        /// </param>
+        /// <param
+        ///     name="subscribeTreeRoot">
+        ///     whether to subscribe to the whole topic tree
+        /// </param>
+        /// <returns>the modified topic filter</returns>
+        public static MqttTopicFilterBuilder BuildFilter(this MqttTopicTemplate topicTemplate, bool subscribeTreeRoot = false)
+        {
+            return new MqttTopicFilterBuilder().WithTopicTemplate(topicTemplate, subscribeTreeRoot);
+        }
+
+        /// <summary>
+        ///     Return a message builder to respond to this message. The
+        ///     message's response topic and correlation data are included
+        ///     in the response message builder.
+        /// </summary>
+        /// <param
+        ///     name="message">
+        ///     a message with a response topic
+        /// </param>
+        /// <returns>a message builder</returns>
+        /// <see
+        ///     cref="AsResponseTo" />
+        public static MqttApplicationMessageBuilder BuildResponse(this MqttApplicationMessage message)
+        {
+            return new MqttApplicationMessageBuilder().AsResponseTo(message);
+        }
+
+        /// <summary>
+        ///     Return whether the message matches the given topic template.
+        /// </summary>
+        /// <param
+        ///     name="message">
+        ///     a message
+        /// </param>
+        /// <param
+        ///     name="topicTemplate">
+        ///     a topic template
+        /// </param>
+        /// <param
+        ///     name="subtree">
+        ///     whether to include the topic subtree
+        /// </param>
+        /// <returns></returns>
+        public static bool MatchesTopicTemplate(this MqttApplicationMessage message, MqttTopicTemplate topicTemplate, bool subtree = false)
+        {
+            return topicTemplate.MatchesTopic(message.Topic, subtree);
+        }
+
+        /// <summary>
+        ///     Set the filter topic according to the template, with
+        ///     template parameters substituted by a single-level
+        ///     wildcard.
+        /// </summary>
+        /// <param
+        ///     name="builder">
+        ///     a filter builder
+        /// </param>
+        /// <param
+        ///     name="topicTemplate">
+        ///     a topic template
+        /// </param>
+        /// <param
+        ///     name="subscribeTreeRoot">
+        ///     whether to subscribe to the whole topic tree
+        /// </param>
+        /// <returns>the modified topic filter</returns>
+        public static MqttTopicFilterBuilder WithTopicTemplate(this MqttTopicFilterBuilder builder, MqttTopicTemplate topicTemplate, bool subscribeTreeRoot = false)
+        {
+            return builder.WithTopic(subscribeTreeRoot ? topicTemplate.TopicTreeRootFilter : topicTemplate.TopicFilter);
+        }
+
+        /// <summary>
+        ///     Set the publication topic according to the topic template. The template
+        ///     must not have remaining (unset) parameters or contain wildcards.
+        /// </summary>
+        /// <param
+        ///     name="builder">
+        ///     a message builder
+        /// </param>
+        /// <param
+        ///     name="topicTemplate">
+        ///     a parameterless topic template
+        /// </param>
+        /// <returns>the modified message builder</returns>
+        /// <exception
+        ///     cref="ArgumentException">
+        ///     if the topic template has parameters
+        /// </exception>
+        public static MqttApplicationMessageBuilder WithTopicTemplate(this MqttApplicationMessageBuilder builder, MqttTopicTemplate topicTemplate)
+        {
+            if (topicTemplate.Parameters.Any())
+            {
+                throw new ArgumentException("topic templates must be parameter-less when sending " + topicTemplate.Template);
+            }
+
+            MqttTopicValidator.ThrowIfInvalid(topicTemplate.Template);
+            return builder.WithTopic(topicTemplate.Template);
+        }
+    }
+}

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

@@ -13,6 +13,7 @@ using MQTTnet.Diagnostics;
 using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
+using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -437,6 +438,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)
@@ -471,16 +575,154 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+        [TestMethod]
+        public async Task Subscribe_Does_Not_Hang_On_Server_Stop()
+        {
+            var timeout = TimeSpan.FromSeconds(2);
+            var testTimeout = TimeSpan.FromSeconds(timeout.TotalSeconds * 2);
+            const string topic = "test_topic_2";
+            using (var testEnvironment = CreateTestEnvironment())
+            using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout))
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                bool reject = true;
+                var receivedOnServer = new SemaphoreSlim(0, 1);
+                var failedOnClient = new SemaphoreSlim(0, 1);
+                testEnvironment.Server.InterceptingInboundPacketAsync += e =>
+                {
+                    if (e.Packet is MqttSubscribePacket)
+                    {
+                        if (reject)
+                        {
+                            e.ProcessPacket = false;
+                        }
+                        receivedOnServer.Release();
+                    }
+
+                    return CompletedTask.Instance;
+                };
+                managedClient.SynchronizingSubscriptionsFailedAsync += e =>
+                {
+                    failedOnClient.Release();
+                    return CompletedTask.Instance;
+                };
+
+                await managedClient.SubscribeAsync(topic);
+                Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout));
+                Assert.IsTrue(await failedOnClient.WaitAsync(testTimeout));
+
+                reject = false;
+                await managedClient.SubscribeAsync(topic);
+                Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout));
+            }
+        }
+
+        [TestMethod]
+        public async Task Unsubscribe_Does_Not_Hang_On_Server_Stop()
+        {
+            var timeout = TimeSpan.FromSeconds(2);
+            var testTimeout = TimeSpan.FromSeconds(timeout.TotalSeconds * 2);
+            const string topic = "test_topic_2";
+            using (var testEnvironment = CreateTestEnvironment())
+            using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout))
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                bool reject = true;
+                var receivedOnServer = new SemaphoreSlim(0, 1);
+                var failedOnClient = new SemaphoreSlim(0, 1);
+                testEnvironment.Server.InterceptingInboundPacketAsync += e =>
+                {
+                    if (e.Packet is MqttUnsubscribePacket)
+                    {
+                        if (reject)
+                        {
+                            e.ProcessPacket = false;
+                        }
+                        receivedOnServer.Release();
+                    }
+                    else if (e.Packet is MqttSubscribePacket)
+                    {
+                        receivedOnServer.Release();
+                    }
+
+                    return CompletedTask.Instance;
+                };
+                managedClient.SynchronizingSubscriptionsFailedAsync += e =>
+                {
+                    failedOnClient.Release();
+                    return CompletedTask.Instance;
+                };
+
+                await managedClient.SubscribeAsync(topic);
+                Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout));
+
+                await managedClient.UnsubscribeAsync(topic);
+                Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout));
+                Assert.IsTrue(await failedOnClient.WaitAsync(testTimeout));
+
+                reject = false;
+                await managedClient.UnsubscribeAsync(topic);
+                Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout));
+            }
+        }
+
+        [TestMethod]
+        public async Task Publish_Does_Not_Hang_On_Server_Error()
+        {
+            var timeout = TimeSpan.FromSeconds(2);
+            var testTimeout = TimeSpan.FromSeconds(timeout.TotalSeconds * 2);
+
+            const string topic = "test_topic_42";
+
+            using (var testEnvironment = CreateTestEnvironment())
+            using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout))
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                bool reject = true;
+                var receivedOnServer = new TaskCompletionSource<object>();
+                managedClient.ApplicationMessageProcessedAsync += e => Task.FromResult(reject &= e.Exception is null);
+                testEnvironment.Server.InterceptingInboundPacketAsync += e =>
+                {
+                    if (e.Packet is MqttPublishPacket)
+                    {
+                        if (reject)
+                        {
+                            e.ProcessPacket = false;
+                        }
+                        else
+                        {
+                            receivedOnServer.TrySetResult(null);
+                        }
+                    }
+
+                    return CompletedTask.Instance;
+                };
+
+                await managedClient.EnqueueAsync(new MqttApplicationMessage { Topic = topic, PayloadSegment = new ArraySegment<byte>(new byte[] { 1 }), Retain = true, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
+
+                var timeoutTask = Task.Delay(testTimeout);
+
+                var firstDone = await Task.WhenAny(receivedOnServer.Task, timeoutTask);
+                Assert.AreEqual(receivedOnServer.Task, firstDone, "Client is hung on publish!");
+            }
+        }
+
         async Task<MQTTnet.Extensions.ManagedClient.ManagedMqttClient> CreateManagedClientAsync(
             TestEnvironment testEnvironment,
             IMqttClient underlyingClient = null,
             TimeSpan? connectionCheckInterval = null,
-            string host = "localhost")
+            string host = "localhost",
+            TimeSpan? timeout = null)
         {
             await testEnvironment.StartServer();
 
             var clientOptions = new MqttClientOptionsBuilder().WithTcpServer(host, testEnvironment.ServerPort);
 
+            if (timeout != null)
+            {
+                clientOptions.WithTimeout(timeout.Value);
+            }
+
             var managedOptions = new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build();
 
             // Use a short connection check interval so that subscription operations are performed quickly

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.Tests/Extensions/MqttTopicTemplate_Tests.cs

@@ -0,0 +1,213 @@
+// 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.Linq;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Exceptions;
+using MQTTnet.Extensions.TopicTemplate;
+
+namespace MQTTnet.Tests.Extensions
+{
+    [TestClass]
+    public sealed class MqttTopicTemplate_Tests
+    {
+        [TestMethod]
+        public void BasicOps()
+        {
+            var template = new MqttTopicTemplate("A/B/{param}/D");
+            Assert.AreEqual("A/B/C/D", template.WithParameter("param", "C").Template);
+            Assert.AreEqual("A/B/+/D", template.WithoutParameter("param").Template);
+            Assert.AreEqual("A/B/+/D", template.TopicFilter);
+            Assert.AreEqual("A/B/+/D/#", template.TopicTreeRootFilter);
+
+            // corner cases
+            Assert.AreEqual("#", new MqttTopicTemplate("#").TopicFilter);
+            Assert.AreEqual("+", new MqttTopicTemplate("+").TopicFilter);
+            Assert.AreEqual("/", new MqttTopicTemplate("/").TopicFilter);
+        }
+
+        [TestMethod]
+        public void BasicOpsWithHash()
+        {
+            var template = new MqttTopicTemplate("A/B/{param}/D/#");
+            Assert.AreEqual("A/B/C/D/#", template.WithParameter("param", "C").Template);
+            Assert.AreEqual("A/B/+/D/#", template.WithoutParameter("param").Template);
+            Assert.AreEqual("A/B/+/D/#", template.TopicFilter);
+            Assert.AreEqual("A/B/+/D/#", template.TopicTreeRootFilter);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(ArgumentException))]
+        public void RejectsReservedChars1()
+        {
+            var template = new MqttTopicTemplate("A/B/{foo}/D");
+            template.WithParameter("foo", "a#");
+        }
+        
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException))]
+        public void RejectsEmptyTemplate()
+        {
+            var _ = new MqttTopicTemplate("");
+        }
+        
+        [TestMethod]
+        [ExpectedException(typeof(ArgumentNullException))]
+        public void RejectsNullTemplate()
+        {
+            var _ = new MqttTopicTemplate(null);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(ArgumentException))]
+        public void RejectsReservedChars2()
+        {
+            var template = new MqttTopicTemplate("A/B/{foo}/D");
+            template.WithParameter("foo", "e/f");
+        }
+        
+        [TestMethod]
+        public void IgnoresEmptyParameters()
+        {
+            var template = new MqttTopicTemplate("A/B/{}/D");
+            Assert.IsFalse(template.Parameters.Any());
+        }
+        
+        [TestMethod]
+        public void AcceptsValidTopics()
+        {
+            _ = new MqttTopicTemplate("A");
+            _ = new MqttTopicTemplate("/");
+            _ = new MqttTopicTemplate("/A");
+            _ = new MqttTopicTemplate("A/#");
+            _ = new MqttTopicTemplate("$SYS/#");
+            _ = new MqttTopicTemplate("////");
+            _ = new MqttTopicTemplate("A/B/{foo}/D/");
+        }
+
+        [TestMethod]
+        public void DynamicRoutingSupport()
+        {
+            var v1template = new MqttTopicTemplate("A/v1/{id}/F");
+            var v2template = new MqttTopicTemplate("A/v2/foo/{id}/{bar}");
+            var incoming_v1_topic = "A/v1/32/F";
+
+            var compat_v2_topic = v2template
+                .WithParameterValuesFrom(
+                    v1template.ParseParameterValues(incoming_v1_topic))
+                .WithParameter("bar", "unknown");
+
+            // id: 32 was relocated and bar: unknown added
+            Assert.AreEqual("A/v2/foo/32/unknown", compat_v2_topic.Template);
+        }
+
+        [TestMethod]
+        public void SubscriptionSupport()
+        {
+            var template = new MqttTopicTemplate("A/v1/{param}/F");
+            var filter = template.BuildFilter()
+                .WithAtLeastOnceQoS()
+                .WithNoLocal()
+                .Build();
+            Assert.AreEqual("A/v1/+/F", filter.Topic);
+        }
+
+        [TestMethod]
+        public void CanonicalPrefixFilter()
+        {
+            var template1 = new MqttTopicTemplate("A/v1/{param}/F");
+            var template2 = new MqttTopicTemplate("A/v1/E/F");
+            var template3 = new MqttTopicTemplate("A/v1/E/F/G/H");
+            var canonicalFilter = MqttTopicTemplate.FindCanonicalPrefix(new[] { template1, template2, template3 });
+            // possible improvement: Assert.AreEqual("A/v1/+/F", canonicalFilter.TopicFilter);
+            Assert.AreEqual("A/v1/+", canonicalFilter.TopicFilter);
+            Assert.AreEqual("A/v1/+/#", canonicalFilter.TopicTreeRootFilter);
+            
+            var template2b = new MqttTopicTemplate("A/v1/E/X");
+            canonicalFilter = MqttTopicTemplate.FindCanonicalPrefix(new[] { template1, template2, template3, template2b });
+            Assert.AreEqual("A/v1/+", canonicalFilter.Template);
+            Assert.AreEqual("A/v1/+", canonicalFilter.TopicFilter);
+            Assert.AreEqual("A/v1/+/#", canonicalFilter.TopicTreeRootFilter);
+
+
+            var template4 = new MqttTopicTemplate("A/v2/{prefix}");
+            var canonicalFilter2 = MqttTopicTemplate.FindCanonicalPrefix(new[] { template1, template2, template3, template4 });
+            Assert.AreEqual("A/+", canonicalFilter2.TopicFilter);
+            Assert.AreEqual("A/+/#", canonicalFilter2.TopicTreeRootFilter);
+        }
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple1()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A"),
+                new MqttTopicTemplate("B")
+            });
+            Assert.AreEqual("#", template.TopicFilter);
+            Assert.AreEqual("#", template.Template);
+        }
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple2()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A/v1/{param}/F/"),
+                new MqttTopicTemplate("A/v1/E/X")
+            });
+            Assert.AreEqual("A/v1/+", template.TopicFilter);
+        }
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple3()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A/v1/{param}/F"),
+                new MqttTopicTemplate("A/v1/{param}/F/X")
+            });
+            Assert.AreEqual("A/v1/+/+", template.TopicFilter);
+            Assert.AreEqual("A/v1/{param}/+", template.Template);
+        }
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple4()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A/v1/{param}/FFA"),
+                new MqttTopicTemplate("A/v1/{param}/FX")
+            });
+            Assert.AreEqual("A/v1/+/+", template.TopicFilter);
+            Assert.AreEqual("A/v1/{param}/+", template.Template);
+        }
+        
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple5()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A/v1/E/F/G/H"),
+                new MqttTopicTemplate("A/v2/{param}")
+            });
+            Assert.AreEqual("A/+", template.TopicFilter);
+            Assert.AreEqual("A/+/#", template.TopicTreeRootFilter);
+        }
+        
+        [TestMethod]
+        public void CanonicalPrefixFilterSimple6()
+        {
+            var template = MqttTopicTemplate.FindCanonicalPrefix(new[]
+            {
+                new MqttTopicTemplate("A/{version}/E/F/G/H"),
+                new MqttTopicTemplate("A/{version}/{param}")
+            });
+            Assert.AreEqual("A/+/+", template.TopicFilter);
+            Assert.AreEqual("A/+/+/#", template.TopicTreeRootFilter);
+        }
+    }
+}

Source/MQTTnet.Tests/Internal/CrossPlatformSocket_Tests.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Net.Sockets;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -17,7 +18,7 @@ namespace MQTTnet.Tests.Internal
         [TestMethod]
         public async Task Connect_Send_Receive()
         {
-            var crossPlatformSocket = new CrossPlatformSocket();
+            var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
             await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None);
 
             var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n");
@@ -36,7 +37,7 @@ namespace MQTTnet.Tests.Internal
         [ExpectedException(typeof(OperationCanceledException))]
         public async Task Try_Connect_Invalid_Host()
         {
-            var crossPlatformSocket = new CrossPlatformSocket();
+            var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
 
             var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5));
             cancellationToken.Token.Register(() => crossPlatformSocket.Dispose());
@@ -65,7 +66,7 @@ namespace MQTTnet.Tests.Internal
         [TestMethod]
         public void Set_Options()
         {
-            var crossPlatformSocket = new CrossPlatformSocket();
+            var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
 
             Assert.IsFalse(crossPlatformSocket.ReuseAddress);
             crossPlatformSocket.ReuseAddress = true;

Source/MQTTnet.Tests/MQTTnet.Tests.csproj

@@ -21,6 +21,7 @@
         <ProjectReference Include="..\..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
         <ProjectReference Include="..\..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj" />
         <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
+        <ProjectReference Include="..\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
         <ProjectReference Include="..\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
     </ItemGroup>
 

Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs

@@ -19,7 +19,7 @@ namespace MQTTnet.Tests
         public async Task Dispose_Channel_While_Used()
         {
             var ct = new CancellationTokenSource();
-            var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork);
+            var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
 
             try
             {
@@ -38,7 +38,7 @@ namespace MQTTnet.Tests
                     }
                 }, ct.Token);
 
-                var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork);
+                var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
                 await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
 
                 var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);

Source/MQTTnet.Tests/Server/Connection_Tests.cs

@@ -23,7 +23,7 @@ namespace MQTTnet.Tests.Server
             {
                 await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithDefaultCommunicationTimeout(TimeSpan.FromSeconds(1)));
 
-                var client = new CrossPlatformSocket(AddressFamily.InterNetwork);
+                var client = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
                 await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None);
 
                 // Don't send anything. The server should close the connection.
@@ -54,7 +54,7 @@ namespace MQTTnet.Tests.Server
 
                 // Send an invalid packet and ensure that the server will close the connection and stay in a waiting state
                 // forever. This is security related.
-                var client = new CrossPlatformSocket(AddressFamily.InterNetwork);
+                var client = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
                 await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None);
 
                 var buffer = Encoding.UTF8.GetBytes("Garbage");

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -141,7 +141,8 @@ namespace MQTTnet.Adapter
 
             try
             {
-                PacketInspector?.BeginReceivePacket();
+                var localPacketInspector = PacketInspector;
+                localPacketInspector?.BeginReceivePacket();
 
                 ReceivedMqttPacket receivedPacket;
                 var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -159,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);
 
@@ -215,7 +219,12 @@ 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);
 

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,8 +13,8 @@ namespace MQTTnet.Adapter
 {
     public sealed class MqttPacketInspector
     {
-        readonly MqttNetSourceLogger _logger;
         readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent;
+        readonly MqttNetSourceLogger _logger;
 
         MemoryStream _receivedPacketBuffer;
 
@@ -21,7 +22,11 @@ namespace MQTTnet.Adapter
         {
             _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));
         }
 
@@ -40,24 +45,11 @@ namespace MQTTnet.Adapter
             _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)
@@ -78,12 +83,12 @@ namespace MQTTnet.Adapter
             _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/Exceptions/MqttClientUnexpectedDisconnectReceivedException.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 MQTTnet.Exceptions;
 using MQTTnet.Packets;
@@ -11,8 +12,9 @@ namespace MQTTnet.Client
 {
     public sealed class MqttClientUnexpectedDisconnectReceivedException : MqttCommunicationException
     {
-        public MqttClientUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) 
+        public MqttClientUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket, Exception innerExcpetion = null) : base(
-            : base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.")
+            $"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.",
+            innerExcpetion)
         {
             ReasonCode = disconnectPacket.ReasonCode;
             SessionExpiryInterval = disconnectPacket.SessionExpiryInterval;
@@ -23,12 +25,12 @@ namespace MQTTnet.Client
 
         public MqttDisconnectReasonCode? ReasonCode { get; }
 
-        public uint? SessionExpiryInterval { get; }
-
         public string ReasonString { get; }
 
-        public List<MqttUserProperty> UserProperties { get; }
-
         public string ServerReference { get; }
+
+        public uint? SessionExpiryInterval { get; }
+
+        public List<MqttUserProperty> UserProperties { get; }
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -48,6 +48,7 @@ namespace MQTTnet.Client
         Task _packetReceiverTask;
         AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
         Task _publishPacketReceiverTask;
+        MqttDisconnectPacket _unexpectedDisconnectPacket;
 
         public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
         {
@@ -123,6 +124,8 @@ namespace MQTTnet.Client
                 var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger);
                 _adapter = adapter ?? throw new InvalidOperationException("The adapter factory did not provide an adapter.");
 
+                _unexpectedDisconnectPacket = null;
+
                 if (cancellationToken.CanBeCanceled)
                 {
                     connectResult = await ConnectInternal(adapter, cancellationToken).ConfigureAwait(false);
@@ -215,7 +218,7 @@ namespace MQTTnet.Client
                 }
 
                 // Sending the DISCONNECT may fail due to connection issues. The resulting exception
-                // must be throw to let the caller know that the disconnect is not a clean one.
+                // must be thrown to let the caller know that the disconnect was not clean.
                 var disconnectPacket = MqttPacketFactories.Disconnect.Create(options);
 
                 if (cancellationToken.CanBeCanceled)
@@ -406,29 +409,37 @@ namespace MQTTnet.Client
 
         Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken)
         {
-            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+            switch (eventArgs.PublishPacket.QualityOfServiceLevel)
             {
-                // no response required
+                case MqttQualityOfServiceLevel.AtMostOnce:
-            }
-            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-            {
-                if (!eventArgs.ProcessingFailed)
                 {
-                    var pubAckPacket = MqttPacketFactories.PubAck.Create(eventArgs);
+                    // no response required
-                    return Send(pubAckPacket, cancellationToken);
+                    break;
                 }
-            }
+                case MqttQualityOfServiceLevel.AtLeastOnce:
-            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-            {
-                if (!eventArgs.ProcessingFailed)
                 {
-                    var pubRecPacket = MqttPacketFactories.PubRec.Create(eventArgs);
+                    if (!eventArgs.ProcessingFailed)
-                    return Send(pubRecPacket, cancellationToken);
+                    {
+                        var pubAckPacket = MqttPacketFactories.PubAck.Create(eventArgs);
+                        return Send(pubAckPacket, cancellationToken);
+                    }
+
+                    break;
+                }
+                case MqttQualityOfServiceLevel.ExactlyOnce:
+                {
+                    if (!eventArgs.ProcessingFailed)
+                    {
+                        var pubRecPacket = MqttPacketFactories.PubRec.Create(eventArgs);
+                        return Send(pubRecPacket, cancellationToken);
+                    }
+
+                    break;
+                }
+                default:
+                {
+                    throw new MqttProtocolViolationException("Received a not supported QoS level.");
                 }
-            }
-            else
-            {
-                throw new MqttProtocolViolationException("Received a not supported QoS level.");
             }
 
             return CompletedTask.Instance;
@@ -445,21 +456,25 @@ namespace MQTTnet.Client
 
                 var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false);
 
-                if (receivedPacket is MqttConnAckPacket connAckPacket)
+                switch (receivedPacket)
                 {
-                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
+                    case MqttConnAckPacket connAckPacket:
-                }
+                    {
-                else if (receivedPacket is MqttAuthPacket)
+                        result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
-                {
+                        break;
-                    throw new NotSupportedException("Extended authentication handler is not yet supported");
+                    }
-                }
+                    case MqttAuthPacket _:
-                else if (receivedPacket == null)
+                    {
-                {
+                        throw new NotSupportedException("Extended authentication handler is not yet supported");
-                    throw new MqttCommunicationException("Connection closed.");
+                    }
-                }
+                    case null:
-                else
+                    {
-                {
+                        throw new MqttCommunicationException("Connection closed.");
-                    throw new InvalidOperationException($"Received an unexpected MQTT packet ({receivedPacket}).");
+                    }
+                    default:
+                    {
+                        throw new InvalidOperationException($"Received an unexpected MQTT packet ({receivedPacket}).");
+                    }
                 }
             }
             catch (Exception exception)
@@ -589,8 +604,8 @@ namespace MQTTnet.Client
                     _disconnectUserProperties,
                     exception);
 
-                // This handler must be executed in a new thread because otherwise a dead lock may happen
+                // This handler must be executed in a new thread to prevent a deadlock
-                // when trying to reconnect in that handler etc.
+                // that may occur when attempting to reconnect within that handler, etc.
                 Task.Run(() => _events.DisconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
             }
         }
@@ -599,12 +614,9 @@ namespace MQTTnet.Client
         {
             var clientWasConnected = IsConnected;
 
-            if (!DisconnectIsPendingOrFinished())
+            return !DisconnectIsPendingOrFinished() ?
-            {
+                DisconnectCore(sender, exception, connectResult, clientWasConnected)
-                return DisconnectCore(sender, exception, connectResult, clientWasConnected);
+                : CompletedTask.Instance;
-            }
-
-            return CompletedTask.Instance;
         }
 
         bool DisconnectIsPendingOrFinished()
@@ -657,24 +669,19 @@ namespace MQTTnet.Client
 
         Task OnConnected(MqttClientConnectResult connectResult)
         {
-            if (_events.ConnectedEvent.HasHandlers)
+            if (!_events.ConnectedEvent.HasHandlers)
             {
-                var eventArgs = new MqttClientConnectedEventArgs(connectResult);
+                return CompletedTask.Instance;
-                return _events.ConnectedEvent.InvokeAsync(eventArgs);
             }
 
-            return CompletedTask.Instance;
+            var eventArgs = new MqttClientConnectedEventArgs(connectResult);
+            return _events.ConnectedEvent.InvokeAsync(eventArgs);
         }
 
         Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
         {
             var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
-            if (extendedAuthenticationExchangeHandler != null)
+            return extendedAuthenticationExchangeHandler != null ? extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)) : CompletedTask.Instance;
-            {
-                return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this));
-            }
-
-            return CompletedTask.Instance;
         }
 
         Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
@@ -682,9 +689,10 @@ namespace MQTTnet.Client
             _disconnectReason = (int)disconnectPacket.ReasonCode;
             _disconnectReasonString = disconnectPacket.ReasonString;
             _disconnectUserProperties = disconnectPacket.UserProperties;
+            _unexpectedDisconnectPacket = disconnectPacket;
 
             // Also dispatch disconnect to waiting threads to generate a proper exception.
-            _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket));
+            _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket, null));
 
             return DisconnectInternal(_packetReceiverTask, null, null);
         }
@@ -724,15 +732,15 @@ namespace MQTTnet.Client
 
         Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
         {
-            if (!_packetDispatcher.TryDispatch(pubRecPacket))
+            if (_packetDispatcher.TryDispatch(pubRecPacket))
             {
-                // The packet is unknown. Probably due to a restart of the client.
+                return CompletedTask.Instance;
-                // So wen send this to the server to trigger a full resend of the message.
-                var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
-                return Send(pubRelPacket, cancellationToken);
             }
 
-            return CompletedTask.Instance;
+            // The packet is unknown, probably due to a restart of the client.
+            // So we send this to the server to trigger a full resend of the message.
+            var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
+            return Send(pubRelPacket, cancellationToken);
         }
 
         Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
@@ -751,10 +759,27 @@ namespace MQTTnet.Client
 
         async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
-            // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
+            try
-            await Send(publishPacket, cancellationToken).ConfigureAwait(false);
+            {
+                // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
+                await Send(publishPacket, cancellationToken).ConfigureAwait(false);
 
-            return MqttClientResultFactory.PublishResult.Create(null);
+                return MqttClientResultFactory.PublishResult.Create(null);
+            }
+            catch (Exception exception)
+            {
+                // We have to check if the server has sent a disconnect packet in response to the published message.
+                // Since we are in QoS 0 we do not get a direct response via an PUBACK packet and thus basically no
+                // feedback at all.
+                var localUnexpectedDisconnectPacket = _unexpectedDisconnectPacket;
+
+                if (localUnexpectedDisconnectPacket != null)
+                {
+                    throw new MqttClientUnexpectedDisconnectReceivedException(localUnexpectedDisconnectPacket, exception);
+                }
+
+                throw;
+            }
         }
 
         async Task<MqttClientPublishResult> PublishExactlyOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -930,7 +955,7 @@ namespace MQTTnet.Client
 
         static void ThrowNotConnected()
         {
-            throw new MqttCommunicationException("The client is not connected.");
+            throw new MqttClientNotConnectedException();
         }
 
         void TryInitiateDisconnect()
@@ -952,39 +977,36 @@ namespace MQTTnet.Client
         {
             try
             {
-                if (packet is MqttPublishPacket publishPacket)
+                switch (packet)
                 {
-                    EnqueueReceivedPublishPacket(publishPacket);
+                    case MqttPublishPacket publishPacket:
-                }
+                        EnqueueReceivedPublishPacket(publishPacket);
-                else if (packet is MqttPubRecPacket pubRecPacket)
+                        break;
-                {
+                    case MqttPubRecPacket pubRecPacket:
-                    await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false);
+                        await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false);
-                }
+                        break;
-                else if (packet is MqttPubRelPacket pubRelPacket)
+                    case MqttPubRelPacket pubRelPacket:
-                {
+                        await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
-                    await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
+                        break;
-                }
+                    case MqttDisconnectPacket disconnectPacket:
-                else if (packet is MqttDisconnectPacket disconnectPacket)
+                        await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
-                {
+                        break;
-                    await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
+                    case MqttAuthPacket authPacket:
-                }
+                        await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
-                else if (packet is MqttAuthPacket authPacket)
+                        break;
-                {
+                    case MqttPingRespPacket _:
-                    await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
+                        _packetDispatcher.TryDispatch(packet);
-                }
+                        break;
-                else if (packet is MqttPingRespPacket)
+                    case MqttPingReqPacket _:
-                {
+                        throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a client to the server only.");
-                    _packetDispatcher.TryDispatch(packet);
+                    default:
-                }
-                else if (packet is MqttPingReqPacket)
-                {
-                    throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a client to the server only.");
-                }
-                else
-                {
-                    if (!_packetDispatcher.TryDispatch(packet))
                     {
-                        throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
+                        if (!_packetDispatcher.TryDispatch(packet))
+                        {
+                            throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
+                        }
+
+                        break;
                     }
                 }
             }
@@ -995,19 +1017,19 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                if (exception is OperationCanceledException)
+                switch (exception)
                 {
-                }
+                    case OperationCanceledException _:
-                else if (exception is MqttCommunicationException)
+                        break;
-                {
+                    case MqttCommunicationException _:
-                    _logger.Warning(exception, "Communication error while receiving packets");
+                        _logger.Warning(exception, "Communication error while receiving packets");
-                }
+                        break;
-                else
+                    default:
-                {
+                        _logger.Error(exception, "Error while processing received {0} packet", packet.GetType().Name);
-                    _logger.Error(exception, "Error while processing received {0} packet", packet.GetType().Name);
+                        break;
                 }
 
-                // The packet dispatcher may already be gone due to disconnect etc!
+                // The packet dispatcher may already be gone due to disconnect etc.!
                 _packetDispatcher?.FailAll(exception);
 
                 await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
@@ -1050,17 +1072,16 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                if (exception is OperationCanceledException)
+                switch (exception)
                 {
-                    return;
+                    case OperationCanceledException _:
-                }
+                        return;
-                else if (exception is MqttCommunicationException)
+                    case MqttCommunicationException _:
-                {
+                        _logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
-                    _logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
+                        break;
-                }
+                    default:
-                else
+                        _logger.Error(exception, "Error exception while sending/receiving keep alive packets");
-                {
+                        break;
-                    _logger.Error(exception, "Error exception while sending/receiving keep alive packets");
                 }
 
                 await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);

Source/MQTTnet/Client/Options/MqttClientCertificateSelectionEventArgs.cs

@@ -0,0 +1,36 @@
+// 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.Security.Cryptography.X509Certificates;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientCertificateSelectionEventArgs : EventArgs
+    {
+        public MqttClientCertificateSelectionEventArgs(
+            string targetHost,
+            X509CertificateCollection localCertificates,
+            X509Certificate remoteCertificate,
+            string[] acceptableIssuers,
+            MqttClientTcpOptions tcpOptions)
+        {
+            TargetHost = targetHost;
+            LocalCertificates = localCertificates;
+            RemoveCertificate = remoteCertificate;
+            AcceptableIssuers = acceptableIssuers;
+            TcpOptions = tcpOptions ?? throw new ArgumentNullException(nameof(tcpOptions));
+        }
+
+        public string[] AcceptableIssuers { get; }
+
+        public X509CertificateCollection LocalCertificates { get; }
+
+        public X509Certificate RemoveCertificate { get; }
+
+        public string TargetHost { get; }
+
+        public MqttClientTcpOptions TcpOptions { get; }
+    }
+}

Source/MQTTnet/Client/Options/MqttClientCertificateValidationEventArgs.cs

@@ -10,11 +10,20 @@ namespace MQTTnet.Client
 {
     public sealed class MqttClientCertificateValidationEventArgs : EventArgs
     {
-        public X509Certificate Certificate { get; set; }
+        public MqttClientCertificateValidationEventArgs(X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors, IMqttClientChannelOptions clientOptions)
+        {
+            Certificate = certificate;
+            Chain = chain;
+            SslPolicyErrors = sslPolicyErrors;
+            ClientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
+        }
 
-        public X509Chain Chain { get; set; }
+        public X509Certificate Certificate { get; }
+
+        public X509Chain Chain { get; }
+
+        public IMqttClientChannelOptions ClientOptions { get; }
 
-        public IMqttClientChannelOptions ClientOptions { get; set; }
 #if NET452 || NET461 || NET48
         /// <summary>
         ///     Can be a host string name or an object derived from WebRequest.
@@ -22,6 +31,6 @@ namespace MQTTnet.Client
         public object Sender { get; set; }
 #endif
 
-        public SslPolicyErrors SslPolicyErrors { get; set; }
+        public SslPolicyErrors SslPolicyErrors { get; }
     }
 }

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -7,6 +7,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Net;
+using System.Net.Sockets;
 using System.Text;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
@@ -17,8 +19,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 +67,46 @@ 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, dns.AddressFamily);
+                        }
+                        else
+                        {
+                            _remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure, dns.AddressFamily);
+                        }
+                    }
+                }
+
+                if (_remoteEndPoint is IPEndPoint ip)
+                {
+                    if (ip.Port == 0)
+                    {
+                        if (_port.HasValue)
+                        {
+                            _remoteEndPoint = new IPEndPoint(ip.Address, _port.Value);
+                        }
+                        else
+                        {
+                            _remoteEndPoint = new IPEndPoint(ip.Address, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
+                        }
+                    }
+                }
+
+                if (_tcpOptions.RemoteEndpoint == null)
+                {
+                    _tcpOptions.RemoteEndpoint = _remoteEndPoint;
+                }
             }
             else if (_webSocketOptions != null)
             {
@@ -86,6 +130,12 @@ namespace MQTTnet.Client
             return _options;
         }
 
+        public MqttClientOptionsBuilder WithAddressFamily(AddressFamily addressFamily)
+        {
+            _tcpOptions.AddressFamily = addressFamily;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data)
         {
             _options.AuthenticationMethod = method;
@@ -117,7 +167,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
         {
             if (uri == null)
@@ -161,7 +210,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 +238,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;
@@ -226,7 +282,7 @@ namespace MQTTnet.Client
         }
 
         /// <summary>
-        ///     The client will not throw an exception when the MQTT server responses with a non success ACK packet.
+        ///     The client will not throw an exception when the MQTT server responds with a non success ACK packet.
         ///     This will become the default behavior in future versions of the library.
         /// </summary>
         public MqttClientOptionsBuilder WithoutThrowOnNonSuccessfulConnectResponse()
@@ -235,6 +291,12 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithProtocolType(ProtocolType protocolType)
+        {
+            _tcpOptions.ProtocolType = protocolType;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
         {
             if (value == MqttProtocolVersion.Unknown)
@@ -305,13 +367,19 @@ namespace MQTTnet.Client
             return this;
         }
 
-        public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
+        public MqttClientOptionsBuilder WithTcpServer(string host, int? port = null, AddressFamily addressFamily = AddressFamily.Unspecified)
         {
-            _tcpOptions = new MqttClientTcpOptions
+            if (host == null)
             {
-                Server = server,
+                throw new ArgumentNullException(nameof(host));
-                Port = port
+            }
-            };
+
+            _tcpOptions = new MqttClientTcpOptions();
+
+            // The value 0 will be updated when building the options.
+            // This a backward compatibility feature.
+            _remoteEndPoint = new DnsEndPoint(host, port ?? 0, addressFamily);
+            _port = port;
 
             return this;
         }

Source/MQTTnet/Client/Options/MqttClientTcpOptions.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.Net;
 using System.Net.Sockets;
 
@@ -11,34 +12,50 @@ 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);
+        [Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
-
-        public bool NoDelay { get; set; } = true;
-
         public int? Port { get; set; }
 
+        [Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
         public string Server { 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; }
+
+        /// <summary>
+        ///     Enables or disables the Nagle algorithm for the socket.
+        ///     This is only supported for TCP.
+        ///     For other protocol types the value is ignored.
+        ///     Default: true
+        /// </summary>
+        public bool NoDelay { get; set; } = true;
+
+        /// <summary>
+        ///     The MQTT transport is usually TCP but when using other endpoint types like
+        ///     unix sockets it must be changed (IP for unix sockets).
+        /// </summary>
+        public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp;
+
+        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/Client/Options/MqttClientTlsOptions.cs

@@ -14,6 +14,8 @@ namespace MQTTnet.Client
     {
         public Func<MqttClientCertificateValidationEventArgs, bool> CertificateValidationHandler { get; set; }
 
+        public Func<MqttClientCertificateSelectionEventArgs, X509Certificate> CertificateSelectionHandler { get; set; }
+
         public bool UseTls { get; set; }
 
         public bool IgnoreCertificateRevocationErrors { get; set; }

Source/MQTTnet/Client/Options/MqttClientTlsOptionsBuilder.cs

@@ -49,6 +49,17 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientTlsOptionsBuilder WithCertificateSelectionHandler(Func<MqttClientCertificateSelectionEventArgs, X509Certificate> certificateSelectionHandler)
+        {
+            if (certificateSelectionHandler == null)
+            {
+                throw new ArgumentNullException(nameof(certificateSelectionHandler));
+            }
+
+            _tlsOptions.CertificateSelectionHandler = certificateSelectionHandler;
+            return this;
+        }
+
         public MqttClientTlsOptionsBuilder WithClientCertificates(IEnumerable<X509Certificate2> certificates)
         {
             if (certificates == null)

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

@@ -352,7 +352,16 @@ namespace MQTTnet.Formatter.V5
 
         MqttPacket DecodeDisconnectPacket(ArraySegment<byte> body)
         {
-            ThrowIfBodyIsEmpty(body);
+            // From RFC: 3.14.2.1 Disconnect Reason Code
+            // Byte 1 in the Variable Header is the Disconnect Reason Code.
+            // If the Remaining Length is less than 1 the value of 0x00 (Normal disconnection) is used.
+            if (body.Count == 0)
+            {
+                return new MqttDisconnectPacket
+                {
+                    ReasonCode = MqttDisconnectReasonCode.NormalDisconnection
+                };
+            }
 
             _bufferReader.SetBuffer(body.Array, body.Offset, body.Count);
 

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);

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -22,20 +22,20 @@ namespace MQTTnet.Implementations
 
         NetworkStream _networkStream;
 
-        public CrossPlatformSocket(AddressFamily addressFamily)
+        public CrossPlatformSocket(AddressFamily addressFamily, ProtocolType protocolType)
         {
-            _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
+            _socket = new Socket(addressFamily, SocketType.Stream, protocolType);
 
 #if !NET5_0_OR_GREATER
             _socketDisposeAction = _socket.Dispose;
 #endif
         }
 
-        public CrossPlatformSocket()
+        public CrossPlatformSocket(ProtocolType protocolType)
         {
             // Having this constructor is important because avoiding the address family as parameter
             // will make use of dual mode in the .net framework.
-            _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+            _socket = new Socket(SocketType.Stream, protocolType);
 
 #if !NET5_0_OR_GREATER
             _socketDisposeAction = _socket.Dispose;
@@ -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,34 @@ 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);
+                    // This is a fix for Mono which behaves differently than dotnet.
+                    // The connection will not be established when the DNS endpoint is used.
+                    if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
+                    {
+                        await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, dns.Host, dns.Port, null).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
+                    }
 #else
-                    await _socket.ConnectAsync(host, port).ConfigureAwait(false);
+
+                    // This is a fix for Mono which behaves differently than dotnet.
+                    // The connection will not be established when the DNS endpoint is used.
+                    if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
+                    {
+                        await _socket.ConnectAsync(dns.Host, dns.Port).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
+                    }
 #endif
                 }
 #endif
@@ -217,7 +241,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;
@@ -15,6 +16,7 @@ using MQTTnet.Channel;
 using MQTTnet.Client;
 using MQTTnet.Exceptions;
 using MQTTnet.Internal;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Implementations
 {
@@ -62,11 +64,11 @@ namespace MQTTnet.Implementations
             {
                 if (_tcpOptions.AddressFamily == AddressFamily.Unspecified)
                 {
-                    socket = new CrossPlatformSocket();
+                    socket = new CrossPlatformSocket(_tcpOptions.ProtocolType);
                 }
                 else
                 {
-                    socket = new CrossPlatformSocket(_tcpOptions.AddressFamily);
+                    socket = new CrossPlatformSocket(_tcpOptions.AddressFamily, _tcpOptions.ProtocolType);
                 }
 
                 if (_tcpOptions.LocalEndpoint != null)
@@ -77,7 +79,12 @@ namespace MQTTnet.Implementations
                 socket.ReceiveBufferSize = _tcpOptions.BufferSize;
                 socket.SendBufferSize = _tcpOptions.BufferSize;
                 socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds;
-                socket.NoDelay = _tcpOptions.NoDelay;
+
+                if (_tcpOptions.ProtocolType == ProtocolType.Tcp)
+                {
+                    // Other protocol types do not support the Nagle algorithm.
+                    socket.NoDelay = _tcpOptions.NoDelay;
+                }
 
                 if (socket.LingerState != null)
                 {
@@ -92,7 +99,30 @@ namespace MQTTnet.Implementations
                     socket.DualMode = _tcpOptions.DualMode.Value;
                 }
 
-                await socket.ConnectAsync(_tcpOptions.Server, _tcpOptions.GetPort(), cancellationToken).ConfigureAwait(false);
+                // This block is only for backward compatibility.
+                if (_tcpOptions.RemoteEndpoint == null && !string.IsNullOrEmpty(_tcpOptions.Server))
+                {
+                    int port;
+                    if (_tcpOptions.Port.HasValue)
+                    {
+                        port = _tcpOptions.Port.Value;
+                    }
+                    else
+                    {
+                        if (_tcpOptions.TlsOptions?.UseTls == true)
+                        {
+                            port = MqttPorts.Secure;
+                        }
+                        else
+                        {
+                            port = MqttPorts.Default;
+                        }
+                    }
+
+                    _tcpOptions.RemoteEndpoint = new DnsEndPoint(_tcpOptions.Server, port, AddressFamily.Unspecified);
+                }
+
+                await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
 
                 cancellationToken.ThrowIfCancellationRequested();
 
@@ -103,10 +133,31 @@ namespace MQTTnet.Implementations
                     var targetHost = _tcpOptions.TlsOptions.TargetHost;
                     if (string.IsNullOrEmpty(targetHost))
                     {
-                        targetHost = _tcpOptions.Server;
+                        if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
+                        {
+                            targetHost = dns.Host;
+                        }
+                    }
+
+                    SslStream sslStream;
+                    if (_tcpOptions.TlsOptions.CertificateSelectionHandler != null)
+                    {
+                        sslStream = new SslStream(
+                            networkStream,
+                            false,
+                            InternalUserCertificateValidationCallback,
+                            InternalUserCertificateSelectionCallback);
+                    }
+                    else
+                    {
+                        // Use a different constructor depending on the options for MQTTnet so that we do not have
+                        // to copy the exact same behavior of the selection handler.
+                        sslStream = new SslStream(
+                            networkStream,
+                            false,
+                            InternalUserCertificateValidationCallback);
                     }
 
-                    var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
 #if NETCOREAPP3_1_OR_GREATER
@@ -131,9 +182,7 @@ namespace MQTTnet.Implementations
                             {
                                 TrustMode = X509ChainTrustMode.CustomRootTrust,
                                 VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
-                                RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
+                                RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode
-                                    ? X509RevocationMode.NoCheck
-                                    : _tcpOptions.TlsOptions.RevocationMode
                             };
 
                             sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
@@ -289,19 +338,34 @@ namespace MQTTnet.Implementations
             }
         }
 
+        X509Certificate InternalUserCertificateSelectionCallback(
+            object sender,
+            string targetHost,
+            X509CertificateCollection localCertificates,
+            X509Certificate remoteCertificate,
+            string[] acceptableIssuers)
+        {
+            var certificateSelectionHandler = _tcpOptions?.TlsOptions?.CertificateSelectionHandler;
+            if (certificateSelectionHandler != null)
+            {
+                var eventArgs = new MqttClientCertificateSelectionEventArgs(targetHost, localCertificates, remoteCertificate, acceptableIssuers, _tcpOptions);
+                return certificateSelectionHandler(eventArgs);
+            }
+
+            if (localCertificates?.Count > 0)
+            {
+                return localCertificates[0];
+            }
+
+            return null;
+        }
+
         bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
         {
             var certificateValidationHandler = _tcpOptions?.TlsOptions?.CertificateValidationHandler;
             if (certificateValidationHandler != null)
             {
-                var eventArgs = new MqttClientCertificateValidationEventArgs
+                var eventArgs = new MqttClientCertificateValidationEventArgs(x509Certificate, chain, sslPolicyErrors, _tcpOptions);
-                {
-                    Certificate = x509Certificate,
-                    Chain = chain,
-                    SslPolicyErrors = sslPolicyErrors,
-                    ClientOptions = _tcpOptions
-                };
-
                 return certificateValidationHandler(eventArgs);
             }
 

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -65,7 +65,7 @@ namespace MQTTnet.Implementations
 
                 _logger.Info("Starting TCP listener (Endpoint={0}, TLS={1})", _localEndPoint, _tlsOptions?.CertificateProvider != null);
 
-                _socket = new CrossPlatformSocket(_addressFamily);
+                _socket = new CrossPlatformSocket(_addressFamily, ProtocolType.Tcp);
 
                 // Usage of socket options is described here: https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socket.setsocketoption?view=netcore-2.2
                 if (_options.ReuseAddress)

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -247,13 +247,9 @@ namespace MQTTnet.Implementations
 #elif NET452 || NET461 || NET48
                 ServicePointManager.ServerCertificateValidationCallback += (sender, certificate, chain, sslPolicyErrors) =>
                 {
-                    var context = new MqttClientCertificateValidationEventArgs
+                    var context = new MqttClientCertificateValidationEventArgs(certificate, chain, sslPolicyErrors, _options)
                     {
-                        Sender = sender,
+                        Sender = sender
-                        Certificate = certificate,
-                        Chain = chain,
-                        SslPolicyErrors = sslPolicyErrors,
-                        ClientOptions = _options
                     };
 
                     return certificateValidationHandler(context);
@@ -262,17 +258,16 @@ namespace MQTTnet.Implementations
                 clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
                 {
                     // TODO: Find a way to add client options to same callback. Problem is that they have a different type.
-                    var context = new MqttClientCertificateValidationEventArgs
+                    var context = new MqttClientCertificateValidationEventArgs(certificate, chain, sslPolicyErrors, _options);
-                    {
-                        Certificate = certificate,
-                        Chain = chain,
-                        SslPolicyErrors = sslPolicyErrors,
-                        ClientOptions = _options
-                    };
-
                     return certificateValidationHandler(context);
                 };
 #endif
+
+                var certificateSelectionHandler = _options.TlsOptions?.CertificateSelectionHandler;
+                if (certificateSelectionHandler != null)
+                {
+                    throw new NotSupportedException("Remote certificate selection callback is not supported for WebSocket connections.");
+                }
             }
         }
     }

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

@@ -257,7 +257,12 @@ namespace MQTTnet.Server
 
                     if (matchingSubscribersCount == 0)
                     {
-                        reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
+                        if (reasonCode == (int)MqttPubAckReasonCode.Success)
+                        {
+                            // Only change the value if it was success. Otherwise, we would hide an error or not authorized status.
+                            reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
+                        }
+
                         await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
                     }
                 }

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)