Update version

* Access exception so that it will not treated as unhandled

* Refactor code

* Improve unit tests

* Fix null reference exception

* Update ReleaseNotes.md

.github/workflows/ReleaseNotes.md

@@ -1,4 +1,3 @@
-* [Client] Fixed _PlatformNotSupportedException_ when using Blazor (#1755, thanks to @Nickztar).
-* [Server] Fixed _NullReferenceException_ in retained messages management (#1762, thanks to @logicaloud).
-* [Server] Exposed new option which allows disabling packet fragmentation (#1753).
-* [Server] Expired sessions will no longer be used when a client connects (#1756).
+* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
+* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
+* [Client] Fixed handling of unobserved tasks exceptions (#1871).

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.2.1.${{github.run_number}}"
+  VERSION: "4.3.2.${{github.run_number}}"
 
 jobs:
   build:
@@ -54,7 +54,7 @@ jobs:
         uses: darenm/Setup-VSTest@v1
 
       - name: Core Tests
-        run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net6.0\MQTTnet.Tests.dll
+        run: vstest.console.exe Source\MQTTnet.Tests\bin\Release\net7.0\MQTTnet.Tests.dll
 
       - name: ASP.NET Tests
         run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll

MQTTnet.sln.DotSettings

@@ -240,6 +240,7 @@ See the LICENSE file in the project root for more information.</s:String>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=catched/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=CONNACK/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>

Samples/Client/Client_Connection_Samples.cs

@@ -7,6 +7,7 @@
 // ReSharper disable InconsistentNaming
 
 using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
 using MQTTnet.Client;
 using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
@@ -70,32 +71,6 @@ public static class Client_Connection_Samples
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
     }
-    
-    public static async Task Connect_With_Amazon_AWS()
-    {
-        /*
-         * This sample creates a simple MQTT client and connects to an Amazon Web Services broker.
-         *
-         * The broker requires special settings which are set here.
-         */
-
-        var mqttFactory = new MqttFactory();
-
-        using (var mqttClient = mqttFactory.CreateMqttClient())
-        {
-            var mqttClientOptions = new MqttClientOptionsBuilder()
-                .WithTcpServer("amazon.web.services.broker")
-                // Disabling packet fragmentation is very important!  
-                .WithoutPacketFragmentation()
-                .Build();
-            
-            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
-
-            Console.WriteLine("The MQTT client is connected.");
-
-            await mqttClient.DisconnectAsync();
-        }
-    }
 
     public static async Task Connect_Client_Timeout()
     {
@@ -161,15 +136,15 @@ public static class Client_Connection_Samples
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
             var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("mqtt.fluux.io")
-                .WithTls(
+                .WithTlsOptions(
                     o =>
                     {
                         // The used public broker sometimes has invalid certificates. This sample accepts all
                         // certificates. This should not be used in live environments.
-                        o.CertificateValidationHandler = _ => true;
+                        o.WithCertificateValidationHandler(_ => true);
 
                         // The default value is determined by the OS. Set manually to force version.
-                        o.SslProtocol = SslProtocols.Tls12;
+                        o.WithSslProtocols(SslProtocols.Tls12);
                     })
                 .Build();
 
@@ -196,7 +171,7 @@ public static class Client_Connection_Samples
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
-            var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").Build();
+            var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).Build();
 
             var response = await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
@@ -218,7 +193,7 @@ public static class Client_Connection_Samples
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
-            var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").Build();
+            var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).Build();
 
             var response = await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
@@ -241,13 +216,11 @@ public static class Client_Connection_Samples
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
             var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
-                .WithTls(
-                    o =>
-                    {
+                .WithTlsOptions(
+                    o => o.WithCertificateValidationHandler(
                         // The used public broker sometimes has invalid certificates. This sample accepts all
                         // certificates. This should not be used in live environments.
-                        o.CertificateValidationHandler = _ => true;
-                    })
+                        _ => true))
                 .Build();
 
             // In MQTTv5 the response contains much more information.
@@ -262,6 +235,31 @@ public static class Client_Connection_Samples
         }
     }
 
+    public static async Task Connect_With_Amazon_AWS()
+    {
+        /*
+         * This sample creates a simple MQTT client and connects to an Amazon Web Services broker.
+         *
+         * The broker requires special settings which are set here.
+         */
+
+        var mqttFactory = new MqttFactory();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("amazon.web.services.broker")
+                // Disabling packet fragmentation is very important!  
+                .WithoutPacketFragmentation()
+                .Build();
+
+            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            Console.WriteLine("The MQTT client is connected.");
+
+            await mqttClient.DisconnectAsync();
+        }
+    }
+
     public static async Task Disconnect_Clean()
     {
         /*
@@ -317,18 +315,19 @@ public static class Client_Connection_Samples
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
             var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("mqtt.fluux.io", 8883)
-                .WithTls(
+                .WithTlsOptions(
                     o =>
                     {
-                        o.CertificateValidationHandler = eventArgs =>
-                        {
-                            eventArgs.Certificate.Subject.DumpToConsole();
-                            eventArgs.Certificate.GetExpirationDateString().DumpToConsole();
-                            eventArgs.Chain.ChainPolicy.RevocationMode.DumpToConsole();
-                            eventArgs.Chain.ChainStatus.DumpToConsole();
-                            eventArgs.SslPolicyErrors.DumpToConsole();
-                            return true;
-                        };
+                        o.WithCertificateValidationHandler(
+                            eventArgs =>
+                            {
+                                eventArgs.Certificate.Subject.DumpToConsole();
+                                eventArgs.Certificate.GetExpirationDateString().DumpToConsole();
+                                eventArgs.Chain.ChainPolicy.RevocationMode.DumpToConsole();
+                                eventArgs.Chain.ChainStatus.DumpToConsole();
+                                eventArgs.SslPolicyErrors.DumpToConsole();
+                                return true;
+                            });
                     })
                 .Build();
 
@@ -432,6 +431,59 @@ public static class Client_Connection_Samples
                         }
                     }
                 });
+
+            Console.WriteLine("Press <Enter> to exit");
+            Console.ReadLine();
         }
     }
-}
+
+    public static async Task ConnectTls_WithCaFile()
+    {
+        var mqttFactory = new MqttFactory();
+
+        X509Certificate2Collection caChain = new X509Certificate2Collection();
+        caChain.ImportFromPem(mosquitto_org); // from https://test.mosquitto.org/ssl/mosquitto.org.crt
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("test.mosquitto.org", 8883)
+                .WithTlsOptions(new MqttClientTlsOptionsBuilder()
+                    .WithTrustChain(caChain) 
+                    .Build())
+                .Build();
+
+            var connAck = await mqttClient.ConnectAsync(mqttClientOptions);
+            Console.WriteLine("Connected to test.moquitto.org:8883 with CaFile mosquitto.org.crt: " + connAck.ResultCode);
+        }
+
+
+    }
+    const string mosquitto_org = @"
+-----BEGIN CERTIFICATE-----
+MIIEAzCCAuugAwIBAgIUBY1hlCGvdj4NhBXkZ/uLUZNILAwwDQYJKoZIhvcNAQEL
+BQAwgZAxCzAJBgNVBAYTAkdCMRcwFQYDVQQIDA5Vbml0ZWQgS2luZ2RvbTEOMAwG
+A1UEBwwFRGVyYnkxEjAQBgNVBAoMCU1vc3F1aXR0bzELMAkGA1UECwwCQ0ExFjAU
+BgNVBAMMDW1vc3F1aXR0by5vcmcxHzAdBgkqhkiG9w0BCQEWEHJvZ2VyQGF0Y2hv
+by5vcmcwHhcNMjAwNjA5MTEwNjM5WhcNMzAwNjA3MTEwNjM5WjCBkDELMAkGA1UE
+BhMCR0IxFzAVBgNVBAgMDlVuaXRlZCBLaW5nZG9tMQ4wDAYDVQQHDAVEZXJieTES
+MBAGA1UECgwJTW9zcXVpdHRvMQswCQYDVQQLDAJDQTEWMBQGA1UEAwwNbW9zcXVp
+dHRvLm9yZzEfMB0GCSqGSIb3DQEJARYQcm9nZXJAYXRjaG9vLm9yZzCCASIwDQYJ
+KoZIhvcNAQEBBQADggEPADCCAQoCggEBAME0HKmIzfTOwkKLT3THHe+ObdizamPg
+UZmD64Tf3zJdNeYGYn4CEXbyP6fy3tWc8S2boW6dzrH8SdFf9uo320GJA9B7U1FW
+Te3xda/Lm3JFfaHjkWw7jBwcauQZjpGINHapHRlpiCZsquAthOgxW9SgDgYlGzEA
+s06pkEFiMw+qDfLo/sxFKB6vQlFekMeCymjLCbNwPJyqyhFmPWwio/PDMruBTzPH
+3cioBnrJWKXc3OjXdLGFJOfj7pP0j/dr2LH72eSvv3PQQFl90CZPFhrCUcRHSSxo
+E6yjGOdnz7f6PveLIB574kQORwt8ePn0yidrTC1ictikED3nHYhMUOUCAwEAAaNT
+MFEwHQYDVR0OBBYEFPVV6xBUFPiGKDyo5V3+Hbh4N9YSMB8GA1UdIwQYMBaAFPVV
+6xBUFPiGKDyo5V3+Hbh4N9YSMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEL
+BQADggEBAGa9kS21N70ThM6/Hj9D7mbVxKLBjVWe2TPsGfbl3rEDfZ+OKRZ2j6AC
+6r7jb4TZO3dzF2p6dgbrlU71Y/4K0TdzIjRj3cQ3KSm41JvUQ0hZ/c04iGDg/xWf
++pp58nfPAYwuerruPNWmlStWAXf0UTqRtg4hQDWBuUFDJTuWuuBvEXudz74eh/wK
+sMwfu1HFvjy5Z0iMDU8PUDepjVolOCue9ashlS4EB5IECdSR2TItnAIiIwimx839
+LdUdRudafMu5T5Xma182OC0/u/xRlEm+tvKGGmfFcN0piqVl8OrSPBgIlb+1IKJE
+m/XriWr/Cq4h/JfB7NTsezVslgkBaoU=
+-----END CERTIFICATE-----
+";
+
+}

Samples/Client/Client_Subscribe_Samples.cs

@@ -5,8 +5,8 @@
 // ReSharper disable UnusedType.Global
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
+// ReSharper disable UnusedMember.Local
 
-using MQTTnet;
 using MQTTnet.Client;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
@@ -177,4 +177,62 @@ public static class Client_Subscribe_Samples
             response.DumpToConsole();
         }
     }
-}
+
+    static void ConcurrentProcessingDisableAutoAcknowledge(CancellationToken shutdownToken, IMqttClient mqttClient)
+    {
+        /*
+         * This sample shows how to achieve concurrent processing and not have message AutoAcknowledged
+         * This to have a proper QoS1 (at-least-once) experience for what at least MQTT specification can provide
+         */
+        mqttClient.ApplicationMessageReceivedAsync += ea =>
+        {
+            ea.AutoAcknowledge = false;
+
+            async Task ProcessAsync()
+            {
+                // DO YOUR WORK HERE!
+                await Task.Delay(1000, shutdownToken);
+                await ea.AcknowledgeAsync(shutdownToken);
+                // WARNING: If process failures are not transient the message will be retried on every restart of the client
+                //          A failed message will not be dispatched again to the client as MQTT does not have a NACK packet to let
+                //          the broker know processing failed
+                //
+                // Optionally: Use a framework like Polly to create a retry policy: https://github.com/App-vNext/Polly#retry
+            }
+
+            _ = Task.Run(ProcessAsync, shutdownToken);
+
+            return Task.CompletedTask;
+        };
+    }
+
+    static void ConcurrentProcessingWithLimit(CancellationToken shutdownToken, IMqttClient mqttClient)
+    {
+        /*
+         * This sample shows how to achieve concurrent processing, with:
+         * - a maximum concurrency limit based on Environment.ProcessorCount
+         */
+
+        var concurrent = new SemaphoreSlim(Environment.ProcessorCount);
+
+        mqttClient.ApplicationMessageReceivedAsync += async ea =>
+        {
+            await concurrent.WaitAsync(shutdownToken).ConfigureAwait(false);
+
+            async Task ProcessAsync()
+            {
+                try
+                {
+                    // DO YOUR WORK HERE!
+                    await Task.Delay(1000, shutdownToken);
+                }
+                finally
+                {
+                    concurrent.Release();
+                }
+            }
+
+            _ = Task.Run(ProcessAsync, shutdownToken);
+        };
+    }
+}

