Refactor code.

commit 0cbf50427ca6fb343cfa22ccbcd06aad5ea2b4c2
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Mon Jun 20 21:02:50 2022 +0200

    Fix firing of _ClientSubscribedTopicAsync_ event.

.github/ISSUE_TEMPLATE/bug_report.md

@@ -24,7 +24,7 @@ A clear and concise description of what the bug is.
 <!-- Remove the items which don't apply from the following list. -->
 - Client
 - ManagedClient
-- RcpClient
+- RpcClient
 - Server
 
 ### To Reproduce

.github/ISSUE_TEMPLATE/custom.md

@@ -14,6 +14,5 @@ A clear and concise description of what you want to know.
 <!-- Remove the items which don't apply from the following list -->
 - Client
 - ManagedClient
-- MQTTnet.Server standalone
 - Server
 - Generic

.github/ISSUE_TEMPLATE/feature_request.md

@@ -15,7 +15,6 @@ A clear and concise description of what the problem is. Example. I'm am trying t
 <!-- Remove the items which don't apply from the following list -->
 - Client
 - ManagedClient
-- MQTTnet.Server standalone
 - Server
 - Generic
 

.github/workflows/ReleaseNotes.md

@@ -1,32 +1,8 @@
-We have joined the .NET Foundation!
+* [Core] Updated nuget packages.
-
+* [Core] The option _IgnoreCertificateChainErrors_ is now respected (thanks to @GodVenn, #1447).
-Version 4 comes with a new API so a lot of breaking changes should be expected.
+* [ManagedClient] The managed client now sends the entire topic filter including new MQTTv5 properties when subscribing.
-Checkout the upgrade guide (https://github.com/dotnet/MQTTnet/wiki/Upgrading-guide) for an overview of the changes.
+* [ManagedClient] Fixed wrong firing of _ApplicationMessageSkippedAsync_ event (thanks to @quackgizmo, #1460).
-Checkout the new samples (https://github.com/dotnet/MQTTnet/tree/feature/master/Samples) how to use the new API. The wiki only remains for version 3 of this library.
+* [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442).
-
+* [Server] Fix cross thread issue in session message storage for QoS 1 and 2.
-Preview builds of this library are available at: https://www.myget.org/feed/mqttnet/package/nuget/MQTTnet
+* [Server] The event _ClientSubscribedTopicAsync_ is now fired after the subscription is completely processed internally (#1435).
-
+* [Server] Improved CPU usage on lower end machines (#788).
-* [Core] Improved memory management when working with large payloads.
-* [Core] Added support for .NET 6.0.
-* [Core] nuget packages are now created by MSBuild including more information (i.e. commit hash).
-* [Client] Exposed socket linger state in options.
-* [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271).
-* [Client] Added support for _ServerKeepAlive_ (MQTTv5).
-* [Client] Exposed user properties and reason string in subscribe result.
-* [Client] Exposed user properties and reason string in unsubscribe result.
-* [Client] Migrated application message handler to a regular .NET event (BREAKING CHANGE!).
-* [Client] The will message is longer a regular application message due to not supported properties by the will message (BREAKING CHANGE!).
-* [Client] Timeouts are no longer handled inside the library. Each method (Connect, Publish etc.) supports a cancellation token so that custom timeouts can and must be used (BREAKING CHANGE!).
-* [Client] Exposed certificate revocation mode on options (@andyolivares).
-* [Server] Exposed socket linger state in options.
-* [Server] Added support for returning individual subscription errors (#80 thanks to @jimch)
-* [Server] Improved topic filter comparisons (support for $).
-* [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!).
-* [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud).
-* [Server] All interceptors and events are migrated from interfaces to simple events. All existing APIs are availble but must be migrated to corresponding events (BREAKING CHANGE!).
-* [Server] Removed all interceptor and event interfaces including the delegate implementations etc. (BREAKING CHANGE!).
-* [Server] Renamed a lot of classes and adjusted namespaces (BREAKING CHANGE!).
-* [Server] Introduced a new queueing approach for internal message process (packet bus).
-* [Server] For security reasons the default endpoint (1883) is no longer enabled by default (BREAKING CHANGE!).
-* [Server] Added support for choosing the cipher suite (thanks to @TimSiefert).
-* [Nuget] Added method summaries etc. to nuget packages (thanks to @SpringHgui).

.github/workflows/ci.yml

@@ -3,8 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  ASSEMBLY_VERSION: "4.0.0.${{github.run_number}}"
+  VERSION: "4.0.2.${{github.run_number}}"
-  NUGET_VERSION:  "4.0.0.${{github.run_number}}"
 
 jobs:
   build:
@@ -40,7 +39,7 @@ jobs:
         run: msbuild MQTTnet.sln /t:Restore /p:Configuration="Release" /verbosity:m
 
       - name: Build solution
-        run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.ASSEMBLY_VERSION }} /p:AssemblyVersion=${{ env.ASSEMBLY_VERSION }} /p:PackageVersion=${{ env.NUGET_VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk
+        run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk
 
       - name: Collect nuget Packages
         uses: actions/upload-artifact@v2

Samples/Client/Client_Connection_Samples.cs

@@ -111,13 +111,21 @@ public static class Client_Connection_Samples
                 .WithTls(
                     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;
+                        
+                        // The default value is determined by the OS. Set manually to force version.
                         o.SslProtocol = SslProtocols.Tls12;
                     })
                 .Build();
 
-            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+            using (var timeout = new CancellationTokenSource(5000))
+            {
+                await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
 
-            Console.WriteLine("The MQTT client is connected.");
+                Console.WriteLine("The MQTT client is connected.");
+            }
         }
     }
 
@@ -159,19 +167,59 @@ public static class Client_Connection_Samples
                 .WithTls(
                     o =>
                     {
-                        o.SslProtocol = SslProtocols.Tls12; // The default value is determined by the OS. Set manually to force version.
+                        // The used public broker sometimes has invalid certificates. This sample accepts all
+                        // certificates. This should not be used in live environments.
+                        o.CertificateValidationHandler = _ => true;
                     })
                 .Build();
 
             // In MQTTv5 the response contains much more information.
-            var response = await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+            using (var timeout = new CancellationTokenSource(5000))
+            {
+                var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+                
+                Console.WriteLine("The MQTT client is connected.");
 
-            Console.WriteLine("The MQTT client is connected.");
+                response.DumpToConsole();
-
+            }
-            response.DumpToConsole();
         }
     }
 
+    public static async Task Inspect_Certificate_Validation_Errors()
+    {
+        /*
+         * This sample prints the certificate information while connection. This data can be used to decide whether a connection is secure or not
+         * including the reason for that status.
+         */
+
+        var mqttFactory = new MqttFactory();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("mqtt.fluux.io", 8883)
+                .WithTls(
+                    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;
+                        };
+                    })
+                .Build();
+
+            // In MQTTv5 the response contains much more information.
+            using (var timeout = new CancellationTokenSource(5000))
+            {
+                await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+            }
+        }
+    }
+    
     public static async Task Ping_Server()
     {
         /*

Samples/MQTTnet.Samples.csproj

@@ -15,8 +15,7 @@
     <ItemGroup>
         <PackageReference Include="Microsoft.CodeAnalysis.Compilers" Version="4.0.1" />
         <PackageReference Include="Microsoft.Net.Compilers" Version="4.0.1" />
-        <PackageReference Include="MSTest.TestFramework" Version="2.1.2" />
+
-        
         <ProjectReference Include="..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj" />

Samples/Program.cs

@@ -39,10 +39,16 @@ Console.WriteLine();
 
 try
 {
-    var task = selectedSampleMethod.Invoke(null, null) as Task;
+    if (selectedSampleMethod.Invoke(null, null) is Task task)
-    task?.Wait();
+    {
+        await task;    
+    }
 }
 catch (Exception exception)
 {
     Console.WriteLine(exception.ToString());
-}
+}
+
+Console.WriteLine();
+Console.WriteLine("Press Enter to exit.");
+Console.ReadLine();

Samples/RpcClient/RpcClient_Samples.cs

@@ -8,7 +8,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Samples.RpcClient;
 
-public static class RcpClient_Samples
+public static class RpcClient_Samples
 {
     /*
      * The extension MQTTnet.Extensions.Rpc (available as nuget) allows sending a request and waiting for the matching reply.
@@ -33,7 +33,7 @@ public static class RcpClient_Samples
             
             using (var mqttRpcClient = mqttFactory.CreateMqttRpcClient(mqttClient))
             {
-                // Access to a fully featured application message is not supported for RCP calls!
+                // Access to a fully featured application message is not supported for RPC calls!
                 // The method will throw an exception when the response was not received in time.
                 await mqttRpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
             }
@@ -81,4 +81,4 @@ public static class RcpClient_Samples
         // ! Do not send response message with the _retain_ flag set to true.
         // ! All required data for a RPC call and the result must be placed into the payload.
      */
-}
+}

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -0,0 +1,74 @@
+using System.Text.Json;
+
+namespace MQTTnet.Samples.Server;
+
+public static class Server_Retained_Messages_Samples
+{
+    public static async Task Persist_Retained_Messages()
+    {
+        /*
+         * This sample starts a simple MQTT server which will store all retained messages in a file.
+         */
+
+        var storePath = Path.Combine(Path.GetTempPath(), "RetainedMessages.json");
+
+        var mqttFactory = new MqttFactory();
+
+        // Due to security reasons the "default" endpoint (which is unencrypted) is not enabled by default!
+        var mqttServerOptions = mqttFactory.CreateServerOptionsBuilder().WithDefaultEndpoint().Build();
+
+        using (var server = mqttFactory.CreateMqttServer(mqttServerOptions))
+        {
+            // Make sure that the server will load the retained messages.
+            server.LoadingRetainedMessageAsync += async eventArgs =>
+            {
+                try
+                {
+                    eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttApplicationMessage>>(File.OpenRead(storePath));
+                    Console.WriteLine("Retained messages loaded.");
+                }
+                catch (FileNotFoundException)
+                {
+                    // Ignore because nothing is stored yet.
+                    Console.WriteLine("No retained messages stored yet.");
+                }
+                catch (Exception exception)
+                {
+                    Console.WriteLine(exception);
+                }
+            };
+
+            // Make sure to persist the changed retained messages.
+            server.RetainedMessageChangedAsync += async eventArgs =>
+            {
+                try
+                {
+                    // This sample uses the property _StoredRetainedMessages_ which will contain all(!) retained messages.
+                    // The event args also contain the affected retained message (property ChangedRetainedMessage). This can be
+                    // used to write all retained messages to dedicated files etc. Then all files must be loaded and a full list
+                    // of retained messages must be provided in the loaded event.
+
+                    var buffer = JsonSerializer.SerializeToUtf8Bytes(eventArgs.StoredRetainedMessages);
+                    await File.WriteAllBytesAsync(storePath, buffer);
+                    Console.WriteLine("Retained messages saved.");
+                }
+                catch (Exception exception)
+                {
+                    Console.WriteLine(exception);
+                }
+            };
+
+            // Make sure to clear the retained messages when they are all deleted via API.
+            server.RetainedMessagesClearedAsync += _ =>
+            {
+                File.Delete(storePath);
+                return Task.CompletedTask;
+            };
+
+            await server.StartAsync();
+
+            Console.WriteLine("Press Enter to exit.");
+            Console.ReadLine();
+        }
+    }
+}

