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,11 +1,3 @@
-* [Client] Fixed _PlatformNotSupportedException_ when using Blazor (#1755, thanks to @Nickztar).
-* [Client] Added hot reload of client certificates (#1781).
-* [Client] Added several new option builders and aligned usage (#1781, BREAKING CHANGE!).
-* [Client] Added support for _RemoteCertificateValidationCallback_ for .NET 4.5.2, 4.6.1 and 4.8 (#1806, thanks to @troky).
-* [Client] Fixed wrong logging of obsolete feature when connection was not successful (#1801, thanks to @ramonsmits).
-* [Client] Fixed _NullReferenceException_ when performing several actions when not connected (#1800, thanks to @ramonsmits).
-* [RpcClient] Added support for passing custom parameters to topic generation context (#1798, thanks to @Temppus).
-* [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 an issue in connection handling for ASP.NET connections (#1819, thanks to @CZEMacLeod).
+* [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.3.0.${{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;
@@ -430,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

@@ -184,7 +184,7 @@ public static class Client_Subscribe_Samples
          * 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 += async ea =>
+        mqttClient.ApplicationMessageReceivedAsync += ea =>
         {
             ea.AutoAcknowledge = false;
 
@@ -201,6 +201,8 @@ public static class Client_Subscribe_Samples
             }
 
             _ = Task.Run(ProcessAsync, shutdownToken);
+
+            return Task.CompletedTask;
         };
     }
 

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.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;
                 }
             }
             

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -25,7 +25,9 @@ namespace MQTTnet.TestApp
                 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.
@@ -34,7 +36,9 @@ namespace MQTTnet.TestApp
                 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
@@ -97,10 +101,10 @@ namespace MQTTnet.TestApp
                 "test.mosquitto.org WS TLS12",
                 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(
@@ -150,7 +154,6 @@ namespace MQTTnet.TestApp
                 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", PayloadSegment = new ArraySegment<byte>( 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;
 

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()
         {

Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs

@@ -52,7 +52,7 @@ namespace MQTTnet.Tests.Extensions
             var paramValue = "123";
             var parameters = new Dictionary<string, object>
             {
-                { TestParametersTopicGenerationStrategy.ExpectedParamName, "123" },
+                { TestParametersTopicGenerationStrategy.ExpectedParamName, "123" }
             };
 
             using (var testEnvironment = CreateTestEnvironment())
@@ -164,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);
                 }
             }
         }

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/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

@@ -1,4 +1,4 @@
-#if !(NET452 || NET461)
+#if !(NET452 || NET461 || NET48)
 using System;
 using System.Collections.Concurrent;
 using System.Diagnostics;
@@ -149,18 +149,14 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        class ClientTestHarness : IDisposable
+        sealed class ClientTestHarness : IDisposable
         {
             readonly HotSwappableClientCertProvider _hotSwapClient = new HotSwappableClientCertProvider();
+
             IMqttClient _client;
 
             public string ClientId => _client.Options.ClientId;
 
-            public void ClearServerCerts()
-            {
-                _hotSwapClient.ClearServerCerts();
-            }
-
             public Task Connect()
             {
                 return Run_Client_Connection();
@@ -169,6 +165,7 @@ namespace MQTTnet.Tests.Server
             public void Dispose()
             {
                 _client.Dispose();
+                _hotSwapClient.Dispose();
             }
 
             public X509Certificate2 GetCurrentClientCert()
@@ -189,6 +186,8 @@ namespace MQTTnet.Tests.Server
 
             public void WaitForConnectOrFail(TimeSpan timeout)
             {
+                Thread.Sleep(100);
+
                 if (!_client.IsConnected)
                 {
                     _client.ReconnectAsync().Wait(timeout);
@@ -210,12 +209,12 @@ namespace MQTTnet.Tests.Server
                 Assert.IsFalse(_client.IsConnected, "Client connection success but test wanted fail");
             }
 
-            public void WaitForDisconnect(TimeSpan timeout)
+            void WaitForDisconnect(TimeSpan timeout)
             {
                 var timer = Stopwatch.StartNew();
                 while ((_client == null || _client.IsConnected) && timer.Elapsed < timeout)
                 {
-                    Thread.Sleep(5);
+                    Thread.Sleep(100);
                 }
             }
 
@@ -252,22 +251,17 @@ namespace MQTTnet.Tests.Server
                 var timer = Stopwatch.StartNew();
                 while ((_client == null || !_client.IsConnected) && timer.Elapsed < timeout)
                 {
-                    Thread.Sleep(5);
+                    Thread.Sleep(100);
                 }
             }
         }
 
-        class ServerTestHarness : IDisposable
+        sealed class ServerTestHarness : IDisposable
         {
             readonly HotSwappableServerCertProvider _hotSwapServer = new HotSwappableServerCertProvider();
 
             MqttServer _server;
 
-            public void ClearClientCerts()
-            {
-                _hotSwapServer.ClearClientCerts();
-            }
-
             public void Dispose()
             {
                 if (_server != null)
@@ -276,15 +270,12 @@ namespace MQTTnet.Tests.Server
                     _server.Dispose();
                 }
 
-                if (_hotSwapServer != null)
-                {
-                    _hotSwapServer.Dispose();
-                }
+                _hotSwapServer?.Dispose();
             }
 
-            public async Task ForceDisconnectAsync(ClientTestHarness client)
+            public Task ForceDisconnectAsync(ClientTestHarness client)
             {
-                await _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
+                return _server.DisconnectClientAsync(client.ClientId, MqttDisconnectReasonCode.UnspecifiedError);
             }
 
             public X509Certificate2 GetCurrentServerCert()
@@ -302,7 +293,7 @@ namespace MQTTnet.Tests.Server
                 _hotSwapServer.InstallNewClientCert(serverCert);
             }
 
-            public async Task StartServer()
+            public Task StartServer()
             {
                 var mqttFactory = new MqttFactory();
 
@@ -312,24 +303,33 @@ namespace MQTTnet.Tests.Server
                     .Build();
 
                 mqttServerOptions.TlsEndpointOptions.ClientCertificateRequired = true;
+
                 _server = mqttFactory.CreateMqttServer(mqttServerOptions);
-                await _server.StartAsync();
+                return _server.StartAsync();
             }
         }
 
-        class HotSwappableClientCertProvider : IMqttClientCertificatesProvider
+        class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable
         {
             X509Certificate2Collection _certificates;
-            ConcurrentBag<X509Certificate2> ServerCerts = new ConcurrentBag<X509Certificate2>();
+            ConcurrentBag<X509Certificate2> _serverCerts = new ConcurrentBag<X509Certificate2>();
 
             public HotSwappableClientCertProvider()
             {
                 _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
             }
-
-            public void ClearServerCerts()
+            
+            public void Dispose()
             {
-                ServerCerts = new ConcurrentBag<X509Certificate2>();
+                if (_certificates != null)
+                {
+                    foreach (var certs in _certificates)
+                    {
+#if !NET452
+                        certs.Dispose();
+#endif
+                    }
+                }
             }
 
             public X509CertificateCollection GetCertificates()
@@ -339,18 +339,17 @@ namespace MQTTnet.Tests.Server
 
             public void HotSwapCert()
             {
-                var newCert = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
-                var oldCerts = Interlocked.Exchange(ref _certificates, newCert);
+                _certificates = new X509Certificate2Collection(CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.2"));
             }
 
             public void InstallNewServerCert(X509Certificate2 serverCert)
             {
-                ServerCerts.Add(serverCert);
+                _serverCerts.Add(serverCert);
             }
 
             public bool OnCertificateValidation(MqttClientCertificateValidationEventArgs certContext)
             {
-                var serverCerts = ServerCerts.ToArray();
+                var serverCerts = _serverCerts.ToArray();
 
                 var providedCert = certContext.Certificate.GetRawCertData();
                 for (int i = 0, n = serverCerts.Length; i < n; i++)
@@ -365,36 +364,18 @@ namespace MQTTnet.Tests.Server
 
                 return false;
             }
-
-            void Dispose()
-            {
-                if (_certificates != null)
-                {
-                    foreach (var certs in _certificates)
-                    {
-#if !NET452
-                        certs.Dispose();
-#endif
-                    }
-                }
-            }
         }
 
-        class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
+        sealed class HotSwappableServerCertProvider : ICertificateProvider, IDisposable
         {
+            readonly ConcurrentBag<X509Certificate2> _clientCerts = new ConcurrentBag<X509Certificate2>();
             X509Certificate2 _certificate;
-            ConcurrentBag<X509Certificate2> ClientCerts = new ConcurrentBag<X509Certificate2>();
 
             public HotSwappableServerCertProvider()
             {
                 _certificate = CreateSelfSignedCertificate("1.3.6.1.5.5.7.3.1");
             }
 
-            public void ClearClientCerts()
-            {
-                ClientCerts = new ConcurrentBag<X509Certificate2>();
-            }
-
             public void Dispose()
             {
 #if !NET452
@@ -418,12 +399,12 @@ namespace MQTTnet.Tests.Server
 
             public void InstallNewClientCert(X509Certificate2 certificate)
             {
-                ClientCerts.Add(certificate);
+                _clientCerts.Add(certificate);
             }
 
             public bool RemoteCertificateValidationCallback(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
             {
-                var serverCerts = ClientCerts.ToArray();
+                var serverCerts = _clientCerts.ToArray();
 
                 var providedCert = certificate.GetRawCertData();
                 for (int i = 0, n = serverCerts.Length; i < n; i++)

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

@@ -238,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);
@@ -441,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)
                 {
@@ -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");
             }
         }
 

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;
@@ -32,7 +34,12 @@ namespace MQTTnet.Client
                 throw new InvalidOperationException("A channel must be set.");
             }
 
-            var tlsOptions = _tlsOptions;
+            // 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)

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -50,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

@@ -131,6 +131,15 @@ namespace MQTTnet.Client
             _tlsOptions.EncryptionPolicy = encryptionPolicy;
             return this;
         }
+#endif
+
+#if NET7_0_OR_GREATER
+        public MqttClientTlsOptionsBuilder WithTrustChain(X509Certificate2Collection chain)
+        {
+            _tlsOptions.TrustChain = chain;
+            return this;
+        }
+
 #endif
     }
 }

Source/MQTTnet/Implementations/MqttTcpChannel.cs

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

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

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
             }
         }
     }
-}
+}