Samples/MQTTnet.Samples.csproj

@@ -2,7 +2,7 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <TargetFramework>net6.0</TargetFramework>
+        <TargetFramework>net7.0</TargetFramework>
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>enable</Nullable>
         <IsPackable>false</IsPackable>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -33,8 +33,11 @@ namespace MQTTnet.AspNetCore
             PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
             _connection = connection ?? throw new ArgumentNullException(nameof(connection));
 
-            _input = connection.Transport.Input;
-            _output = connection.Transport.Output;
+            if (!(_connection is TcpConnection tcp) || tcp.IsConnected)
+            {
+                _input = connection.Transport.Input;
+                _output = connection.Transport.Output;
+            }
         }
 
         public long BytesReceived { get; private set; }

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Protocol;
@@ -7,8 +8,8 @@ namespace MQTTnet.Extensions.Rpc
 {
     public interface IMqttRpcClient : IDisposable
     {
-        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
+        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null);
         
-        Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default);
+        Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default);
     }
 }

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -42,13 +43,13 @@ namespace MQTTnet.Extensions.Rpc
             _waitingCalls.Clear();
         }
 
-        public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null)
         {
             using (var timeoutToken = new CancellationTokenSource(timeout))
             {
                 try
                 {
-                    return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, timeoutToken.Token).ConfigureAwait(false);
+                    return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, parameters, timeoutToken.Token).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException exception)
                 {
@@ -62,14 +63,14 @@ namespace MQTTnet.Extensions.Rpc
             }
         }
 
-        public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default)
+        public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
         {
             if (methodName == null)
             {
                 throw new ArgumentNullException(nameof(methodName));
             }
 
-            var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
+            var context = new TopicGenerationContext(_mqttClient, _options, methodName, parameters, qualityOfServiceLevel);
             var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);
 
             var requestTopic = topicNames.RequestTopic;

Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
 using System.Text;
 using System.Threading.Tasks;
 using MQTTnet.Protocol;
@@ -11,13 +12,13 @@ namespace MQTTnet.Extensions.Rpc
 {
     public static class MqttRpcClientExtensions
     {
-        public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));
 
             var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);
 
-            return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel);
+            return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel, parameters);
         }
     }
 }

Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
 using MQTTnet.Client;
 using MQTTnet.Protocol;
 
@@ -10,9 +11,11 @@ namespace MQTTnet.Extensions.Rpc
 {
     public sealed class TopicGenerationContext
     {
-        public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName,
+            IDictionary<string, object> parameters, MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
             MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
+            Parameters = parameters;
             QualityOfServiceLevel = qualityOfServiceLevel;
             MqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
             Options = options ?? throw new ArgumentNullException(nameof(options));
@@ -20,6 +23,8 @@ namespace MQTTnet.Extensions.Rpc
 
         public string MethodName { get; }
 
+        public IDictionary<string, object> Parameters { get; }
+
         public IMqttClient MqttClient { get; }
 
         public MqttRpcClientOptions Options { get; }

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -47,13 +47,13 @@ namespace MQTTnet.Extensions.WebSocket4Net
             var uri = _webSocketOptions.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
             {
-                if (_webSocketOptions.TlsOptions?.UseTls == false)
+                if (_webSocketOptions.TlsOptions?.UseTls == true)
                 {
-                    uri = "ws://" + uri;
+                    uri = "wss://" + uri;
                 }
                 else
                 {
-                    uri = "wss://" + uri;
+                    uri = "ws://" + uri;
                 }
             }
             
@@ -92,18 +92,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
             var webSocketVersion = WebSocketVersion.None;
             var receiveBufferSize = 0;
 
-            var certificates = new X509CertificateCollection();
-            if (_webSocketOptions.TlsOptions?.Certificates != null)
-            {
-                foreach (var certificate in _webSocketOptions.TlsOptions.Certificates)
-                {
-#if WINDOWS_UWP
-                    certificates.Add(new X509Certificate(certificate));
-#else
-                    certificates.Add(certificate);
-#endif
-                }
-            }
+            var certificates = _webSocketOptions.TlsOptions?.ClientCertificatesProvider?.GetCertificates();
 
             _webSocket = new WebSocket(uri, subProtocol, cookies, customHeaders, userAgent, origin, webSocketVersion, proxy, sslProtocols, receiveBufferSize)
             {

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -20,21 +20,25 @@ namespace MQTTnet.TestApp
         {
 #if NET5_0_OR_GREATER
             // TLS13 is only available in Net5.0
-            var unsafeTls13 = new MqttClientOptionsBuilderTlsParameters
+            var unsafeTls13 = new MqttClientTlsOptions
             {
                 UseTls = true,
                 SslProtocol = SslProtocols.Tls13,
                 // Don't use this in production code. This handler simply allows any invalid certificate to work.
-                CertificateValidationHandler = w => true
+                AllowUntrustedCertificates = true,
+                IgnoreCertificateChainErrors = true,
+                CertificateValidationHandler = _ => true
             };
 #endif
             // Also defining TLS12 for servers that don't seem no to support TLS13.
-            var unsafeTls12 = new MqttClientOptionsBuilderTlsParameters
+            var unsafeTls12 = new MqttClientTlsOptions
             {
                 UseTls = true,
                 SslProtocol = SslProtocols.Tls12,
                 // Don't use this in production code. This handler simply allows any invalid certificate to work.
-                CertificateValidationHandler = w => true
+                AllowUntrustedCertificates = true,
+                IgnoreCertificateChainErrors = true,
+                CertificateValidationHandler = _ => true
             };
 
             // mqtt.eclipseprojects.io
@@ -44,16 +48,16 @@ namespace MQTTnet.TestApp
 
             await ExecuteTestAsync(
                 "mqtt.eclipseprojects.io WS",
-                new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:80/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
+                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",
-                new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:443/mqtt")
-                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
+                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)",
-                new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:443/mqtt")
-                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build(),
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
+                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build(),
                     true);
 #endif
 
@@ -68,12 +72,12 @@ namespace MQTTnet.TestApp
 
             await ExecuteTestAsync(
                 "test.mosquitto.org TCP TLS12",
-                new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+                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",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
-                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
+                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
             await ExecuteTestAsync(
@@ -81,26 +85,26 @@ namespace MQTTnet.TestApp
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8885)
                     .WithCredentials("rw", "readwrite")
                     .WithProtocolVersion(MqttProtocolVersion.V311)
-                    .WithTls(unsafeTls12)
+                    .WithTlsOptions(unsafeTls12)
                     .Build());
 
             await ExecuteTestAsync(
                 "test.mosquitto.org WS",
-                new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
             await ExecuteTestAsync(
                 "test.mosquitto.org WS (WebSocket4Net)",
-                new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
             await ExecuteTestAsync(
                 "test.mosquitto.org WS TLS12",
-                new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            // await ExecuteTestAsync(
-            //     "test.mosquitto.org WS TLS12 (WebSocket4Net)",
-            //     new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build(),
-            //     true);
+            await ExecuteTestAsync(
+                "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(
@@ -109,30 +113,30 @@ namespace MQTTnet.TestApp
 
             await ExecuteTestAsync(
                 "broker.emqx.io TCP TLS12",
-                new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+                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",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
-                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
+                    .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
             await ExecuteTestAsync(
                 "broker.emqx.io WS",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8083/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8083/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
             await ExecuteTestAsync(
                 "broker.emqx.io WS (WebSocket4Net)",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
             await ExecuteTestAsync(
                 "broker.emqx.io WS TLS12",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
             await ExecuteTestAsync(
                 "broker.emqx.io WS TLS12 (WebSocket4Net)",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build(),
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
                 true);
 
             // broker.hivemq.com
@@ -142,15 +146,14 @@ namespace MQTTnet.TestApp
 
             await ExecuteTestAsync(
                 "broker.hivemq.com WS",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
             await ExecuteTestAsync(
                 "broker.hivemq.com WS (WebSocket4Net)",
-                new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+                new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
             // mqtt.swifitch.cz: Does not seem to operate any more
-
             // cloudmqtt.com: Cannot test because it does not offer a free plan any more.
 
             Write("Finished.", ConsoleColor.White);
@@ -197,9 +200,10 @@ namespace MQTTnet.TestApp
 
                 Write("[OK]\n", ConsoleColor.Green);
             }
-            catch (Exception e)
+            catch (Exception exception)
             {
-                Write("[FAILED] " + e.Message + "\n", ConsoleColor.Red);
+                Write("[FAILED]" + Environment.NewLine, ConsoleColor.Red);
+                Write(exception + Environment.NewLine, ConsoleColor.Red);
             }
         }
 

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

@@ -22,99 +22,6 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
     [TestClass]
     public sealed class ManagedMqttClient_Tests : BaseTestClass
     {
-        [TestMethod]
-        public async Task Expose_Custom_Connection_Error()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var server = await testEnvironment.StartServer();
-                
-                server.ValidatingConnectionAsync += args =>
-                {
-                    args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
-                    return CompletedTask.Instance;
-                };
-
-                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
-
-                MqttClientDisconnectedEventArgs disconnectedArgs = null;
-                managedClient.DisconnectedAsync += args =>
-                {
-                    disconnectedArgs = args;
-                    return CompletedTask.Instance;
-                };
-                
-                var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
-                await managedClient.StartAsync(clientOptions);
-
-                await LongTestDelay();
-
-                Assert.IsNotNull(disconnectedArgs);
-                Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Receive_While_Not_Cleanly_Disconnected()
-        {
-            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-            {
-                await testEnvironment.StartServer(o => o.WithPersistentSessions());
-
-                var senderClient = await testEnvironment.ConnectClient();
-
-                // Prepare managed client.
-                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
-                await managedClient.SubscribeAsync("#");
-                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
-                
-                var managedClientOptions = new ManagedMqttClientOptions
-                {
-                    ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
-                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                        .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
-                        .WithCleanSession(false)
-                        .Build()
-                };
-
-                await managedClient.StartAsync(managedClientOptions);
-                await LongTestDelay();
-                await LongTestDelay();
-                
-                // Send test data.
-                await senderClient.PublishStringAsync("topic1");
-                await LongTestDelay();
-                await LongTestDelay();
-                
-                receivedMessages.AssertReceivedCountEquals(1);
-
-                // Stop the managed client but keep session at server (not clean disconnect required).
-                await managedClient.StopAsync(false);
-                await LongTestDelay();
-                
-                // Send new messages in the meantime.
-                await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
-                await LongTestDelay();
-                
-                // Start the managed client, it should receive the new message.
-                await managedClient.StartAsync(managedClientOptions);
-                await LongTestDelay();
-                
-                receivedMessages.AssertReceivedCountEquals(2);
-                
-                // Stop and start again, no new message should be received.
-                for (var i = 0; i < 3; i++)
-                {
-                    await managedClient.StopAsync(false);
-                    await LongTestDelay();
-                    await managedClient.StartAsync(managedClientOptions);
-                    await LongTestDelay();
-                }
-
-                receivedMessages.AssertReceivedCountEquals(2);
-            }
-        }
-
         [TestMethod]
         public async Task Connect_To_Invalid_Server()
         {
@@ -181,6 +88,38 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+        [TestMethod]
+        public async Task Expose_Custom_Connection_Error()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                server.ValidatingConnectionAsync += args =>
+                {
+                    args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
+                    return CompletedTask.Instance;
+                };
+
+                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
+
+                MqttClientDisconnectedEventArgs disconnectedArgs = null;
+                managedClient.DisconnectedAsync += args =>
+                {
+                    disconnectedArgs = args;
+                    return CompletedTask.Instance;
+                };
+
+                var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
+                await managedClient.StartAsync(clientOptions);
+
+                await LongTestDelay();
+
+                Assert.IsNotNull(disconnectedArgs);
+                Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
+            }
+        }
+
         [TestMethod]
         public async Task ManagedClients_Will_Message_Send()
         {
@@ -224,6 +163,67 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+        [TestMethod]
+        public async Task Receive_While_Not_Cleanly_Disconnected()
+        {
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                var senderClient = await testEnvironment.ConnectClient();
+
+                // Prepare managed client.
+                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
+                await managedClient.SubscribeAsync("#");
+                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
+
+                var managedClientOptions = new ManagedMqttClientOptions
+                {
+                    ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
+                        .WithCleanSession(false)
+                        .Build()
+                };
+
+                await managedClient.StartAsync(managedClientOptions);
+                await LongTestDelay();
+                await LongTestDelay();
+
+                // Send test data.
+                await senderClient.PublishStringAsync("topic1");
+                await LongTestDelay();
+                await LongTestDelay();
+
+                receivedMessages.AssertReceivedCountEquals(1);
+
+                // Stop the managed client but keep session at server (not clean disconnect required).
+                await managedClient.StopAsync(false);
+                await LongTestDelay();
+
+                // Send new messages in the meantime.
+                await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
+                await LongTestDelay();
+
+                // Start the managed client, it should receive the new message.
+                await managedClient.StartAsync(managedClientOptions);
+                await LongTestDelay();
+
+                receivedMessages.AssertReceivedCountEquals(2);
+
+                // Stop and start again, no new message should be received.
+                for (var i = 0; i < 3; i++)
+                {
+                    await managedClient.StopAsync(false);
+                    await LongTestDelay();
+                    await managedClient.StartAsync(managedClientOptions);
+                    await LongTestDelay();
+                }
+
+                receivedMessages.AssertReceivedCountEquals(2);
+            }
+        }
+
         [TestMethod]
         public async Task Start_Stop()
         {
@@ -375,7 +375,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort);
 
                 var receivedManagedMessages = new List<MqttApplicationMessage>();
-                
+
                 var managedClient = testEnvironment.Factory.CreateManagedMqttClient(testEnvironment.CreateClient());
                 managedClient.ApplicationMessageReceivedAsync += e =>
                 {
@@ -403,7 +403,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 // Make sure that it gets received after subscribing again.
                 await managedClient.SubscribeAsync("topic");
                 await LongTestDelay();
-                
+
                 Assert.AreEqual(2, receivedManagedMessages.Count);
             }
         }
@@ -421,7 +421,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 var receivingClient = await CreateManagedClientAsync(testEnvironment, null, connectionCheckInterval);
                 var sendingClient = await testEnvironment.ConnectClient();
 
-                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", Payload = new byte[] { 1 }, Retain = true });
+                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>(new byte[] { 1 }), Retain = true });
 
                 var subscribeTime = DateTime.UtcNow;
 
@@ -454,7 +454,7 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
                 //wait a bit for the subscription to become established
                 await Task.Delay(500);
 
-                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", Payload = new byte[] { 1 }, Retain = true });
+                await sendingClient.PublishAsync(new MqttApplicationMessage { Topic = "topic", PayloadSegment = new ArraySegment<byte>(new byte[] { 1 }), Retain = true });
 
                 var messages = await SetupReceivingOfMessages(managedClient, 1);
 

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs

@@ -167,6 +167,38 @@ namespace MQTTnet.Tests.Clients.MqttClient
             }
         }
 
+        [TestMethod]
+        public async Task No_Unobserved_Exception()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                var client = testEnvironment.CreateClient();
+                var options = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithTimeout(TimeSpan.FromSeconds(2)).Build();
+
+                try
+                {
+                    using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(0.5)))
+                    {
+                        await client.ConnectAsync(options, timeout.Token);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                }
+
+                client.Dispose();
+
+                // These delays and GC calls are required in order to make calling the finalizer reproducible.
+                GC.Collect();
+                GC.WaitForPendingFinalizers();
+                await LongTestDelay();
+                await LongTestDelay();
+                await LongTestDelay();
+            }
+        }
+
         [TestMethod]
         public async Task Return_Non_Success()
         {
@@ -196,5 +228,27 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 Assert.AreEqual(response.UserProperties[0].Value, "Value");
             }
         }
+
+        [TestMethod]
+        public async Task Throw_Proper_Exception_When_Not_Connected()
+        {
+            try
+            {
+                var mqttFactory = new MqttFactory();
+                using (var mqttClient = mqttFactory.CreateMqttClient())
+                {
+                    await mqttClient.SubscribeAsync("test", MqttQualityOfServiceLevel.AtLeastOnce);
+                }
+            }
+            catch (MqttCommunicationException exception)
+            {
+                if (exception.Message == "The client is not connected.")
+                {
+                    return;
+                }
+            }
+
+            Assert.Fail();
+        }
     }
 }

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs

@@ -296,7 +296,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
 
                 Assert.IsNotNull(receivedMessage);
                 Assert.AreEqual("A", receivedMessage.Topic);
-                Assert.AreEqual(null, receivedMessage.Payload);
+                Assert.AreEqual(null, receivedMessage.PayloadSegment.Array);
             }
         }
 
@@ -507,7 +507,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
 
                 client2.ApplicationMessageReceivedAsync += e =>
                 {
-                    client2TopicResults.Add(Encoding.UTF8.GetString(e.ApplicationMessage.Payload));
+                    client2TopicResults.Add(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment.ToArray()));
                     return CompletedTask.Instance;
                 };
 

Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -45,6 +46,34 @@ namespace MQTTnet.Tests.Extensions
             }
         }
 
+        [TestMethod]
+        public async Task Execute_Success_Parameters_Propagated_Correctly()
+        {
+            var paramValue = "123";
+            var parameters = new Dictionary<string, object>
+            {
+                { TestParametersTopicGenerationStrategy.ExpectedParamName, "123" }
+            };
+
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var responseSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
+                await responseSender.SubscribeAsync($"MQTTnet.RPC/+/ping/{paramValue}");
+
+                responseSender.ApplicationMessageReceivedAsync += e => responseSender.PublishStringAsync(e.ApplicationMessage.Topic + "/response", "pong");
+
+                using (var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder()
+                           .WithTopicGenerationStrategy(new TestParametersTopicGenerationStrategy()).Build()))
+                {
+                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce, parameters);
+
+                    Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                }
+            }
+        }
+
         [TestMethod]
         public Task Execute_Success_With_QoS_0()
         {
@@ -135,7 +164,7 @@ namespace MQTTnet.Tests.Extensions
 
                 using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
                 {
-                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+                    await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
                 }
             }
         }
@@ -222,5 +251,32 @@ namespace MQTTnet.Tests.Extensions
                 };
             }
         }
+
+        class TestParametersTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
+        {
+            internal const string ExpectedParamName = "test_param_name";
+
+            public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
+            {
+                if (context.Parameters == null)
+                {
+                    throw new InvalidOperationException("Parameters dictionary expected to be not null");
+                }
+
+                if (!context.Parameters.TryGetValue(ExpectedParamName, out var paramValue))
+                {
+                    throw new InvalidOperationException($"Parameter with name {ExpectedParamName} not present");
+                }
+
+                var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{context.MethodName}/{paramValue}";
+                var responseTopic = requestTopic + "/response";
+
+                return new MqttRpcTopicPair
+                {
+                    RequestTopic = requestTopic,
+                    ResponseTopic = responseTopic
+                };
+            }
+        }
     }
 }

Source/MQTTnet.Tests/Extensions/WebSocket4Net_Tests.cs

@@ -23,7 +23,7 @@ namespace MQTTnet.Tests.Extensions
 
             using (var client = factory.CreateMqttClient())
             {
-                var options = new MqttClientOptionsBuilder().WithWebSocketServer("ws://a.b/mqtt").WithTimeout(TimeSpan.FromSeconds(2)).Build();
+                var options = new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("ws://a.b/mqtt")).WithTimeout(TimeSpan.FromSeconds(2)).Build();
                 await client.ConnectAsync(options).ConfigureAwait(false);
             }
         }

Source/MQTTnet.Tests/MQTTnet.Tests.csproj

@@ -1,7 +1,8 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
-        <TargetFrameworks>net452;net461;net48;netcoreapp3.1;net6.0;net7.0</TargetFrameworks>
+        <TargetFrameworks>netcoreapp3.1;net7.0</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net48</TargetFrameworks>
         <IsPackable>false</IsPackable>
         <LangVersion>7.3</LangVersion>
         <EnableNETAnalyzers>false</EnableNETAnalyzers>

Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs

@@ -273,7 +273,7 @@ namespace MQTTnet.Tests.MQTTv5
                 Assert.AreEqual(applicationMessage.ResponseTopic, receivedMessage.ResponseTopic);
                 Assert.AreEqual(applicationMessage.MessageExpiryInterval, receivedMessage.MessageExpiryInterval);
                 CollectionAssert.AreEqual(applicationMessage.CorrelationData, receivedMessage.CorrelationData);