Samples/Server/Server_Simple_Samples.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using MQTTnet.Diagnostics;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 
@@ -9,27 +10,95 @@ namespace MQTTnet.Samples.Server;
 
 public static class Server_Simple_Samples
 {
+    public static async Task Force_Disconnecting_Client()
+    {
+        /*
+         * This sample will disconnect a client.
+         *
+         * See _Run_Minimal_Server_ for more information.
+         */
+
+        using (var mqttServer = await StartMqttServer())
+        {
+            // Let the client connect.
+            await Task.Delay(TimeSpan.FromSeconds(5));
+
+            // Now disconnect the client (if connected).
+            var affectedClient = (await mqttServer.GetClientsAsync()).FirstOrDefault(c => c.Id == "MyClient");
+            if (affectedClient != null)
+            {
+                await affectedClient.DisconnectAsync();
+            }
+        }
+    }
+
+    public static async Task Publish_Message_From_Broker()
+    {
+        /*
+         * This sample will publish a message directly at the broker.
+         *
+         * See _Run_Minimal_Server_ for more information.
+         */
+
+        using (var mqttServer = await StartMqttServer())
+        {
+            // Create a new message using the builder as usual.
+            var message = new MqttApplicationMessageBuilder().WithTopic("HelloWorld").WithPayload("Test").Build();
+
+            // Now inject the new message at the broker.
+            await mqttServer.InjectApplicationMessage(
+                new InjectedMqttApplicationMessage(message)
+                {
+                    SenderClientId = "SenderClientId"
+                });
+        }
+    }
+
     public static async Task Run_Minimal_Server()
     {
         /*
          * This sample starts a simple MQTT server which will accept any TCP connection.
          */
-        
+
         var mqttFactory = new MqttFactory();
 
         // The port for the default endpoint is 1883.
         // The default endpoint is NOT encrypted!
         // Use the builder classes where possible.
-        var mqttServerOptions = new MqttServerOptionsBuilder()
+        var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
-            .WithDefaultEndpoint()
+
-            .Build();
-        
         // The port can be changed using the following API (not used in this example).
-        new MqttServerOptionsBuilder()
+        // new MqttServerOptionsBuilder()
-            .WithDefaultEndpoint()
+        //     .WithDefaultEndpoint()
-            .WithDefaultEndpointPort(1234)
+        //     .WithDefaultEndpointPort(1234)
-            .Build();
+        //     .Build();
-        
+
+        using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
+        {
+            await mqttServer.StartAsync();
+
+            Console.WriteLine("Press Enter to exit.");
+            Console.ReadLine();
+
+            // Stop and dispose the MQTT server if it is no longer needed!
+            await mqttServer.StopAsync();
+        }
+    }
+
+    public static async Task Run_Server_With_Logging()
+    {
+        /*
+         * This sample starts a simple MQTT server and prints the logs to the output.
+         *
+         * IMPORTANT! Do not enable logging in live environment. It will decrease performance.
+         *
+         * See sample "Run_Minimal_Server" for more details.
+         */
+
+        var mqttFactory = new MqttFactory(new ConsoleLogger());
+
+        var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
+
         using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
         {
             await mqttServer.StartAsync();
@@ -52,9 +121,7 @@ public static class Server_Simple_Samples
 
         var mqttFactory = new MqttFactory();
 
-        var mqttServerOptions = new MqttServerOptionsBuilder()
+        var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
-            .WithDefaultEndpoint()
-            .Build();
 
         using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
         {
@@ -88,4 +155,61 @@ public static class Server_Simple_Samples
             await mqttServer.StopAsync();
         }
     }
+
+    static async Task<MqttServer> StartMqttServer()
+    {
+        var mqttFactory = new MqttFactory();
+
+        // Due to security reasons the "default" endpoint (which is unencrypted) is not enabled by default!
+        var mqttServerOptions = mqttFactory.CreateServerOptionsBuilder().WithDefaultEndpoint().Build();
+        var server = mqttFactory.CreateMqttServer(mqttServerOptions);
+        await server.StartAsync();
+        return server;
+    }
+
+    class ConsoleLogger : IMqttNetLogger
+    {
+        readonly object _consoleSyncRoot = new();
+
+        public bool IsEnabled => true;
+
+        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
+        {
+            var foregroundColor = ConsoleColor.White;
+            switch (logLevel)
+            {
+                case MqttNetLogLevel.Verbose:
+                    foregroundColor = ConsoleColor.White;
+                    break;
+
+                case MqttNetLogLevel.Info:
+                    foregroundColor = ConsoleColor.Green;
+                    break;
+
+                case MqttNetLogLevel.Warning:
+                    foregroundColor = ConsoleColor.DarkYellow;
+                    break;
+
+                case MqttNetLogLevel.Error:
+                    foregroundColor = ConsoleColor.Red;
+                    break;
+            }
+
+            if (parameters?.Length > 0)
+            {
+                message = string.Format(message, parameters);
+            }
+
+            lock (_consoleSyncRoot)
+            {
+                Console.ForegroundColor = foregroundColor;
+                Console.WriteLine(message);
+
+                if (exception != null)
+                {
+                    Console.WriteLine(exception);
+                }
+            }
+        }
+    }
 }

Source/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj

@@ -9,9 +9,9 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.8.0" />
+        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.2.0" />
-        <PackageReference Include="MSTest.TestAdapter" Version="2.1.2" />
+        <PackageReference Include="MSTest.TestAdapter" Version="2.2.10" />
-        <PackageReference Include="MSTest.TestFramework" Version="2.1.2" />
+        <PackageReference Include="MSTest.TestFramework" Version="2.2.10" />
         <PackageReference Include="Microsoft.CodeAnalysis.Compilers" Version="4.0.1" />
         <PackageReference Include="Microsoft.Net.Compilers" Version="4.0.1" />
     </ItemGroup>

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -16,7 +16,7 @@
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <IncludeSymbols>true</IncludeSymbols>
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
         <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
         <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
@@ -27,7 +27,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
@@ -41,16 +41,9 @@
             <PackagePath>\</PackagePath>
         </None>
     </ItemGroup>
-
+    
     <ItemGroup>
-        <None Include="..\..\LICENSE">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
-    <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
     </ItemGroup>
 
     <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">

Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -46,7 +46,7 @@ namespace MQTTnet.Benchmarks
 
             _stream.Position = 0;
 
-            var channel = new TestMqttChannel(_stream);
+            var channel = new MemoryMqttChannel(_stream);
             
             _channelAdapter = new MqttChannelAdapter(channel, serializer, null, new MqttNetEventLogger());
         }