-                CollectionAssert.AreEqual(applicationMessage.Payload, receivedMessage.Payload);
+                CollectionAssert.AreEqual(applicationMessage.PayloadSegment.ToArray(), receivedMessage.PayloadSegment.ToArray());
                 CollectionAssert.AreEqual(applicationMessage.UserProperties, receivedMessage.UserProperties);
             }
         }

Source/MQTTnet.Tests/Mockups/TestEnvironment.cs

@@ -3,7 +3,6 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -25,9 +24,9 @@ namespace MQTTnet.Tests.Mockups
     public sealed class TestEnvironment : IDisposable
     {
         readonly List<string> _clientErrors = new List<string>();
-        readonly ConcurrentBag<IMqttClient> _clients = new ConcurrentBag<IMqttClient>();
+        readonly List<IMqttClient> _clients = new List<IMqttClient>();
         readonly List<Exception> _exceptions = new List<Exception>();
-        readonly ConcurrentBag<ILowLevelMqttClient> _lowLevelClients = new ConcurrentBag<ILowLevelMqttClient>();
+        readonly List<ILowLevelMqttClient> _lowLevelClients = new List<ILowLevelMqttClient>();
         readonly MqttProtocolVersion _protocolVersion;
         readonly List<string> _serverErrors = new List<string>();
 
@@ -40,6 +39,8 @@ namespace MQTTnet.Tests.Mockups
             _protocolVersion = protocolVersion;
             TestContext = testContext;
 
+            TaskScheduler.UnobservedTaskException += TrackUnobservedTaskException;
+
             ServerLogger.LogMessagePublished += (s, e) =>
             {
                 if (Debugger.IsAttached)
@@ -216,8 +217,9 @@ namespace MQTTnet.Tests.Mockups
         public IMqttClient CreateClient()
         {
             var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
-            
+
             var client = Factory.CreateMqttClient(logger);
+
             client.ConnectingAsync += e =>
             {
                 if (TestContext != null)
@@ -232,9 +234,12 @@ namespace MQTTnet.Tests.Mockups
 
                 return CompletedTask.Instance;
             };
-            
-            _clients.Add(client);
-            
+
+            lock (_clients)
+            {
+                _clients.Add(client);
+            }
+
             return client;
         }
 
@@ -245,13 +250,20 @@ namespace MQTTnet.Tests.Mockups
 
         public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder()
         {
-            return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort).WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
+            return Factory.CreateClientOptionsBuilder()
+                .WithProtocolVersion(_protocolVersion)
+                .WithTcpServer("127.0.0.1", ServerPort)
+                .WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
         }
 
         public ILowLevelMqttClient CreateLowLevelClient()
         {
             var client = Factory.CreateLowLevelMqttClient(ClientLogger);
-            _lowLevelClients.Add(client);
+
+            lock (_lowLevelClients)
+            {
+                _lowLevelClients.Add(client);
+            }
 
             return client;
         }
@@ -287,45 +299,68 @@ namespace MQTTnet.Tests.Mockups
 
         public void Dispose()
         {
-            foreach (var mqttClient in _clients)
+            try
             {
+                lock (_clients)
+                {
+                    foreach (var mqttClient in _clients)
+                    {
+                        try
+                        {
+                            //mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+                        }
+                        catch
+                        {
+                            // This can happen when the test already disconnected the client.
+                        }
+                        finally
+                        {
+                            mqttClient?.Dispose();
+                        }
+                    }
+
+                    _clients.Clear();
+                }
+
+                lock (_lowLevelClients)
+                {
+                    foreach (var lowLevelMqttClient in _lowLevelClients)
+                    {
+                        lowLevelMqttClient.Dispose();
+                    }
+
+                    _lowLevelClients.Clear();
+                }
+
                 try
                 {
-                    //mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+                    Server?.StopAsync().GetAwaiter().GetResult();
                 }
                 catch
                 {
-                    // This can happen when the test already disconnected the client.
+                    // This can happen when the test already stopped the server.
                 }
                 finally
                 {
-                    mqttClient?.Dispose();
+                    Server?.Dispose();
                 }
-            }
 
-            foreach (var lowLevelMqttClient in _lowLevelClients)
-            {
-                lowLevelMqttClient.Dispose();
-            }
+                Server = null;
 
-            try
-            {
-                Server?.StopAsync().GetAwaiter().GetResult();
-            }
-            catch
-            {
-                // This can happen when the test already stopped the server.
+                ThrowIfLogErrors();
+
+                GC.Collect();
+                GC.WaitForFullGCComplete();
+                GC.WaitForPendingFinalizers();
+                
+                if (_exceptions.Any())
+                {
+                    throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
+                }
             }
             finally
             {
-                Server?.Dispose();
-            }
-
-            ThrowIfLogErrors();
-
-            if (_exceptions.Any())
-            {
-                throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
+                TaskScheduler.UnobservedTaskException -= TrackUnobservedTaskException;
             }
         }
 
@@ -342,7 +377,7 @@ namespace MQTTnet.Tests.Mockups
 
             var options = optionsBuilder.Build();
             var server = CreateServer(options);
-            await server.StartAsync();
+            await server.StartAsync().ConfigureAwait(false);
 
             // The OS has chosen the port to we have to properly expose it to the tests.
             ServerPort = options.DefaultEndpointOptions.Port;
@@ -370,13 +405,16 @@ namespace MQTTnet.Tests.Mockups
 
         public void ThrowIfLogErrors()
         {
-            lock (_serverErrors)
+            if (!IgnoreServerLogErrors)
             {
-                if (!IgnoreServerLogErrors && _serverErrors.Count > 0)
+                lock (_serverErrors)
                 {
-                    var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
-                    Console.WriteLine(message);
-                    throw new Exception(message);
+                    if (_serverErrors.Count > 0)
+                    {
+                        var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
+                        Console.WriteLine(message);
+                        throw new Exception(message);
+                    }
                 }
             }
 
@@ -396,10 +434,20 @@ namespace MQTTnet.Tests.Mockups
 
         public void TrackException(Exception exception)
         {
+            if (exception == null)
+            {
+                return;
+            }
+
             lock (_exceptions)
             {
                 _exceptions.Add(exception);
             }
         }
+
+        void TrackUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e)
+        {
+            TrackException(e.Exception);
+        }
     }
 }

Source/MQTTnet.Tests/MqttApplicationMessageBuilder_Tests.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.IO;
+using System.Linq;
 using System.Text;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Protocol;
@@ -11,7 +12,7 @@ using MQTTnet.Protocol;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttApplicationMessageBuilder_Tests
+    public sealed class MqttApplicationMessageBuilder_Tests
     {
         [TestMethod]
         public void CreateApplicationMessage_TopicOnly()
@@ -29,7 +30,7 @@ namespace MQTTnet.Tests
             Assert.AreEqual("xyz", message.Topic);
             Assert.IsFalse(message.Retain);
             Assert.AreEqual(MqttQualityOfServiceLevel.AtMostOnce, message.QualityOfServiceLevel);
-            Assert.AreEqual(Encoding.UTF8.GetString(message.Payload), "00:06:00");
+            Assert.AreEqual(Encoding.UTF8.GetString(message.PayloadSegment.ToArray()), "00:06:00");
         }
 
         [TestMethod]
@@ -41,7 +42,7 @@ namespace MQTTnet.Tests
             Assert.AreEqual("123", message.Topic);
             Assert.IsFalse(message.Retain);
             Assert.AreEqual(MqttQualityOfServiceLevel.AtMostOnce, message.QualityOfServiceLevel);
-            Assert.AreEqual(Encoding.UTF8.GetString(message.Payload), "Hello");
+            Assert.AreEqual(Encoding.UTF8.GetString(message.PayloadSegment.ToArray()), "Hello");
         }
 
         [TestMethod]

Source/MQTTnet.Tests/MqttApplicationMessageTest.cs

@@ -1,3 +1,5 @@
+#pragma warning disable CS0618 // Type or member is obsolete
+
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using System;
 using System.Collections.Generic;
@@ -8,7 +10,7 @@ using System.Threading.Tasks;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttApplicationMessageTest
+    public sealed class MqttApplicationMessageTest
     {
         [TestMethod]
         public void PayloadSegment()
@@ -17,32 +19,27 @@ namespace MQTTnet.Tests
             Assert.AreEqual(0, message.Payload.Length);
             Assert.AreEqual(0, message.PayloadSegment.Count);
 
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.Payload = new byte[] { 1, 2 };
             Assert.AreEqual(2, message.Payload.Length);
             Assert.AreEqual(2, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.Payload = new byte[] { 1, 2, 3 };
             Assert.AreEqual(3, message.Payload.Length);
             Assert.AreEqual(3, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 });
             Assert.AreEqual(3, message.Payload.Length);
             Assert.AreEqual(3, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsTrue(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
 
             message.PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3 }, 1, 1);
             Assert.AreEqual(1, message.Payload.Length);
             Assert.AreEqual(1, message.PayloadSegment.Count);
-            Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload));
-            Assert.IsFalse(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array));
+            Assert.IsFalse(ReferenceEquals(message.Payload, message.PayloadSegment.Array));
         }
     }
 }

Source/MQTTnet.Tests/Server/HotSwapCerts_Tests.cs

@@ -0,0 +1,426 @@
+#if !(NET452 || NET461 || NET48)
+using System;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Linq;
+using System.Net;
+using System.Net.Security;
+using System.Security.Authentication;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Certificates;
+using MQTTnet.Client;
+using MQTTnet.Formatter;
+using MQTTnet.Protocol;
+using MQTTnet.Server;
+
+namespace MQTTnet.Tests.Server
+{
+    // missing certificate builder api means tests won't work for older frameworks
+
+    [TestClass]
+    public sealed class HotSwapCerts_Tests
+    {
+        static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10);
+
+        [TestMethod]
+        public async Task ClientCertChangeWithoutServerUpdateFailsReconnect()
+        {
+            using (var server = new ServerTestHarness())
+            using (var client01 = new ClientTestHarness())
+            {
+                server.InstallNewClientCert(client01.GetCurrentClientCert());
+                client01.InstallNewServerCert(server.GetCurrentServerCert());
+
+                await server.StartServer();
+
+                await client01.Connect();
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+
+                client01.HotSwapClientCert();
+                server.ForceDisconnectAsync(client01).Wait(DefaultTimeout);
+                client01.WaitForDisconnectOrFail(DefaultTimeout);
+
+                client01.WaitForConnectToFail(DefaultTimeout);
+            }
+        }
+
+        [TestMethod]
+        public async Task ClientCertChangeWithServerUpdateAcceptsReconnect()
+        {
+            using (var server = new ServerTestHarness())
+            using (var client01 = new ClientTestHarness())
+            {
+                server.InstallNewClientCert(client01.GetCurrentClientCert());
+                client01.InstallNewServerCert(server.GetCurrentServerCert());
+
+                await server.StartServer();
+
+                await client01.Connect();
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+
+                client01.HotSwapClientCert();
+                server.ForceDisconnectAsync(client01).Wait(DefaultTimeout);
+                client01.WaitForDisconnectOrFail(DefaultTimeout);
+
+                server.InstallNewClientCert(client01.GetCurrentClientCert());
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+            }
+        }
+
+        [TestMethod]
+        public async Task ServerCertChangeWithClientCertUpdateAllowsReconnect()
+        {
+            using (var server = new ServerTestHarness())
+            using (var client01 = new ClientTestHarness())
+            {
+                server.InstallNewClientCert(client01.GetCurrentClientCert());
+                client01.InstallNewServerCert(server.GetCurrentServerCert());
+
+                await server.StartServer();
+                await client01.Connect();
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+                server.HotSwapServerCert();
+
+                server.ForceDisconnectAsync(client01).Wait(DefaultTimeout);
+                client01.WaitForDisconnectOrFail(DefaultTimeout);
+                client01.InstallNewServerCert(server.GetCurrentServerCert());
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+            }
+        }
+
+        [TestMethod]
+        public async Task ServerCertChangeWithoutClientCertUpdateFailsReconnect()
+        {
+            using (var server = new ServerTestHarness())
+            using (var client01 = new ClientTestHarness())
+            {
+                server.InstallNewClientCert(client01.GetCurrentClientCert());
+                client01.InstallNewServerCert(server.GetCurrentServerCert());
+
+                await server.StartServer();
+                await client01.Connect();
+
+                client01.WaitForConnectOrFail(DefaultTimeout);
+                server.HotSwapServerCert();
+
+                server.ForceDisconnectAsync(client01).Wait(DefaultTimeout);
+                client01.WaitForDisconnectOrFail(DefaultTimeout);
+
+                client01.WaitForConnectToFail(DefaultTimeout);
+            }
+        }
+
+        static X509Certificate2 CreateSelfSignedCertificate(string oid)
+        {
+            var sanBuilder = new SubjectAlternativeNameBuilder();
+            sanBuilder.AddIpAddress(IPAddress.Loopback);
+            sanBuilder.AddIpAddress(IPAddress.IPv6Loopback);
+            sanBuilder.AddDnsName("localhost");
+
+            using (var rsa = RSA.Create())
+            {
+                var certRequest = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA512, RSASignaturePadding.Pkcs1);
+
+                certRequest.CertificateExtensions.Add(
+                    new X509KeyUsageExtension(X509KeyUsageFlags.DataEncipherment | X509KeyUsageFlags.KeyEncipherment | X509KeyUsageFlags.DigitalSignature, false));
+
+                certRequest.CertificateExtensions.Add(new X509EnhancedKeyUsageExtension(new OidCollection { new Oid(oid) }, false));
+
+                certRequest.CertificateExtensions.Add(sanBuilder.Build());
+
+                using (var certificate = certRequest.CreateSelfSigned(DateTimeOffset.Now.AddMinutes(-10), DateTimeOffset.Now.AddMinutes(10)))
+                {
+                    var pfxCertificate = new X509Certificate2(
+                        certificate.Export(X509ContentType.Pfx),
+                        (string)null,
+                        X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
+
+                    return pfxCertificate;
+                }
+            }
+        }
+
+        sealed class ClientTestHarness : IDisposable
+        {
+            readonly HotSwappableClientCertProvider _hotSwapClient = new HotSwappableClientCertProvider();
+
+            IMqttClient _client;
+
+            public string ClientId => _client.Options.ClientId;
+
+            public Task Connect()
+            {
+                return Run_Client_Connection();
+            }
+
+            public void Dispose()
+            {
+                _client.Dispose();
+                _hotSwapClient.Dispose();
+            }
+
+            public X509Certificate2 GetCurrentClientCert()
+            {
+                var result = _hotSwapClient.GetCertificates()[0];
+                return new X509Certificate2(result);
+            }
+
+            public void HotSwapClientCert()
+            {
+                _hotSwapClient.HotSwapCert();
+            }
+
+            public void InstallNewServerCert(X509Certificate2 serverCert)
+            {
+                _hotSwapClient.InstallNewServerCert(serverCert);
+            }
+
+            public void WaitForConnectOrFail(TimeSpan timeout)
+            {
+                Thread.Sleep(100);
+
+                if (!_client.IsConnected)
+                {
+                    _client.ReconnectAsync().Wait(timeout);
+                }
+
+                WaitForConnect(timeout);
+
+                Assert.IsNotNull(_client, "Client was never initialized");
+                Assert.IsTrue(_client.IsConnected, $"Client connection failed after {timeout}");
+            }
+
+            public void WaitForConnectToFail(TimeSpan timeout)
+            {
+                Assert.IsFalse(_client.IsConnected, "Client should be disconnected before waiting for connect.");
+
+                WaitForConnect(timeout);
+
+                Assert.IsNotNull(_client, "Client was never initialized");
+                Assert.IsFalse(_client.IsConnected, "Client connection success but test wanted fail");
+            }
+
+            void WaitForDisconnect(TimeSpan timeout)
+            {
+                var timer = Stopwatch.StartNew();
+                while ((_client == null || _client.IsConnected) && timer.Elapsed < timeout)
+                {
+                    Thread.Sleep(100);
+                }
+            }
+
+            public void WaitForDisconnectOrFail(TimeSpan timeout)
+            {
+                WaitForDisconnect(timeout);
+
+                Assert.IsNotNull(_client, "Client was never initialized");
+                Assert.IsFalse(_client.IsConnected, $"Client connection should have disconnected after {timeout}");
+            }
+
+            async Task Run_Client_Connection()
+            {
+                var optionsBuilder = new MqttClientOptionsBuilder()
+                    .WithTlsOptions(
+                        o => o.WithClientCertificatesProvider(_hotSwapClient)
+                            .WithCertificateValidationHandler(_hotSwapClient.OnCertificateValidation)
+                            .WithSslProtocols(SslProtocols.Tls12))
+                    .WithTcpServer("localhost")
+                    .WithCleanSession()
+                    .WithProtocolVersion(MqttProtocolVersion.V500);
+
+                var mqttClientOptions = optionsBuilder.Build();
+
+                var factory = new MqttFactory();
+                var mqttClient = factory.CreateMqttClient();
+                _client = mqttClient;
+
+                await mqttClient.ConnectAsync(mqttClientOptions);
+            }
+
+            void WaitForConnect(TimeSpan timeout)
+            {
+                var timer = Stopwatch.StartNew();
+                while ((_client == null || !_client.IsConnected) && timer.Elapsed < timeout)
+                {
+                    Thread.Sleep(100);
+                }
+            }
+        }
+
+        sealed class ServerTestHarness : IDisposable
+        {
+            readonly HotSwappableServerCertProvider _hotSwapServer = new HotSwappableServerCertProvider();
+
+            MqttServer _server;
+
+            public void Dispose()
+            {
+                if (_server != null)
+                {
+                    _server.StopAsync().Wait();
+                    _server.Dispose();
+                }
+
+                _hotSwapServer?.Dispose();
+            }
+
+            public Task ForceDisconnectAsync(ClientTestHarness client)
+            {
+                return _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
+            }
+
+            public X509Certificate2 GetCurrentServerCert()
+            {
+                return _hotSwapServer.GetCertificate();
+            }
+
+            public void HotSwapServerCert()
+            {
+                _hotSwapServer.HotSwapCert();
+            }
+
+            public void InstallNewClientCert(X509Certificate2 serverCert)
+            {
+                _hotSwapServer.InstallNewClientCert(serverCert);
+            }
+
+            public Task StartServer()
+            {
+                var mqttFactory = new MqttFactory();
+
+                var mqttServerOptions = new MqttServerOptionsBuilder().WithEncryptionCertificate(_hotSwapServer)
+                    .WithRemoteCertificateValidationCallback(_hotSwapServer.RemoteCertificateValidationCallback)
+                    .WithEncryptedEndpoint()
+                    .Build();
+
+                mqttServerOptions.TlsEndpointOptions.ClientCertificateRequired = true;
+
+                _server = mqttFactory.CreateMqttServer(mqttServerOptions);
+                return _server.StartAsync();
+            }
+        }
+
+        class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable
+        {
+            X509Certificate2Collection _certificates;
+            ConcurrentBag<X509Certificate2> _serverCerts = new ConcurrentBag<X509Certificate2>();
+
+            public HotSwappableClientCertProvider()
+            {
+                _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
+            }
+            
+            public void Dispose()
+            {
+                if (_certificates != null)
+                {
+                    foreach (var certs in _certificates)
+                    {
+#if !NET452
+                        certs.Dispose();
+#endif
+                    }
+                }
+            }
+
+            public X509CertificateCollection GetCertificates()
+            {
+                return new X509Certificate2Collection(_certificates);
+            }
+
+            public void HotSwapCert()
+            {
+                _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
+            }
+
+            public void InstallNewServerCert(X509Certificate2 serverCert)
+            {
+                _serverCerts.Add(serverCert);
+            }
+
+            public bool OnCertificateValidation(MqttClientCertificateValidationEventArgs certContext)
+            {
+                var serverCerts = _serverCerts.ToArray();
+
+                var providedCert = certContext.Certificate.GetRawCertData();
+                for (int i = 0, n = serverCerts.Length; i < n; i++)
+                {
+                    var currentcert = serverCerts[i];
+
+                    if (currentcert.RawData.SequenceEqual(providedCert))
+                    {
+                        return true;
+                    }
+                }
+
+                return false;
+            }
+        }
+
+        sealed class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
+        {
+            readonly ConcurrentBag<X509Certificate2> _clientCerts = new ConcurrentBag<X509Certificate2>();
+            X509Certificate2 _certificate;
+
+            public HotSwappableServerCertProvider()
+            {
+                _certificate = CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.1");
+            }
+
+            public void Dispose()
+            {
+#if !NET452
+                _certificate.Dispose();
+#endif
+            }
+
+            public X509Certificate2 GetCertificate()
+            {
+                return _certificate;
+            }
+
+            public void HotSwapCert()
+            {
+                var newCert = CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.1");
+                var oldCert = Interlocked.Exchange(ref _certificate, newCert);
+#if !NET452
+                oldCert.Dispose();
+#endif
+            }
+
+            public void InstallNewClientCert(X509Certificate2 certificate)
+            {
+                _clientCerts.Add(certificate);
+            }
+
+            public bool RemoteCertificateValidationCallback(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
+            {
+                var serverCerts = _clientCerts.ToArray();
+
+                var providedCert = certificate.GetRawCertData();
+                for (int i = 0, n = serverCerts.Length; i < n; i++)
+                {
+                    var currentCert = serverCerts[i];
+
+                    if (currentCert.RawData.SequenceEqual(providedCert))
+                    {
+                        return true;
+                    }
+                }
+
+                return false;
+            }
+        }
+    }
+}
+
+#endif

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -26,9 +26,7 @@ namespace MQTTnet.Adapter
         readonly IMqttChannel _channel;
         readonly byte[] _fixedHeaderBuffer = new byte[2];
         readonly MqttNetSourceLogger _logger;
-
         readonly byte[] _singleByteBuffer = new byte[1];
-
         readonly AsyncLock _syncRoot = new AsyncLock();
 
         Statistics _statistics; // mutable struct, don't make readonly!
@@ -47,6 +45,8 @@ namespace MQTTnet.Adapter
             _logger = logger.WithSource(nameof(MqttChannelAdapter));
         }
 