Source/MQTTnet.Benchmarks/LoggerBenchmark.cs

@@ -8,7 +8,7 @@ using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter]
     [MemoryDiagnoser]
     public class LoggerBenchmark

Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -12,7 +12,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="BenchmarkDotNet" Version="0.12.1" />
+        <PackageReference Include="BenchmarkDotNet" Version="0.13.1" />
         <PackageReference Include="Microsoft.AspNetCore" Version="2.2.0" />
     </ItemGroup>
 

Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs

@@ -6,11 +6,10 @@ using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Client;
 using MQTTnet.Server;
-using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter, RankColumn]
     [MemoryDiagnoser]
     public class MessageProcessingBenchmark

Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs

@@ -13,7 +13,7 @@ using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [MemoryDiagnoser]
     public class MessageProcessingMqttConnectionContextBenchmark
     {

Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs

@@ -7,7 +7,7 @@ using MQTTnet.Tests.Mockups;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [MemoryDiagnoser]
     public class MqttPacketReaderWriterBenchmark
     {

Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -14,7 +14,7 @@ using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [MemoryDiagnoser]
     public sealed class MqttTcpChannelBenchmark
     {

Source/MQTTnet.Benchmarks/RoundtripProcessingBenchmark.cs

@@ -5,7 +5,7 @@ using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter, RankColumn]
     [MemoryDiagnoser]
     public class RoundtripProcessingBenchmark

Source/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -17,7 +17,7 @@ using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter]
     [MemoryDiagnoser]
     public class SerializerBenchmark : BaseBenchmark

Source/MQTTnet.Benchmarks/ServerProcessingBenchmark.cs

@@ -5,7 +5,7 @@ using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter, RankColumn]
     [MemoryDiagnoser]
     public class ServerProcessingBenchmark