+        public bool AllowPacketFragmentation { get; set; } = true;
+
         public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived);
 
         public long BytesSent => Volatile.Read(ref _statistics._bytesSent);
@@ -63,8 +63,6 @@ namespace MQTTnet.Adapter
 
         public MqttPacketInspector PacketInspector { get; set; }
 
-        public bool AllowPacketFragmentation { get; set; } = true;
-
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
@@ -78,12 +76,30 @@ namespace MQTTnet.Adapter
                  * block forever. Even a cancellation token is not supported properly.
                  */
 
-                var connectTask = _channel.ConnectAsync(cancellationToken);
-
                 var timeout = new TaskCompletionSource<object>();
                 using (cancellationToken.Register(() => timeout.TrySetResult(null)))
                 {
+                    var connectTask = Task.Run(
+                        async () =>
+                        {
+                            try
+                            {
+                                await _channel.ConnectAsync(cancellationToken).ConfigureAwait(false);
+                            }
+                            catch
+                            {
+                                // If the timeout is already reached the exception is no longer of interest and
+                                // must be catched. Otherwise it will arrive at the TaskScheduler.UnobservedTaskException.
+                                if (!timeout.Task.IsCompleted)
+                                {
+                                    throw;
+                                }
+                            }
+                        },
+                        CancellationToken.None);
+
                     await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
+
                     if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
                     {
                         throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
@@ -204,7 +220,7 @@ namespace MQTTnet.Adapter
                     PacketInspector?.BeginSendPacket(packetBuffer);
 
                     _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
-                    
+
                     if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
                     {
                         await _channel.WriteAsync(packetBuffer.Join(), true, cancellationToken).ConfigureAwait(false);

Source/MQTTnet/Client/MqttClient.cs

@@ -140,6 +140,7 @@ namespace MQTTnet.Client
                 if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
                 {
                     _logger.Warning("Connecting failed: {0}", connectResult.ResultCode);
+                    await DisconnectInternal(null, null, connectResult).ConfigureAwait(false);
                     return connectResult;
                 }
 
@@ -237,6 +238,11 @@ namespace MQTTnet.Client
 
         public async Task PingAsync(CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+            
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
             if (cancellationToken.CanBeCanceled)
             {
                 await Request<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false);
@@ -321,14 +327,14 @@ namespace MQTTnet.Client
                 MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
             }
 
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+            
             if (Options.ValidateFeatures)
             {
                 MqttClientSubscribeOptionsValidator.ThrowIfNotSupported(options, _adapter.PacketFormatterAdapter.ProtocolVersion);
             }
 
-            ThrowIfDisposed();
-            ThrowIfNotConnected();
-
             var subscribePacket = MqttPacketFactories.Subscribe.Create(options);
             subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
 
@@ -440,7 +446,7 @@ namespace MQTTnet.Client
 
                 if (receivedPacket is MqttConnAckPacket connAckPacket)
                 {
-                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
+                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
                 }
                 else if (receivedPacket is MqttAuthPacket)
                 {
@@ -465,11 +471,10 @@ namespace MQTTnet.Client
             // did send a proper ACK packet with a non success response.
             if (options.ThrowOnNonSuccessfulConnectResponse)
             {
-                _logger.Warning(
-                    "Client will now throw an _MqttConnectingFailedException_. This is obsolete and will be removed in the future. Consider setting _ThrowOnNonSuccessfulResponseFromServer=False_ in client options.");
-
                 if (result.ResultCode != MqttClientConnectResultCode.Success)
                 {
+                    _logger.Warning(
+                        "Client will now throw an _MqttConnectingFailedException_. This is obsolete and will be removed in the future. Consider setting _ThrowOnNonSuccessfulResponseFromServer=False_ in client options.");
                     throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
                 }
             }
@@ -513,7 +518,7 @@ namespace MQTTnet.Client
             using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
             {
                 _logger.Verbose("Trying to connect with server '{0}'", Options.ChannelOptions);
-                await _adapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
+                await channelAdapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
                 _logger.Verbose("Connection with server established");
 
                 _publishPacketReceiverQueue?.Dispose();
@@ -703,6 +708,9 @@ namespace MQTTnet.Client
                         await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
                     }
                 }
+                catch (ObjectDisposedException)
+                {
+                }
                 catch (OperationCanceledException)
                 {
                 }
@@ -782,7 +790,7 @@ namespace MQTTnet.Client
         {
             try
             {
-                _logger.Verbose("Start receiving packets.");
+                _logger.Verbose("Start receiving packets");
 
                 while (!cancellationToken.IsCancellationRequested)
                 {
@@ -820,20 +828,22 @@ namespace MQTTnet.Client
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Communication error while receiving packets.");
+                    _logger.Warning(exception, "Communication error while receiving packets");
                 }
                 else
                 {
-                    _logger.Error(exception, "Error while receiving packets.");
+                    _logger.Error(exception, "Error while receiving packets");
                 }
 
-                _packetDispatcher.FailAll(exception);
+                // The packet dispatcher is set to null when the client is being disposed so it may
+                // already being gone!
+                _packetDispatcher?.FailAll(exception);
 
                 await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
             }
             finally
             {
-                _logger.Verbose("Stopped receiving packets.");
+                _logger.Verbose("Stopped receiving packets");
             }
         }
 
@@ -1060,4 +1070,4 @@ namespace MQTTnet.Client
             }
         }
     }
-}
+}

Source/MQTTnet/Client/Options/DefaultMqttCertificatesProvider.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.Collections.Generic;
+using System.Security.Cryptography.X509Certificates;
+
+namespace MQTTnet.Client
+{
+    public sealed class DefaultMqttCertificatesProvider : IMqttClientCertificatesProvider
+    {
+        readonly X509Certificate2Collection _certificates;
+
+        public DefaultMqttCertificatesProvider(X509Certificate2Collection certificates)
+        {
+            _certificates = certificates;
+        }
+
+        public DefaultMqttCertificatesProvider(IEnumerable<X509Certificate> certificates)
+        {
+            if (certificates != null)
+            {
+                _certificates = new X509Certificate2Collection();
+                foreach (var certificate in certificates)
+                {
+                    _certificates.Add(certificate);
+                }
+            }
+        }
+
+        public X509CertificateCollection GetCertificates()
+        {
+            return _certificates;
+        }
+    }
+}

Source/MQTTnet/Client/Options/IMqttClientCertificatesProvider.cs

@@ -0,0 +1,11 @@
+// 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.Client
+{
+    public interface IMqttClientCertificatesProvider
+    {
+        System.Security.Cryptography.X509Certificates.X509CertificateCollection GetCertificates();
+    }
+}

Source/MQTTnet/Client/Options/MqttClientCertificateValidationEventArgs.cs

@@ -14,8 +14,14 @@ namespace MQTTnet.Client
 
         public X509Chain Chain { get; set; }
 
-        public SslPolicyErrors SslPolicyErrors { get; set; }
-
         public IMqttClientChannelOptions ClientOptions { get; set; }
+#if NET452 || NET461 || NET48
+        /// <summary>
+        ///     Can be a host string name or an object derived from WebRequest.
+        /// </summary>
+        public object Sender { get; set; }
+#endif
+
+        public SslPolicyErrors SslPolicyErrors { get; set; }
     }
-}
+}

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -2,6 +2,8 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+#pragma warning disable CS0612 // Type or member is obsolete
+
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -15,10 +17,14 @@ namespace MQTTnet.Client
     public sealed class MqttClientOptionsBuilder
     {
         readonly MqttClientOptions _options = new MqttClientOptions();
-        MqttClientWebSocketProxyOptions _proxyOptions;
+
+        [Obsolete] MqttClientWebSocketProxyOptions _proxyOptions;
 
         MqttClientTcpOptions _tcpOptions;
-        MqttClientOptionsBuilderTlsParameters _tlsParameters;
+        MqttClientTlsOptions _tlsOptions;
+
+        [Obsolete] MqttClientOptionsBuilderTlsParameters _tlsParameters;
+
         MqttClientWebSocketOptions _webSocketOptions;
 
         public MqttClientOptions Build()
@@ -28,11 +34,17 @@ namespace MQTTnet.Client
                 throw new InvalidOperationException("A channel must be set.");
             }
 
+            // The user can specify the TCP options with already configured TLS options
+            // or start with TLS settings not knowing which transport will be used (depending
+            // on the order of called methods from the builder).
+            // The builder prefers the explicitly set TLS options!
+            var tlsOptions = _tlsOptions ?? _tcpOptions?.TlsOptions;
+
             if (_tlsParameters != null)
             {
                 if (_tlsParameters?.UseTls == true)
                 {
-                    var tlsOptions = new MqttClientTlsOptions
+                    tlsOptions = new MqttClientTlsOptions
                     {
                         UseTls = true,
                         SslProtocol = _tlsParameters.SslProtocol,
@@ -40,28 +52,23 @@ namespace MQTTnet.Client
                         CertificateValidationHandler = _tlsParameters.CertificateValidationHandler,
                         IgnoreCertificateChainErrors = _tlsParameters.IgnoreCertificateChainErrors,
                         IgnoreCertificateRevocationErrors = _tlsParameters.IgnoreCertificateRevocationErrors,
-#if WINDOWS_UWP
-                        Certificates = _tlsParameters.Certificates?.Select(c => c.ToArray()).ToList(),
-#else
-                        Certificates = _tlsParameters.Certificates?.ToList(),
-#endif
-
+                        ClientCertificatesProvider = _tlsParameters.CertificatesProvider,
 #if NETCOREAPP3_1_OR_GREATER
                         ApplicationProtocols = _tlsParameters.ApplicationProtocols,
 #endif
                     };
-
-                    if (_tcpOptions != null)
-                    {
-                        _tcpOptions.TlsOptions = tlsOptions;
-                    }
-                    else if (_webSocketOptions != null)
-                    {
-                        _webSocketOptions.TlsOptions = tlsOptions;
-                    }
                 }
             }
 
+            if (_tcpOptions != null)
+            {
+                _tcpOptions.TlsOptions = tlsOptions;
+            }
+            else if (_webSocketOptions != null)
+            {
+                _webSocketOptions.TlsOptions = tlsOptions;
+            }
+
             if (_proxyOptions != null)
             {
                 if (_webSocketOptions == null)
@@ -78,17 +85,7 @@ namespace MQTTnet.Client
 
             return _options;
         }
-        
-        /// <summary>
-        /// The client will not throw an exception when the MQTT server responses with a non success ACK packet.
-        /// This will become the default behavior in future versions of the library.
-        /// </summary>
-        public MqttClientOptionsBuilder WithoutThrowOnNonSuccessfulConnectResponse()
-        {
-            _options.ThrowOnNonSuccessfulConnectResponse = false;
-            return this;
-        }
-        
+
         public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data)
         {
             _options.AuthenticationMethod = method;
@@ -120,6 +117,7 @@ namespace MQTTnet.Client
             return this;
         }
 