Source/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -13,7 +13,7 @@ using MQTTnet.AspNetCore.Client.Tcp;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [MemoryDiagnoser]
     public class TcpPipesBenchmark
     {

Source/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs

@@ -8,7 +8,7 @@ using BenchmarkDotNet.Jobs;
 
 namespace MQTTnet.Benchmarks
 {
-    [SimpleJob(RuntimeMoniker.NetCoreApp50)]
+    [SimpleJob(RuntimeMoniker.Net60)]
     [RPlotExporter]
     [MemoryDiagnoser]
     public class TopicFilterComparerBenchmark

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -15,13 +15,13 @@ namespace MQTTnet.Extensions.ManagedClient
         
         event Func<ApplicationMessageSkippedEventArgs, Task> ApplicationMessageSkippedAsync;
         
-        event Func<EventArgs, Task> ConnectedAsync;
+        event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
         
         event Func<ConnectingFailedEventArgs, Task> ConnectingFailedAsync;
         
         event Func<EventArgs, Task> ConnectionStateChangedAsync;
         
-        event Func<EventArgs, Task> DisconnectedAsync;
+        event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
         
         event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
         

Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj

@@ -18,7 +18,7 @@
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <IncludeSymbols>true</IncludeSymbols>
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
         <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
         <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,16 +55,9 @@
             <PackagePath>\</PackagePath>
         </None>
     </ItemGroup>
-
+    
     <ItemGroup>
-        <None Include="..\..\LICENSE">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
-    <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
     </ItemGroup>
 
     <ItemGroup>

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -51,9 +51,9 @@ namespace MQTTnet.Extensions.ManagedClient
         ///         cref="MaintainConnectionAsync" />
         ///     .
         /// </summary>
-        readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
+        readonly Dictionary<string, MqttTopicFilter> _reconnectSubscriptions = new Dictionary<string, MqttTopicFilter>();
 
-        readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
+        readonly Dictionary<string, MqttTopicFilter> _subscriptions = new Dictionary<string, MqttTopicFilter>();
         readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
         readonly HashSet<string> _unsubscriptions = new HashSet<string>();
 
@@ -93,7 +93,7 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => InternalClient.ApplicationMessageReceivedAsync -= value;
         }
 
-        public event Func<EventArgs, Task> ConnectedAsync
+        public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
         {
             add => InternalClient.ConnectedAsync += value;
             remove => InternalClient.ConnectedAsync -= value;
@@ -111,7 +111,7 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => _connectionStateChangedEvent.RemoveHandler(value);
         }
 
-        public event Func<EventArgs, Task> DisconnectedAsync
+        public event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync
         {
             add => InternalClient.DisconnectedAsync += value;
             remove => InternalClient.DisconnectedAsync -= value;
@@ -201,7 +201,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
             finally
             {
-                if (_applicationMessageSkippedEvent.HasHandlers)
+                if (applicationMessageSkippedEventArgs != null && _applicationMessageSkippedEvent.HasHandlers)
                 {
                     await _applicationMessageSkippedEvent.InvokeAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false);
                 }
@@ -289,7 +289,7 @@ namespace MQTTnet.Extensions.ManagedClient
             {
                 foreach (var topicFilter in topicFilters)
                 {
-                    _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    _subscriptions[topicFilter.Topic] = topicFilter;
                     _unsubscriptions.Remove(topicFilter.Topic);
                 }
             }
@@ -455,18 +455,11 @@ namespace MQTTnet.Extensions.ManagedClient
             {
                 if (_reconnectSubscriptions.Any())
                 {
-                    var subscriptions = _reconnectSubscriptions.Select(
-                        i => new MqttTopicFilter
-                        {
-                            Topic = i.Key,
-                            QualityOfServiceLevel = i.Value
-                        });
-
                     var topicFilters = new List<MqttTopicFilter>();
 
-                    foreach (var sub in subscriptions)
+                    foreach (var sub in _reconnectSubscriptions)
                     {
-                        topicFilters.Add(sub);
+                        topicFilters.Add(sub.Value);
 
                         if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
                         {
@@ -495,14 +488,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
                 lock (_subscriptions)
                 {
-                    subscriptions = _subscriptions.Select(
+                    subscriptions = _subscriptions.Values.ToList();
-                            i => new MqttTopicFilter
-                            {
-                                Topic = i.Key,
-                                QualityOfServiceLevel = i.Value
-                            })
-                        .ToList();
-
                     _subscriptions.Clear();
 
                     unsubscriptions = new HashSet<string>(_unsubscriptions);
@@ -523,7 +509,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
                 foreach (var subscription in subscriptions)
                 {
-                    _reconnectSubscriptions[subscription.Topic] = subscription.QualityOfServiceLevel;
+                    _reconnectSubscriptions[subscription.Topic] = subscription;
                 }
 
                 var addedTopicFilters = new List<MqttTopicFilter>();

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

@@ -18,7 +18,7 @@
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <IncludeSymbols>true</IncludeSymbols>
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
         <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
         <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,16 +55,9 @@
             <PackagePath>\</PackagePath>
         </None>
     </ItemGroup>
-
+    
     <ItemGroup>
-        <None Include="..\..\LICENSE">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
-    <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
     </ItemGroup>
 
     <ItemGroup>

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

@@ -18,7 +18,7 @@
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <IncludeSymbols>true</IncludeSymbols>
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
         <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
         <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,24 +55,17 @@
             <PackagePath>\</PackagePath>
         </None>
     </ItemGroup>
-
+    
     <ItemGroup>
-        <None Include="..\..\LICENSE">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
+        <PackageReference Include="WebSocket4Net" Version="0.15.2" />
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="WebSocket4Net" Version="0.15.2"/>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
-    </ItemGroup>
-
-    <ItemGroup>
-        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
     </ItemGroup>
 
 </Project>

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -12,7 +12,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />
+        <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
         <PackageReference Include="Microsoft.CodeAnalysis.Compilers" Version="4.0.1" />
         <PackageReference Include="Microsoft.Net.Compilers" Version="4.0.1" />
     </ItemGroup>
@@ -23,10 +23,4 @@
         <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
     </ItemGroup>
 
-    <ItemGroup>
-        <None Update="Start.bat">
-            <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-        </None>
-    </ItemGroup>
-
 </Project>

Source/MQTTnet.TestApp/Start.bat

@@ -1 +0,0 @@
-dotnet MQTTnet.TestApp.NetCore.dll

Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs

@@ -563,7 +563,7 @@ namespace MQTTnet.Tests
 
             using (var headerStream = new MemoryStream(buffer1.Join().ToArray()))
             {
-                using (var channel = new TestMqttChannel(headerStream))
+                using (var channel = new MemoryMqttChannel(headerStream))
                 {
                     using (var adapter = new MqttChannelAdapter(
                                channel,
@@ -583,7 +583,7 @@ namespace MQTTnet.Tests
 
         MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter packetFormatterAdapter, byte[] buffer)
         {
-            var channel = new TestMqttChannel(buffer);
+            var channel = new MemoryMqttChannel(buffer);
             var adapter = new MqttChannelAdapter(channel, packetFormatterAdapter, null, new MqttNetEventLogger());
 
             adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
@@ -603,7 +603,7 @@ namespace MQTTnet.Tests
             var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
             var buffer = serializer.Encode(packet);
 
-            using (var channel = new TestMqttChannel(buffer.Join().ToArray()))
+            using (var channel = new MemoryMqttChannel(buffer.Join().ToArray()))
             {
                 var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), null, new MqttNetEventLogger());
                 return (TPacket)adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();

Source/MQTTnet.Tests/MQTTnet.Tests.csproj

@@ -10,9 +10,9 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="MSTest.TestAdapter" Version="2.1.2" />
+        <PackageReference Include="MSTest.TestAdapter" Version="2.2.10" />
-        <PackageReference Include="MSTest.TestFramework" Version="2.1.2" />
+        <PackageReference Include="MSTest.TestFramework" Version="2.2.10" />
-        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.8.0" />
+        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.2.0" />
 
         <PackageReference Include="Microsoft.CodeAnalysis.Compilers" Version="4.0.1" />
         <PackageReference Include="Microsoft.Net.Compilers" Version="4.0.1" />

Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs

@@ -212,7 +212,7 @@ namespace MQTTnet.Tests.MQTTv5
                 
                 var client1 = await testEnvironment.ConnectClient(o => o.WithProtocolVersion(MqttProtocolVersion.V500).WithClientId("client1"));
 
-                var testMessageHandler = new TestApplicationMessageReceivedHandler(client1);
+                var testMessageHandler = testEnvironment.CreateApplicationMessageHandler(client1);
 
                 await client1.SubscribeAsync("a");
 

Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs

@@ -10,16 +10,16 @@ using MQTTnet.Channel;
 
 namespace MQTTnet.Tests.Mockups
 {
-    public sealed class TestMqttChannel : IMqttChannel
+    public sealed class MemoryMqttChannel : IMqttChannel
     {
         readonly MemoryStream _stream;
 
-        public TestMqttChannel(MemoryStream stream)
+        public MemoryMqttChannel(MemoryStream stream)
         {
             _stream = stream;
         }
 
-        public TestMqttChannel(byte[] buffer)
+        public MemoryMqttChannel(byte[] buffer)
         {
             _stream = new MemoryStream(buffer);
         }

Source/MQTTnet.Tests/Mockups/MqttPacketAsserts.cs

@@ -1,22 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Packets;
-
-namespace MQTTnet.Tests.Mockups
-{
-    public sealed class MqttPacketAsserts
-    {
-        public void AssertIsConnectPacket(MqttPacket packet)
-        {
-            Assert.AreEqual(packet.GetType(), typeof(MqttConnectPacket));
-        }
-        
-        public void AssertIsConnAckPacket(MqttPacket packet)
-        {
-            Assert.AreEqual(packet.GetType(), typeof(MqttConnAckPacket));
-        }
-    }
-}

Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs

@@ -27,7 +27,7 @@ namespace MQTTnet.Tests
 
         public MqttPacket Decode(MqttPacketBuffer buffer)
         {
-            using (var channel = new TestMqttChannel(buffer.ToArray()))
+            using (var channel = new MemoryMqttChannel(buffer.ToArray()))
             {
                 var formatterAdapter = new MqttPacketFormatterAdapter(_protocolVersion, new MqttBufferWriter(4096, 65535));
 

Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs

@@ -16,67 +16,52 @@ namespace MQTTnet.Tests.Server
     public sealed class Retained_Messages_Tests : BaseTestClass
     {
         [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V3()
+        public async Task Clear_Retained_Message_With_Empty_Payload()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
-                var client = testEnvironment.CreateClient();
-                
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V311)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V5()
-        {
-            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-            {
-                await testEnvironment.StartServer();
-
-                var client = testEnvironment.CreateClient();
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V500)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Retained_Messages_Flow()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
-
-                await testEnvironment.StartServer();
                 var c1 = await testEnvironment.ConnectClient();
-    
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
+
+                await c1.DisconnectAsync();
+
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
-                await c1.PublishAsync(retainedMessage);
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
+            }
+        }
+
+        [TestMethod]
+        public async Task Clear_Retained_Message_With_Null_Payload()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var c1 = await testEnvironment.ConnectClient();
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+
                 await c1.DisconnectAsync();
-                await LongTestDelay();
 
-                for (var i = 0; i < 5; i++)
+                var c2 = await testEnvironment.ConnectClient();
-                {
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                    await c2.UnsubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i);
-                    
-                    await c2.SubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i + 1);
-                }
 
-                await c2.DisconnectAsync();
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
             }
         }
 
@@ -114,7 +99,7 @@ namespace MQTTnet.Tests.Server
 
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                 await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained").Build());
 
                 await Task.Delay(500);
@@ -125,52 +110,102 @@ namespace MQTTnet.Tests.Server
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message_With_Empty_Payload()
+        public async Task Receive_Retained_Messages_From_Higher_Qos_Level()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
+                // Upload retained message.
                 var c1 = await testEnvironment.ConnectClient();
-
+                await c1.PublishAsync(
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                    new MqttApplicationMessageBuilder().WithTopic("retained")
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
+                        .WithPayload(new byte[1])
-
+                        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
+                        .WithRetainFlag()
+                        .Build());
+                
                 await c1.DisconnectAsync();
 
+                // Subscribe using a new client.
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                 await Task.Delay(200);
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                // Using QoS 2 will lead to 1 instead because the publish was made with QoS level 1 (see 3.8.4 SUBSCRIBE Actions)!
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
                 await Task.Delay(500);
 
-                messageHandler.AssertReceivedCountEquals(0);
+                messageHandler.AssertReceivedCountEquals(1);
             }
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message_With_Null_Payload()
+        public async Task Retained_Messages_Flow()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
+
+                await testEnvironment.StartServer();
+                var c1 = await testEnvironment.ConnectClient();
+
+                var c2 = await testEnvironment.ConnectClient();
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+
+                await c1.PublishAsync(retainedMessage);
+                await c1.DisconnectAsync();
+                await LongTestDelay();
+
+                for (var i = 0; i < 5; i++)
+                {
+                    await c2.UnsubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i);
+
+                    await c2.SubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i + 1);
+                }
+
+                await c2.DisconnectAsync();
+            }
+        }
+
+        [TestMethod]
+        public async Task Server_Reports_Retained_Messages_Supported_V3()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
-                var c1 = await testEnvironment.ConnectClient();
+                var client = testEnvironment.CreateClient();
 
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                var connectResult = await client.ConnectAsync(
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V311)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());
 
-                await c1.DisconnectAsync();
+                Assert.IsTrue(connectResult.RetainAvailable);
+            }
+        }
 
-                var c2 = await testEnvironment.ConnectClient();
+        [TestMethod]
-                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+        public async Task Server_Reports_Retained_Messages_Supported_V5()
-                
+        {
-                await Task.Delay(200);
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+            {
-                await Task.Delay(500);
+                await testEnvironment.StartServer();
 
-                messageHandler.AssertReceivedCountEquals(0);
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V500)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());
+
+                Assert.IsTrue(connectResult.RetainAvailable);
             }
         }
     }

Source/MQTTnet.Tests/Server/Security_Tests.cs

@@ -2,7 +2,10 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
+using Microsoft.CodeAnalysis;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Client;
@@ -15,6 +18,104 @@ namespace MQTTnet.Tests.Server
     [TestClass]
     public sealed class Security_Tests : BaseTestClass
     {
+        [TestMethod]
+        public async Task Do_Not_Affect_Authorized_Clients()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                
+                await testEnvironment.StartServer();
+
+                var publishedApplicationMessages = new List<MqttApplicationMessage>();
+                testEnvironment.Server.InterceptingPublishAsync += eventArgs =>
+                {
+                    lock (publishedApplicationMessages)
+                    {
+                        publishedApplicationMessages.Add(eventArgs.ApplicationMessage);
+                    }
+
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                testEnvironment.Server.ValidatingConnectionAsync += eventArgs =>
+                {
+                    if (eventArgs.Username == "SECRET")
+                    {
+                        eventArgs.ReasonCode = MqttConnectReasonCode.Success;
+                    }
+                    else
+                    {
+                        eventArgs.ReasonCode = MqttConnectReasonCode.NotAuthorized;
+                    }
+
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                using (var validClient = testEnvironment.CreateClient())
+                {
+                    await validClient.ConnectAsync(
+                        testEnvironment.Factory.CreateClientOptionsBuilder()
+                            .WithTcpServer("localhost", testEnvironment.ServerPort)
+                            .WithCredentials("SECRET")
+                            .WithClientId("CLIENT")
+                            .Build());
+
+                    await validClient.PublishStringAsync("HELLO 1");
+
+                    // The following code tries to connect a new client with the same client ID but invalid
+                    // credentials. This should block the second client but keep the first one up and running.
+                    try
+                    {
+                        using (var invalidClient = testEnvironment.CreateClient())
+                        {
+                            await invalidClient.ConnectAsync(
+                                testEnvironment.Factory.CreateClientOptionsBuilder()
+                                    .WithTcpServer("localhost", testEnvironment.ServerPort)
+                                    .WithCredentials("???")
+                                    .WithClientId("CLIENT")
+                                    .Build());
+                        }
+                    }
+                    catch
+                    {
+                        // Ignore errors from the second client.
+                    }
+
+                    await LongTestDelay();
+                    
+                    await validClient.PublishStringAsync("HELLO 2");
+                    
+                    await LongTestDelay();
+                    
+                    await validClient.PublishStringAsync("HELLO 3");
+                    
+                    await LongTestDelay();
+                    
+                    Assert.AreEqual(3, publishedApplicationMessages.Count);
+                    Assert.AreEqual(1, testEnvironment.Server.GetClientsAsync().GetAwaiter().GetResult().Count);
+                }
+            }
+        }
+
+        [TestMethod]
+        public Task Handle_Wrong_Password()
+        {
+            return TestCredentials("UserName", "x");
+        }
+
+        [TestMethod]
+        public Task Handle_Wrong_UserName()
+        {
+            return TestCredentials("x", "Password1");
+        }
+
+        [TestMethod]
+        public Task Handle_Wrong_UserName_And_Password()
+        {
+            return TestCredentials("x", "x");
+        }
+
         [TestMethod]
         public async Task Use_Username_Null_Password_Empty()
         {
@@ -29,10 +130,7 @@ namespace MQTTnet.Tests.Server
 
                 var client = testEnvironment.CreateClient();
 
-                var clientOptions = new MqttClientOptionsBuilder()
+                var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithCredentials(username, password).Build();
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                    .WithCredentials(username, password)
-                    .Build();
 
                 var ex = await Assert.ThrowsExceptionAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions));
                 Assert.IsInstanceOfType(ex.InnerException, typeof(MqttProtocolViolationException));