+        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
         {
             if (uri == null)
@@ -136,12 +134,16 @@ namespace MQTTnet.Client
                     break;
 
                 case "mqtts":
-                    WithTcpServer(uri.Host, port).WithTls();
+                    WithTcpServer(uri.Host, port)
+                        .WithTlsOptions(
+                            o =>
+                            {
+                            });
                     break;
 
                 case "ws":
                 case "wss":
-                    WithWebSocketServer(uri.ToString());
+                    WithWebSocketServer(o => o.WithUri(uri.ToString()));
                     break;
 
                 default:
@@ -159,6 +161,7 @@ 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));
@@ -222,6 +225,16 @@ namespace MQTTnet.Client
             return this;
         }
 
+        /// <summary>
+        ///     The client will not throw an exception when the MQTT server responses with a non success ACK packet.
+        ///     This will become the default behavior in future versions of the library.
+        /// </summary>
+        public MqttClientOptionsBuilder WithoutThrowOnNonSuccessfulConnectResponse()
+        {
+            _options.ThrowOnNonSuccessfulConnectResponse = false;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
         {
             if (value == MqttProtocolVersion.Unknown)
@@ -233,6 +246,7 @@ namespace MQTTnet.Client
             return this;
         }
 
+        [Obsolete("Use WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithProxy(
             string address,
             string username = null,
@@ -254,6 +268,7 @@ namespace MQTTnet.Client
             return this;
         }
 
+        [Obsolete("Use WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithProxy(Action<MqttClientWebSocketProxyOptions> optionsBuilder)
         {
             if (optionsBuilder == null)
@@ -324,17 +339,20 @@ namespace MQTTnet.Client
             return this;
         }
 
+        [Obsolete("Use WithTlsOptions(... configure) instead.")]
         public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters)
         {
             _tlsParameters = parameters;
             return this;
         }
 
+        [Obsolete("Use WithTlsOptions(... configure) instead.")]
         public MqttClientOptionsBuilder WithTls()
         {
             return WithTls(new MqttClientOptionsBuilderTlsParameters { UseTls = true });
         }
 
+        [Obsolete("Use WithTlsOptions(... configure) instead.")]
         public MqttClientOptionsBuilder WithTls(Action<MqttClientOptionsBuilderTlsParameters> optionsBuilder)
         {
             if (optionsBuilder == null)
@@ -351,6 +369,26 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithTlsOptions(MqttClientTlsOptions tlsOptions)
+        {
+            _tlsOptions = tlsOptions;
+            return this;
+        }
+
+        public MqttClientOptionsBuilder WithTlsOptions(Action<MqttClientTlsOptionsBuilder> configure)
+        {
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
+
+            var builder = new MqttClientTlsOptionsBuilder();
+            configure.Invoke(builder);
+
+            _tlsOptions = builder.Build();
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithTopicAliasMaximum(ushort topicAliasMaximum)
         {
             _options.TopicAliasMaximum = topicAliasMaximum;
@@ -382,6 +420,7 @@ namespace MQTTnet.Client
             return this;
         }
 
+        [Obsolete("Use WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithWebSocketServer(string uri, MqttClientOptionsBuilderWebSocketParameters parameters = null)
         {
             _webSocketOptions = new MqttClientWebSocketOptions
@@ -394,6 +433,21 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithWebSocketServer(Action<MqttClientWebSocketOptionsBuilder> configure)
+        {
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
+
+            var webSocketOptionsBuilder = new MqttClientWebSocketOptionsBuilder();
+            configure.Invoke(webSocketOptionsBuilder);
+
+            _webSocketOptions = webSocketOptionsBuilder.Build();
+            return this;
+        }
+
+        [Obsolete("Use WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithWebSocketServer(Action<MqttClientWebSocketOptions> optionsBuilder)
         {
             if (optionsBuilder == null)
@@ -425,6 +479,12 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithWillMessageExpiryInterval(uint willMessageExpiryInterval)
+        {
+            _options.WillMessageExpiryInterval = willMessageExpiryInterval;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithWillPayload(byte[] willPayload)
         {
             _options.WillPayload = willPayload;

Source/MQTTnet/Client/Options/MqttClientOptionsBuilderTlsParameters.cs

@@ -5,12 +5,14 @@
 using System;
 using System.Collections.Generic;
 using System.Security.Authentication;
-using System.Security.Cryptography.X509Certificates;
 
 namespace MQTTnet.Client
 {
+    [Obsolete("Use methods from MqttClientOptionsBuilder instead.")]
     public sealed class MqttClientOptionsBuilderTlsParameters
     {
+        IEnumerable<System.Security.Cryptography.X509Certificates.X509Certificate> _obsoleteCertificates;
+
         public bool UseTls { get; set; }
 
         public Func<MqttClientCertificateValidationEventArgs, bool> CertificateValidationHandler { get; set; }
@@ -24,7 +26,24 @@ namespace MQTTnet.Client
 #if WINDOWS_UWP
         public IEnumerable<IEnumerable<byte>> Certificates { get; set; }
 #else
-        public IEnumerable<X509Certificate> Certificates { get; set; }
+        [Obsolete("Use CertificatesProvider instead.")]
+        public IEnumerable<System.Security.Cryptography.X509Certificates.X509Certificate> Certificates
+        {
+            get => _obsoleteCertificates;
+            set
+            {
+                _obsoleteCertificates = value;
+
+                if (value == null)
+                {
+                    CertificatesProvider = null;
+                }
+                else
+                {
+                    CertificatesProvider = new DefaultMqttCertificatesProvider(value);
+                }
+            }
+        }
 #endif
 
 #if NETCOREAPP3_1_OR_GREATER
@@ -36,5 +55,7 @@ namespace MQTTnet.Client
         public bool IgnoreCertificateChainErrors { get; set; }
 
         public bool IgnoreCertificateRevocationErrors { get; set; }
+
+        public IMqttClientCertificatesProvider CertificatesProvider { get; set; }
     }
 }

Source/MQTTnet/Client/Options/MqttClientOptionsBuilderWebSocketParameters.cs

@@ -2,11 +2,13 @@
 // 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.Net;
 
 namespace MQTTnet.Client
 {
+    [Obsolete("Use dedicated methods in MqttClientOptionsBuilder.")]
     public class MqttClientOptionsBuilderWebSocketParameters
     {
         public IDictionary<string, string> RequestHeaders { get; set; }

Source/MQTTnet/Client/Options/MqttClientTcpOptionsExtensions.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Client
                 return options.Port.Value;
             }
 
-            return !options.TlsOptions.UseTls ? 1883 : 8883;
+            return !(options.TlsOptions?.UseTls ?? false) ? 1883 : 8883;
         }
     }
 }

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -24,11 +24,13 @@ namespace MQTTnet.Client
 
         public X509RevocationMode RevocationMode { get; set; } = X509RevocationMode.Online;
 
-#if WINDOWS_UWP
-        public List<byte[]> Certificates { get; set; }
-#else
-        public List<X509Certificate> Certificates { get; set; }
-#endif
+        /// <summary>
+        ///     Gets or sets the provider for certificates.
+        ///     This provider gets called whenever the client wants to connect
+        ///     with the server and requires certificates for authentication.
+        ///     The implementation may return different certificates each time.
+        /// </summary>
+        public IMqttClientCertificatesProvider ClientCertificatesProvider { get; set; }
 
 #if NETCOREAPP3_1_OR_GREATER
         public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
@@ -48,8 +50,13 @@ namespace MQTTnet.Client
 
 #if NET48 || NETCOREAPP3_1_OR_GREATER
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
+
 #else
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | (SslProtocols)0x00003000 /*Tls13*/;
 #endif
+
+#if NET7_0_OR_GREATER
+        public X509Certificate2Collection TrustChain { get; set; }
+#endif
     }
 }

Source/MQTTnet/Client/Options/MqttClientTlsOptionsBuilder.cs

@@ -0,0 +1,145 @@
+// 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.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
+#if NETCOREAPP3_1_OR_GREATER
+using System.Net.Security;
+#endif
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientTlsOptionsBuilder
+    {
+        readonly MqttClientTlsOptions _tlsOptions = new MqttClientTlsOptions
+        {
+            // If someone used this builder the change is very very high that TLS
+            // should be actually used.
+            UseTls = true
+        };
+
+        public MqttClientTlsOptions Build()
+        {
+            return _tlsOptions;
+        }
+
+        public MqttClientTlsOptionsBuilder UseTls(bool useTls = true)
+        {
+            _tlsOptions.UseTls = useTls;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithAllowUntrustedCertificates(bool allowUntrustedCertificates = true)
+        {
+            _tlsOptions.AllowUntrustedCertificates = allowUntrustedCertificates;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithCertificateValidationHandler(Func<MqttClientCertificateValidationEventArgs, bool> certificateValidationHandler)
+        {
+            if (certificateValidationHandler == null)
+            {
+                throw new ArgumentNullException(nameof(certificateValidationHandler));
+            }
+
+            _tlsOptions.CertificateValidationHandler = certificateValidationHandler;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithClientCertificates(IEnumerable<X509Certificate2> certificates)
+        {
+            if (certificates == null)
+            {
+                throw new ArgumentNullException(nameof(certificates));
+            }
+
+            _tlsOptions.ClientCertificatesProvider = new DefaultMqttCertificatesProvider(certificates);
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithClientCertificates(X509Certificate2Collection certificates)
+        {
+            if (certificates == null)
+            {
+                throw new ArgumentNullException(nameof(certificates));
+            }
+
+            _tlsOptions.ClientCertificatesProvider = new DefaultMqttCertificatesProvider(certificates);
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithClientCertificatesProvider(IMqttClientCertificatesProvider clientCertificatesProvider)
+        {
+            _tlsOptions.ClientCertificatesProvider = clientCertificatesProvider;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithIgnoreCertificateChainErrors(bool ignoreCertificateChainErrors = true)
+        {
+            _tlsOptions.IgnoreCertificateChainErrors = ignoreCertificateChainErrors;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithIgnoreCertificateRevocationErrors(bool ignoreCertificateRevocationErrors = true)
+        {
+            _tlsOptions.IgnoreCertificateRevocationErrors = ignoreCertificateRevocationErrors;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithRevocationMode(X509RevocationMode revocationMode)
+        {
+            _tlsOptions.RevocationMode = revocationMode;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithSslProtocols(SslProtocols sslProtocols)
+        {
+            _tlsOptions.SslProtocol = sslProtocols;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithTargetHost(string targetHost)
+        {
+            _tlsOptions.TargetHost = targetHost;
+            return this;
+        }
+
+#if NETCOREAPP3_1_OR_GREATER
+        public MqttClientTlsOptionsBuilder WithAllowRenegotiation(bool allowRenegotiation = true)
+        {
+            _tlsOptions.AllowRenegotiation = allowRenegotiation;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithApplicationProtocols(List<SslApplicationProtocol> applicationProtocols)
+        {
+            _tlsOptions.ApplicationProtocols = applicationProtocols;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithCipherSuitesPolicy(CipherSuitesPolicy cipherSuitePolicy)
+        {
+            _tlsOptions.CipherSuitesPolicy = cipherSuitePolicy;
+            return this;
+        }
+
+        public MqttClientTlsOptionsBuilder WithCipherSuitesPolicy(EncryptionPolicy encryptionPolicy)
+        {
+            _tlsOptions.EncryptionPolicy = encryptionPolicy;
+            return this;
+        }
+#endif
+
+#if NET7_0_OR_GREATER
+        public MqttClientTlsOptionsBuilder WithTrustChain(X509Certificate2Collection chain)
+        {
+            _tlsOptions.TrustChain = chain;
+            return this;
+        }
+
+#endif
+    }
+}

Source/MQTTnet/Client/Options/MqttClientWebSocketOptionsBuilder.cs

@@ -0,0 +1,85 @@
+// 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.Net;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientWebSocketOptionsBuilder
+    {
+        readonly MqttClientWebSocketOptions _webSocketOptions = new MqttClientWebSocketOptions();
+
+        public MqttClientWebSocketOptions Build()
+        {
+            return _webSocketOptions;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithCookieContainer(CookieContainer cookieContainer)
+        {
+            _webSocketOptions.CookieContainer = cookieContainer;
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithCookieContainer(ICredentials credentials)
+        {
+            _webSocketOptions.Credentials = credentials;
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithProxyOptions(MqttClientWebSocketProxyOptions proxyOptions)
+        {
+            _webSocketOptions.ProxyOptions = proxyOptions;
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithProxyOptions(Action<MqttClientWebSocketProxyOptionsBuilder> configure)
+        {
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
+
+            var proxyOptionsBuilder = new MqttClientWebSocketProxyOptionsBuilder();
+            configure.Invoke(proxyOptionsBuilder);
+
+            _webSocketOptions.ProxyOptions = proxyOptionsBuilder.Build();
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithRequestHeaders(IDictionary<string, string> requestHeaders)
+        {
+            _webSocketOptions.RequestHeaders = requestHeaders;
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithSubProtocols(ICollection<string> subProtocols)
+        {
+            _webSocketOptions.SubProtocols = subProtocols;
+            return this;
+        }
+
+        public MqttClientWebSocketOptionsBuilder WithUri(string uri)
+        {
+            _webSocketOptions.Uri = uri;
+            return this;
+        }
+
+#if !NETSTANDARD1_3
+        public MqttClientWebSocketOptionsBuilder WithKeepAliveInterval(TimeSpan keepAliveInterval)
+        {
+            _webSocketOptions.KeepAliveInterval = keepAliveInterval;
+            return this;
+        }
+#endif
+#if !WINDOWS_UWP && !NETSTANDARD1_3
+        public MqttClientWebSocketOptionsBuilder WithUseDefaultCredentials(bool useDefaultCredentials = true)
+        {
+            _webSocketOptions.UseDefaultCredentials = useDefaultCredentials;
+            return this;
+        }
+#endif
+    }
+}

Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptionsBuilder.cs

@@ -0,0 +1,67 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using System.Linq;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientWebSocketProxyOptionsBuilder
+    {
+        readonly MqttClientWebSocketProxyOptions _proxyOptions = new MqttClientWebSocketProxyOptions();
+
+        public MqttClientWebSocketProxyOptionsBuilder WithAddress(string address)
+        {
+            _proxyOptions.Address = address;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithUsername(string username)
+        {
+            _proxyOptions.Username = username;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithPassword(string password)
+        {
+            _proxyOptions.Password = password;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithDomain(string domain)
+        {
+            _proxyOptions.Domain = domain;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithBypassOnLocal(bool bypassOnLocal = true)
+        {
+            _proxyOptions.BypassOnLocal = bypassOnLocal;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithBypassList(string[] bypassList)
+        {
+            _proxyOptions.BypassList = bypassList;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithBypassList(IEnumerable<string> bypassList)
+        {
+            _proxyOptions.BypassList = bypassList?.ToArray();
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptionsBuilder WithUseDefaultCredentials(bool useDefaultCredentials = true)
+        {
+            _proxyOptions.UseDefaultCredentials = useDefaultCredentials;
+            return this;
+        }
+        
+        public MqttClientWebSocketProxyOptions Build()
+        {
+            return _proxyOptions;
+        }
+    }
+}

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -6,8 +6,6 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
-using System.Linq;
-using System.Runtime.InteropServices.WindowsRuntime;
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
@@ -18,6 +16,7 @@ using Windows.Security.Cryptography.Certificates;
 using MQTTnet.Channel;
 using MQTTnet.Client;
 using MQTTnet.Server;
+using System.Runtime.InteropServices.WindowsRuntime;
 
 namespace MQTTnet.Implementations
 {
@@ -126,17 +125,19 @@ namespace MQTTnet.Implementations
 
         private static Certificate LoadCertificate(IMqttClientChannelOptions options)
         {
-            if (options.TlsOptions.Certificates == null || !options.TlsOptions.Certificates.Any())
+            var certificates = options.TlsOptions?.ClientCertificatesProvider?.GetCertificates();
+
+            if (certificates == null || certificates.Count == 0)
             {
                 return null;
             }
 
-            if (options.TlsOptions.Certificates.Count > 1)
+            if (certificates.Count > 1)
             {
                 throw new NotSupportedException("Only one client certificate is supported when using 'uap10.0'.");
             }
 
-            return new Certificate(options.TlsOptions.Certificates.First().AsBuffer());
+            return new Certificate(certificates[0].Export(X509ContentType.Cert).AsBuffer());
         }
 
         private IEnumerable<ChainValidationResult> ResolveIgnorableServerCertificateErrors()

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -3,9 +3,6 @@
 // See the LICENSE file in the project root for more information.
 
 #if !WINDOWS_UWP
-using MQTTnet.Channel;
-using MQTTnet.Client;
-using MQTTnet.Exceptions;
 using System;
 using System.IO;
 using System.Net.Security;
@@ -14,6 +11,9 @@ using System.Runtime.ExceptionServices;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Channel;
+using MQTTnet.Client;
+using MQTTnet.Exceptions;
 using MQTTnet.Internal;
 
 namespace MQTTnet.Implementations
@@ -73,7 +73,7 @@ namespace MQTTnet.Implementations
                 {
                     socket.Bind(_tcpOptions.LocalEndpoint);
                 }
-                
+
                 socket.ReceiveBufferSize = _tcpOptions.BufferSize;
                 socket.SendBufferSize = _tcpOptions.BufferSize;
                 socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds;
@@ -105,7 +105,7 @@ namespace MQTTnet.Implementations
                     {
                         targetHost = _tcpOptions.Server;
                     }
-                    
+
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
@@ -115,14 +115,31 @@ namespace MQTTnet.Implementations
                             ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
                             ClientCertificates = LoadCertificates(),
                             EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
-                            CertificateRevocationCheckMode =
- _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
+                            CertificateRevocationCheckMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
+                                ? X509RevocationMode.NoCheck
+                                : _tcpOptions.TlsOptions.RevocationMode,
                             TargetHost = targetHost,
                             CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
                             EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
                             AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
                         };
 
+#if NET7_0_OR_GREATER
+                        if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
+                        {
+                            sslOptions.CertificateChainPolicy = new X509ChainPolicy
+                            {
+                                TrustMode = X509ChainTrustMode.CustomRootTrust,
+                                VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
+                                RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors
+                                    ? X509RevocationMode.NoCheck
+                                    : _tcpOptions.TlsOptions.RevocationMode
+                            };
+
+                            sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
+                        }
+#endif
+
                         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
 #else
                         await sslStream.AuthenticateAsClientAsync(
@@ -298,18 +315,7 @@ namespace MQTTnet.Implementations
 
         X509CertificateCollection LoadCertificates()
         {
-            if (_tcpOptions.TlsOptions.Certificates == null)
-            {
-                return null;
-            }
-
-            var certificates = new X509CertificateCollection();
-            foreach (var certificate in _tcpOptions.TlsOptions.Certificates)
-            {
-                certificates.Add(certificate);
-            }
-
-            return certificates;
+            return _tcpOptions.TlsOptions.ClientCertificatesProvider?.GetCertificates();
         }
     }
 }

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -46,13 +46,13 @@ namespace MQTTnet.Implementations
             var uri = _options.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
             {
-                if (_options.TlsOptions?.UseTls == false)
+                if (_options.TlsOptions?.UseTls == true)
                 {
-                    uri = "ws://" + uri;
+                    uri = "wss://" + uri;
                 }
                 else
                 {
-                    uri = "wss://" + uri;
+                    uri = "ws://" + uri;
                 }
             }
 
@@ -206,16 +206,12 @@ namespace MQTTnet.Implementations
                 clientWebSocket.Options.Cookies = _options.CookieContainer;
             }
 
-            if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null)
+            if (_options.TlsOptions?.UseTls == true)
             {
-                clientWebSocket.Options.ClientCertificates = new X509CertificateCollection();
-                foreach (var certificate in _options.TlsOptions.Certificates)
+                var certificates = _options.TlsOptions?.ClientCertificatesProvider?.GetCertificates();
+                if (certificates?.Count > 0)
                 {
-#if WINDOWS_UWP
-                    clientWebSocket.Options.ClientCertificates.Add(new X509Certificate(certificate));
-#else
-                    clientWebSocket.Options.ClientCertificates.Add(certificate);
-#endif
+                    clientWebSocket.Options.ClientCertificates = certificates;
                 }
             }
 
@@ -248,12 +244,20 @@ namespace MQTTnet.Implementations
                 throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard2.0'.");
 #elif WINDOWS_UWP
                 throw new NotSupportedException("Remote certificate validation callback is not supported when using 'uap10.0'.");
-#elif NET452
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
-#elif NET461
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'.");
-#elif NET48
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net48'.");
+#elif NET452 || NET461 || NET48
+                ServicePointManager.ServerCertificateValidationCallback += (sender, certificate, chain, sslPolicyErrors) => 
+                {
+                    var context = new MqttClientCertificateValidationEventArgs
+                    {
+                        Sender = sender,
+                        Certificate = certificate,
+                        Chain = chain,
+                        SslPolicyErrors = sslPolicyErrors,
+                        ClientOptions = _options
+                    };
+
+                    return certificateValidationHandler(context);
+                };
 #else
                 clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
                 {
@@ -272,4 +276,4 @@ namespace MQTTnet.Implementations
             }
         }
     }
-}
+}

Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs

@@ -19,7 +19,6 @@ namespace MQTTnet.LowLevelClient
         readonly IMqttClientAdapterFactory _clientAdapterFactory;
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
         readonly MqttNetSourceLogger _logger;
-
         readonly IMqttNetLogger _rootLogger;
 
         IMqttChannelAdapter _adapter;

Source/MQTTnet/Server/MqttServer.cs

@@ -362,7 +362,7 @@ namespace MQTTnet.Server
             ThrowIfDisposed();
             ThrowIfNotStarted();
 
-            return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
+            return _retainedMessagesManager?.UpdateMessage(string.Empty, retainedMessage);
         }
 
         protected override void Dispose(bool disposing)
@@ -405,4 +405,4 @@ namespace MQTTnet.Server
             }
         }
     }
-}
+}