@@ -41,25 +139,7 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        [TestMethod]
+        async Task TestCredentials(string userName, string password)
-        public Task Handle_Wrong_UserName()
-        {
-            return TestCredentials("x", "Password1");
-        }
-
-        [TestMethod]
-        public Task Handle_Wrong_Password()
-        {
-            return TestCredentials("UserName", "x");
-        }
-
-        [TestMethod]
-        public Task Handle_Wrong_UserName_And_Password()
-        {
-            return TestCredentials("x", "x");
-        }
-
-        private async Task TestCredentials(string userName, string password)
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
@@ -84,10 +164,7 @@ namespace MQTTnet.Tests.Server
 
                 var client = testEnvironment.CreateClient();
 
-                var clientOptions = new MqttClientOptionsBuilder()
+                var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithCredentials(userName, password).Build();
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                    .WithCredentials(userName, password)
-                    .Build();
 
                 var ex = await Assert.ThrowsExceptionAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions));
                 Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, ex.Result.ResultCode);

Source/MQTTnet.Tests/Server/Subscribe_Tests.cs

@@ -12,6 +12,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Implementations;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Tests.Server
 {
@@ -45,6 +46,30 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Enqueue_Message_After_Subscription()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                server.ClientSubscribedTopicAsync += e =>
+                {
+                    server.InjectApplicationMessage(new InjectedMqttApplicationMessage(new MqttApplicationMessageBuilder().WithTopic("test_topic").Build()));
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client = await testEnvironment.ConnectClient();
+                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(client);
+
+                await client.SubscribeAsync("test_topic");
+
+                await LongTestDelay();
+
+                Assert.AreEqual(1, receivedMessages.ReceivedEventArgs.Count);
+            }
+        }
+
         [TestMethod]
         public async Task Intercept_Subscription()
         {
@@ -101,7 +126,7 @@ namespace MQTTnet.Tests.Server
                     .WithTopicFilter("c", MqttQualityOfServiceLevel.ExactlyOnce)
                     .WithTopicFilter("d")
                     .Build();
-                
+
                 var response = await client.SubscribeAsync(subscribeOptions);
 
                 Assert.AreEqual(subscribeOptions.TopicFilters.Count, response.Items.Count);

Source/MQTTnet/Client/Options/MqttClientOptions.cs

@@ -98,6 +98,16 @@ namespace MQTTnet.Client
         /// </summary>
         public ushort TopicAliasMaximum { get; set; }
 
+        /// <summary>
+        ///     If set to true, the bridge will attempt to indicate to the remote broker that it is a bridge not an ordinary
+        ///     client.
+        ///     If successful, this means that loop detection will be more effective and that retained messages will be propagated
+        ///     correctly.
+        ///     Not all brokers support this feature so it may be necessary to set it to false if your bridge does not connect
+        ///     properly.
+        /// </summary>
+        public bool TryPrivate { get; set; } = true;
+
         /// <summary>
         ///     Gets or sets the user properties.
         ///     In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -96,16 +96,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        /// <summary>
-        ///     Sets the timeout which will be applied at socket level and internal operations.
-        ///     The default value is the same as for sockets in .NET in general.
-        /// </summary>
-        public MqttClientOptionsBuilder WithTimeout(TimeSpan value)
-        {
-            _options.Timeout = value;
-            return this;
-        }
-
         public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
         {
             if (uri == null)
@@ -289,6 +279,16 @@ namespace MQTTnet.Client
             return this;
         }
 
+        /// <summary>
+        ///     Sets the timeout which will be applied at socket level and internal operations.
+        ///     The default value is the same as for sockets in .NET in general.
+        /// </summary>
+        public MqttClientOptionsBuilder WithTimeout(TimeSpan value)
+        {
+            _options.Timeout = value;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters)
         {
             _tlsParameters = parameters;
@@ -322,6 +322,20 @@ namespace MQTTnet.Client
             return this;
         }
 
+        /// <summary>
+        ///     If set to true, the bridge will attempt to indicate to the remote broker that it is a bridge not an ordinary
+        ///     client.
+        ///     If successful, this means that loop detection will be more effective and that retained messages will be propagated
+        ///     correctly.
+        ///     Not all brokers support this feature so it may be necessary to set it to false if your bridge does not connect
+        ///     properly.
+        /// </summary>
+        public MqttClientOptionsBuilder WithTryPrivate(bool tryPrivate = true)
+        {
+            _options.TryPrivate = true;
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithUserProperty(string name, string value)
         {
             if (name is null)
@@ -379,6 +393,17 @@ namespace MQTTnet.Client
             _options.WillPayload = willPayload;
             return this;
         }
+        
+        public MqttClientOptionsBuilder WithWillPayload(string willPayload)
+        {
+            if (string.IsNullOrEmpty(willPayload))
+            {
+                return WithWillPayload((byte[])null);
+            }
+            
+            _options.WillPayload = Encoding.UTF8.GetBytes(willPayload);
+            return this;
+        }
 
         public MqttClientOptionsBuilder WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel willQualityOfServiceLevel)
         {
@@ -391,5 +416,46 @@ namespace MQTTnet.Client
             _options.WillTopic = willTopic;
             return this;
         }
+
+        public MqttClientOptionsBuilder WithWillRetain(bool willRetain = true)
+        {
+            _options.WillRetain = willRetain;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillContentType(string willContentType)
+        {
+            _options.WillContentType = willContentType;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData)
+        {
+            _options.WillCorrelationData = willCorrelationData;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic)
+        {
+            _options.WillResponseTopic = willResponseTopic;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator)
+        {
+            _options.WillPayloadFormatIndicator = willPayloadFormatIndicator;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillUserProperty(string name, string value)
+        {
+            if (_options.WillUserProperties == null)
+            {
+                _options.WillUserProperties = new List<MqttUserProperty>();
+            }
+            
+            _options.WillUserProperties.Add(new MqttUserProperty(name, value));
+            return this;
+        }
     }
 }

Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs

@@ -4,6 +4,7 @@
 
 using System;
 using MQTTnet.Packets;
+using MQTTnet.Protocol;
 using MQTTnet.Server;
 
 namespace MQTTnet.Formatter
@@ -25,6 +26,7 @@ namespace MQTTnet.Formatter
                 SubscriptionIdentifiersAvailable = true,
                 SharedSubscriptionAvailable = false,
                 TopicAliasMaximum = ushort.MaxValue,
+                MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce,
                 WildcardSubscriptionAvailable = true,
 
                 AuthenticationMethod = validatingConnectionEventArgs.AuthenticationMethod,
@@ -32,7 +34,11 @@ namespace MQTTnet.Formatter
                 AssignedClientIdentifier = validatingConnectionEventArgs.AssignedClientIdentifier,
                 ReasonString = validatingConnectionEventArgs.ReasonString,
                 ServerReference = validatingConnectionEventArgs.ServerReference,
-                UserProperties = validatingConnectionEventArgs.ResponseUserProperties
+                UserProperties = validatingConnectionEventArgs.ResponseUserProperties,
+
+                ResponseInformation = null,
+                MaximumPacketSize = 0, // Unlimited,
+                ReceiveMaximum = 0 // Unlimited
             };
 
             return connAckPacket;

Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs

@@ -33,7 +33,8 @@ namespace MQTTnet.Formatter
                 RequestResponseInformation = clientOptions.RequestResponseInformation,
                 SessionExpiryInterval = clientOptions.SessionExpiryInterval,
                 TopicAliasMaximum = clientOptions.TopicAliasMaximum,
-                UserProperties = clientOptions.UserProperties
+                UserProperties = clientOptions.UserProperties,
+                TryPrivate = clientOptions.TryPrivate
             };
 
             if (!string.IsNullOrEmpty(clientOptions.WillTopic))

Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs

@@ -95,6 +95,10 @@ namespace MQTTnet.Formatter
             var protocolName = _bufferReader.ReadString();
             var protocolLevel = _bufferReader.ReadByte();
 
+            // Remove the mosquitto try_private flag (MQTT 3.1.1 Bridge).
+            // This flag is accepted but not yet used.
+            protocolLevel &= 0x7F;
+            
             if (protocolName == "MQTT")
             {
                 if (protocolLevel == 5)

Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs

@@ -175,12 +175,18 @@ namespace MQTTnet.Formatter.V3
                 throw new MqttProtocolViolationException("MQTT protocol name do not match MQTT v3.");
             }
 
+            var tryPrivate = (protocolVersion & 0x80) > 0;
+            protocolVersion &= 0x7F;
+            
             if (protocolVersion != 3 && protocolVersion != 4)
             {
                 throw new MqttProtocolViolationException("MQTT protocol version do not match MQTT v3.");
             }
 
-            var packet = new MqttConnectPacket();
+            var packet = new MqttConnectPacket
+            {
+                TryPrivate = tryPrivate
+            };
 
             var connectFlags = _bufferReader.ReadByte();
             if ((connectFlags & 0x1) > 0)
@@ -409,7 +415,14 @@ namespace MQTTnet.Formatter.V3
             ValidateConnectPacket(packet);
 
             bufferWriter.WriteString("MQIsdp");
-            bufferWriter.WriteByte(3); // Protocol Level 3
+
+            var protocolVersion = 3;
+            if (packet.TryPrivate)
+            {
+                protocolVersion |= 0x80;
+            }
+            
+            bufferWriter.WriteByte((byte)protocolVersion);
 
             byte connectFlags = 0x0;
             if (packet.CleanSession)

Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs

@@ -72,11 +72,11 @@ namespace MQTTnet.Formatter.V5
 
             if (value == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                Write(MqttPropertyId.MaximumQoS, true);
+                Write(MqttPropertyId.MaximumQoS, 0x1);
             }
             else
             {
-                Write(MqttPropertyId.MaximumQoS, false);
+                Write(MqttPropertyId.MaximumQoS, 0x0);
             }
         }
 
@@ -84,6 +84,11 @@ namespace MQTTnet.Formatter.V5
         {
             // If absent, the Application Message does not expire.
             // This library uses 0 to indicate no expiration.
+            if (value == 0)
+            {
+                return;
+            }
+            
             WriteAsFourByteInteger(MqttPropertyId.MessageExpiryInterval, value);
         }
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -3,6 +3,9 @@
 // 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;
@@ -11,9 +14,6 @@ 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;
 
 namespace MQTTnet.Implementations
 {
@@ -77,7 +77,7 @@ namespace MQTTnet.Implementations
                 {
                     socket.LingerState = _tcpOptions.LingerState;
                 }
-                
+
                 if (_tcpOptions.DualMode.HasValue)
                 {
                     // It is important to avoid setting the flag if no specific value is set by the user
@@ -103,15 +103,20 @@ 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 = _tcpOptions.Server,
                             CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy
                         };
 
                         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
 #else
-                        await sslStream.AuthenticateAsClientAsync(_tcpOptions.Server, LoadCertificates(), _tcpOptions.TlsOptions.SslProtocol,
+                        await sslStream.AuthenticateAsClientAsync(
-                            !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+                                _tcpOptions.Server,
+                                LoadCertificates(),
+                                _tcpOptions.TlsOptions.SslProtocol,
+                                !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors)
+                            .ConfigureAwait(false);
 #endif
                     }
                     catch
@@ -269,6 +274,11 @@ namespace MQTTnet.Implementations
                 return certificateValidationHandler(eventArgs);
             }
 
+            if (_tcpOptions?.TlsOptions?.IgnoreCertificateChainErrors ?? false)
+            {
+                sslPolicyErrors &= ~SslPolicyErrors.RemoteCertificateChainErrors;
+            }
+
             return sslPolicyErrors == SslPolicyErrors.None;
         }
 

Source/MQTTnet/MQTTnet.csproj

@@ -29,7 +29,7 @@
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <IncludeSymbols>true</IncludeSymbols>
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
         <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
         <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
@@ -40,7 +40,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <NoWarn>1591</NoWarn>
     </PropertyGroup>
@@ -76,14 +76,7 @@
     </ItemGroup>
 
     <ItemGroup>
-        <None Include="..\..\LICENSE">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
-    <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
     </ItemGroup>
 
     <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">

Source/MQTTnet/Packets/MqttConnectPacket.cs

@@ -64,6 +64,8 @@ namespace MQTTnet.Packets
 
         public List<MqttUserProperty> WillUserProperties { get; set; }
 
+        public bool TryPrivate { get; set; }
+        
         public override string ToString()
         {
             var passwordText = string.Empty;

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -188,6 +188,8 @@ namespace MQTTnet.Server
                 // own exception in the reading loop!
                 while (!cancellationToken.IsCancellationRequested)
                 {
+                    await Task.Yield();
+                    
                     var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
                     if (packet == null)
                     {
@@ -315,6 +317,10 @@ namespace MQTTnet.Server
                     {
                         packetBusItem.MarkAsFailed(exception);
                     }
+                    finally
+                    {
+                        await Task.Yield();
+                    }
                 }
             }
             catch (OperationCanceledException)
@@ -392,7 +398,7 @@ namespace MQTTnet.Server
                 foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
                 {
                     var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
-                    Session.EnqueueControlPacket(new MqttPacketBusItem(publishPacket));
+                    Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
                 }
             }
         }

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -175,13 +175,14 @@ namespace MQTTnet.Server
             };
 
             var addedSubscriptions = new List<string>();
+            var finalTopicFilters = new List<MqttTopicFilter>();
 
             // The topic filters are order by its QoS so that the higher QoS will win over a
             // lower one.
-            foreach (var originalTopicFilter in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
+            foreach (var topicFilterItem in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
             {
-                var subscriptionEventArgs = await InterceptSubscribe(originalTopicFilter, cancellationToken).ConfigureAwait(false);
+                var subscriptionEventArgs = await InterceptSubscribe(topicFilterItem, cancellationToken).ConfigureAwait(false);
-                var finalTopicFilter = subscriptionEventArgs.TopicFilter;
+                var topicFilter = subscriptionEventArgs.TopicFilter;
                 var processSubscription = subscriptionEventArgs.ProcessSubscription && subscriptionEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
 
                 result.UserProperties = subscriptionEventArgs.UserProperties;
@@ -195,16 +196,26 @@ namespace MQTTnet.Server
                     result.CloseConnection = true;
                 }
 
-                if (!processSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
+                if (!processSubscription || string.IsNullOrEmpty(topicFilter.Topic))
                 {
                     continue;
                 }
 
-                var createSubscriptionResult = CreateSubscription(finalTopicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
+                var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
 
-                addedSubscriptions.Add(finalTopicFilter.Topic);
+                addedSubscriptions.Add(topicFilter.Topic);
+                finalTopicFilters.Add(topicFilter);
 
-                if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
+                FilterRetainedApplicationMessages(retainedApplicationMessages, createSubscriptionResult, result);
+            }
+
+            // This call will add the new subscription to the internal storage.
+            // So the event _ClientSubscribedTopicEvent_ must be called afterwards.
+            _subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
+            
+            if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
+            {
+                foreach (var finalTopicFilter in finalTopicFilters)
                 {
                     var eventArgs = new ClientSubscribedTopicEventArgs
                     {
@@ -214,12 +225,8 @@ namespace MQTTnet.Server
 
                     await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
-
-                FilterRetainedApplicationMessages(retainedApplicationMessages, createSubscriptionResult, result);
             }
-
+            
-            _subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
-
             return result;
         }
 
@@ -263,26 +270,30 @@ namespace MQTTnet.Server
                         _subscriptions.Remove(topicFilter);
 
                         // must remove subscription object from topic hash dictionary also
-
+                        if (existingSubscription != null)
-                        if (existingSubscription.TopicHasWildcard)
                         {
-                            if (_wildcardSubscriptionsByTopicHash.TryGetValue(existingSubscription.TopicHash, out var subs))
+                            var topicHash = existingSubscription.TopicHash;
+                            
+                            if (existingSubscription.TopicHasWildcard)
                             {
-                                subs.Subscriptions.Remove(existingSubscription);
+                                if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
-                                if (subs.Subscriptions.Count == 0)
                                 {
-                                    _wildcardSubscriptionsByTopicHash.Remove(existingSubscription.TopicHash);
+                                    subscriptions.Subscriptions.Remove(existingSubscription);
+                                    if (subscriptions.Subscriptions.Count == 0)
+                                    {
+                                        _wildcardSubscriptionsByTopicHash.Remove(topicHash);
+                                    }
                                 }
                             }
-                        }
+                            else
-                        else
-                        {
-                            if (_noWildcardSubscriptionsByTopicHash.TryGetValue(existingSubscription.TopicHash, out var subs))
                             {
-                                subs.Remove(existingSubscription);
+                                if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
-                                if (subs.Count == 0)
                                 {
-                                    _noWildcardSubscriptionsByTopicHash.Remove(existingSubscription.TopicHash);
+                                    subscriptions.Remove(existingSubscription);
+                                    if (subscriptions.Count == 0)
+                                    {
+                                        _noWildcardSubscriptionsByTopicHash.Remove(topicHash);
+                                    }
                                 }
                             }
                         }
@@ -294,11 +305,7 @@ namespace MQTTnet.Server
             finally
             {
                 _subscriptionsLock.Release();
-
+                _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
-                if (_subscriptionChangedNotification != null)
-                {
-                    _subscriptionChangedNotification.OnSubscriptionsRemoved(_session, removedSubscriptions);
-                }
             }
 
             if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers)

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -19,10 +20,11 @@ namespace MQTTnet.Server
     {
         readonly MqttClientSessionsManager _clientSessionsManager;
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
+        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         readonly MqttServerOptions _serverOptions;
 
-        readonly Dictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new Dictionary<ushort, MqttPublishPacket>();
+        readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
 
         // Bookkeeping to know if this is a subscribing client; lazy initialize later.
         HashSet<string> _subscribedTopics;
@@ -71,7 +73,7 @@ namespace MQTTnet.Server
 
         public void AcknowledgePublishPacket(ushort packetIdentifier)
         {
-            _unacknowledgedPublishPackets.Remove(packetIdentifier);
+            _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out _);
         }
 
         public void AddSubscribedTopic(string topic)
@@ -100,6 +102,11 @@ namespace MQTTnet.Server
             SubscriptionsManager.Dispose();
         }
 
+        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
+        {
+            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
+        }
+
         public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
         {
             if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
@@ -118,28 +125,27 @@ namespace MQTTnet.Server
             }
 
             var publishPacket = (MqttPublishPacket)packetBusItem.Packet;
+
             if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
                 _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
         }
 
-        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
-        {
-            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
-        }
-
         public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
         {
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
         }
-        
+
         public void Recover()
         {
             // TODO: Keep the bus and only insert pending items again.
             // TODO: Check if packet identifier must be restarted or not.
+            // TODO: Recover package identifier.
             _packetBus.Clear();
 
             foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())