Enable nullable

* Call the event handler also for retained messages

* Update release notes

* Add Unit Test

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "5.0.0.${{github.run_number}}"
+  VERSION: "5.1.0.${{github.run_number}}"
   PACKAGE_SUFFIX: ""
 
 jobs:
@@ -21,8 +21,15 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
 
+      - name: Load strong name certificate
+        run: |
+          echo "$SNC_BASE64" | base64 --decode > "${{ github.workspace }}/certificate.snk"
+        shell: bash
+        env:
+          SNC_BASE64: ${{ secrets.SNC_BASE64 }}
+
       - name: Build package
-        run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }}
+        run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}/certificate.snk
 
       - name: Upload nuget packages
         uses: actions/upload-artifact@v4
@@ -46,7 +53,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Execute tests
-        run: dotnet test --framework net8.0 Source/MQTTnet.Tests/MQTTnet.Tests.csproj
+        run: dotnet test --framework net8.0 --configuration Release Source/MQTTnet.Tests/MQTTnet.Tests.csproj
 
   sign:
     needs: build

Directory.Build.props

@@ -0,0 +1,9 @@
+<Project>
+
+  <PropertyGroup>
+    <TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
+    <TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
+    <EnableMSTestRunner>true</EnableMSTestRunner>
+  </PropertyGroup>
+
+</Project>

MQTTnet.sln.DotSettings

@@ -1,4 +1,8 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
+	<s:String x:Key="/Default/CodeInspection/Highlighting/InspectionSeverities/=ConvertIfStatementToSwitchStatement/@EntryIndexedValue">DO_NOT_SHOW</s:String>
+	<s:String x:Key="/Default/CodeInspection/Highlighting/InspectionSeverities/=InconsistentNaming/@EntryIndexedValue">SUGGESTION</s:String>
+	<s:String x:Key="/Default/CodeInspection/Highlighting/InspectionSeverities/=UncatchableException/@EntryIndexedValue">DO_NOT_SHOW</s:String>
+	<s:String x:Key="/Default/CodeInspection/Highlighting/InspectionSeverities/=UnthrowableException/@EntryIndexedValue">DO_NOT_SHOW</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_FOR/@EntryValue">Required</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_FOREACH/@EntryValue">Required</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_IFELSE/@EntryValue">Required</s:String>

README.md

@@ -3,9 +3,9 @@
 <br/>
 </p>
 
-[![NuGet Badge](https://buildstats.info/nuget/MQTTnet)](https://www.nuget.org/packages/MQTTnet)
+[![NuGet Badge](https://img.shields.io/nuget/dt/MQTTnet)](https://www.nuget.org/packages/MQTTnet)
 [![CI](https://github.com/dotnet/MQTTnet/actions/workflows/ci.yml/badge.svg?branch=master)](https://github.com/dotnet/MQTTnet/actions/workflows/ci.yml)
-[![MyGet](https://img.shields.io/myget/mqttnet/v/mqttnet?color=orange&label=MyGet-Preview)](https://www.myget.org/feed/mqttnet/package/nuget/MQTTnet)
+[![MyGet](https://img.shields.io/myget/mqttnet/v/mqttnet?color=orange&label=preview)](https://www.myget.org/feed/mqttnet/package/nuget/MQTTnet)
 ![Size](https://img.shields.io/github/repo-size/dotnet/MQTTnet.svg)
 [![Join the chat at https://gitter.im/MQTTnet/community](https://badges.gitter.im/MQTTnet/community.svg)](https://gitter.im/MQTTnet/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 [![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://raw.githubusercontent.com/dotnet/MQTTnet/master/LICENSE)

Samples/Client/Client_Connection_Samples.cs

@@ -413,7 +413,7 @@ m/XriWr/Cq4h/JfB7NTsezVslgkBaoU=
             if (e.ClientWasConnected)
             {
                 // Use the current options as the new options.
-                await mqttClient.ConnectAsync(mqttClient.Options);
+                await mqttClient.ConnectAsync(mqttClientOptions);
             }
         };
 

Samples/Client/Client_Publish_Samples.cs

@@ -23,8 +23,7 @@ public static class Client_Publish_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder()
             .WithTcpServer("broker.hivemq.com")
             .Build();
@@ -42,7 +41,6 @@ public static class Client_Publish_Samples
 
         Console.WriteLine("MQTT application message is published.");
     }
-    }
 
     public static async Task Publish_Multiple_Application_Messages()
     {
@@ -54,8 +52,7 @@ public static class Client_Publish_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder()
             .WithTcpServer("broker.hivemq.com")
             .Build();
@@ -88,4 +85,3 @@ public static class Client_Publish_Samples
         Console.WriteLine("MQTT application message is published.");
     }
 }
-}

Samples/Client/Client_Subscribe_Samples.cs

@@ -26,8 +26,7 @@ public static class Client_Subscribe_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         // Setup message handling before connecting so that queued messages
@@ -52,7 +51,6 @@ public static class Client_Subscribe_Samples
         Console.WriteLine("Press enter to exit.");
         Console.ReadLine();
     }
-    }
 
     public static async Task Send_Responses()
     {
@@ -62,8 +60,7 @@ public static class Client_Subscribe_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         mqttClient.ApplicationMessageReceivedAsync += delegate(MqttApplicationMessageReceivedEventArgs args)
         {
             // Do some work with the message...
@@ -92,7 +89,6 @@ public static class Client_Subscribe_Samples
         // The response contains additional data sent by the server after subscribing.
         response.DumpToConsole();
     }
-    }
 
     public static async Task Subscribe_Multiple_Topics()
     {
@@ -102,8 +98,7 @@ public static class Client_Subscribe_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
@@ -123,7 +118,6 @@ public static class Client_Subscribe_Samples
         // The response contains additional data sent by the server after subscribing.
         response.DumpToConsole();
     }
-    }
 
     public static async Task Subscribe_Topic()
     {
@@ -133,8 +127,7 @@ public static class Client_Subscribe_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
@@ -148,7 +141,6 @@ public static class Client_Subscribe_Samples
         // The response contains additional data sent by the server after subscribing.
         response.DumpToConsole();
     }
-    }
 
     static void ConcurrentProcessingDisableAutoAcknowledge(CancellationToken shutdownToken, IMqttClient mqttClient)
     {

Samples/Diagnostics/Logger_Samples.cs

@@ -26,8 +26,7 @@ public static class Logger_Samples
             .WithTcpServer("broker.hivemq.com")
             .Build();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
         Console.WriteLine("MQTT client is connected.");
@@ -37,7 +36,6 @@ public static class Logger_Samples
 
         await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
     }
-    }
 
     public static async Task Use_Event_Logger()
     {
@@ -50,7 +48,7 @@ public static class Logger_Samples
         // The logger ID is optional but can be set do distinguish different logger instances.
         var mqttEventLogger = new MqttNetEventLogger("MyCustomLogger");
 
-        mqttEventLogger.LogMessagePublished += (sender, args) =>
+        mqttEventLogger.LogMessagePublished += (_, args) =>
         {
             var output = new StringBuilder();
             output.AppendLine($">> [{args.LogMessage.Timestamp:O}] [{args.LogMessage.ThreadId}] [{args.LogMessage.Source}] [{args.LogMessage.Level}]: {args.LogMessage.Message}");
@@ -68,8 +66,7 @@ public static class Logger_Samples
             .WithTcpServer("broker.hivemq.com")
             .Build();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
         Console.WriteLine("MQTT client is connected.");
@@ -79,7 +76,6 @@ public static class Logger_Samples
 
         await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
     }
-    }
 
     sealed class MyLogger : IMqttNetLogger
     {

Samples/Diagnostics/PackageInspection_Samples.cs

@@ -20,8 +20,7 @@ public static class PackageInspection_Samples
 
         var mqttFactory = new MqttClientFactory();
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
             .WithTcpServer("broker.hivemq.com")
             .Build();
@@ -37,7 +36,6 @@ public static class PackageInspection_Samples
 
         await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
     }
-    }
 
     static Task OnInspectPacket(InspectMqttPacketEventArgs eventArgs)
     {

Samples/MQTTnet.Samples.csproj

@@ -15,6 +15,7 @@
         <NuGetAudit>true</NuGetAudit>
         <NuGetAuditLevel>low</NuGetAuditLevel>
         <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <LangVersion>default</LangVersion>
     </PropertyGroup>
 
     <ItemGroup>

Samples/RpcClient/RpcClient_Samples.cs

@@ -26,8 +26,7 @@ public static class RpcClient_Samples
         // The RPC client is an addon for the existing client. So we need a regular client
         // which is wrapped later.
 
-        using (var mqttClient = mqttFactory.CreateMqttClient())
+        using var mqttClient = mqttFactory.CreateMqttClient();
-        {
         var mqttClientOptions = new MqttClientOptionsBuilder()
             .WithTcpServer("broker.hivemq.com")
             .Build();
@@ -43,7 +42,6 @@ public static class RpcClient_Samples
 
         Console.WriteLine("The RPC call was successful.");
     }
-    }
 
     /*
      * The device must respond to the request using the correct topic. The following C code shows how a

Samples/Server/Server_Diagnostics_Samples.cs

@@ -25,8 +25,7 @@ public static class Server_Diagnostics_Samples
         var mqttServerFactory = new MqttServerFactory();
         var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         // Attach the event handler.
         mqttServer.ClientAcknowledgedPublishPacketAsync += e =>
         {
@@ -37,7 +36,7 @@ public static class Server_Diagnostics_Samples
             Console.WriteLine($"QoS 1 reason code: {qos1AcknowledgePacket?.ReasonCode}");
 
             var qos2AcknowledgePacket = e.AcknowledgePacket as MqttPubCompPacket;
-                Console.WriteLine($"QoS 2 reason code: {qos1AcknowledgePacket?.ReasonCode}");
+            Console.WriteLine($"QoS 2 reason code: {qos2AcknowledgePacket?.ReasonCode}");
             return CompletedTask.Instance;
         };
 
@@ -49,4 +48,3 @@ public static class Server_Diagnostics_Samples
         await mqttServer.StopAsync();
     }
 }
-}

Samples/Server/Server_Intercepting_Samples.cs

@@ -25,8 +25,7 @@ public static class Server_Intercepting_Samples
         var mqttServerFactory = new MqttServerFactory();
         var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         mqttServer.InterceptingPublishAsync += args =>
         {
             // Here we only change the topic of the received application message.
@@ -44,4 +43,3 @@ public static class Server_Intercepting_Samples
         await mqttServer.StopAsync();
     }
 }
-}

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -29,8 +29,7 @@ public static class Server_Retained_Messages_Samples
         // Due to security reasons the "default" endpoint (which is unencrypted) is not enabled by default!
         var mqttServerOptions = mqttServerFactory.CreateServerOptionsBuilder().WithDefaultEndpoint().Build();
 
-        using (var server = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var server = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         // Make sure that the server will load the retained messages.
         server.LoadingRetainedMessageAsync += async eventArgs =>
         {
@@ -87,7 +86,6 @@ public static class Server_Retained_Messages_Samples
         Console.WriteLine("Press Enter to exit.");
         Console.ReadLine();
     }
-    }
 
     sealed class MqttRetainedMessageModel
     {

Samples/Server/Server_Simple_Samples.cs

@@ -22,8 +22,7 @@ public static class Server_Simple_Samples
          * See _Run_Minimal_Server_ for more information.
          */
 
-        using (var mqttServer = await StartMqttServer())
+        using var mqttServer = await StartMqttServer();
-        {
         // Let the client connect.
         await Task.Delay(TimeSpan.FromSeconds(5));
 
@@ -34,7 +33,6 @@ public static class Server_Simple_Samples
             await affectedClient.DisconnectAsync();
         }
     }
-    }
 
     public static async Task Publish_Message_From_Broker()
     {
@@ -44,8 +42,7 @@ public static class Server_Simple_Samples
          * See _Run_Minimal_Server_ for more information.
          */
 
-        using (var mqttServer = await StartMqttServer())
+        using var mqttServer = await StartMqttServer();
-        {
         // Create a new message using the builder as usual.
         var message = new MqttApplicationMessageBuilder().WithTopic("HelloWorld").WithPayload("Test").Build();
 
@@ -56,7 +53,6 @@ public static class Server_Simple_Samples
                 SenderClientId = "SenderClientId"
             });
     }
-    }
 
     public static async Task Run_Minimal_Server()
     {
@@ -77,8 +73,7 @@ public static class Server_Simple_Samples
         //     .WithDefaultEndpointPort(1234)
         //     .Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         await mqttServer.StartAsync();
 
         Console.WriteLine("Press Enter to exit.");
@@ -87,7 +82,6 @@ public static class Server_Simple_Samples
         // Stop and dispose the MQTT server if it is no longer needed!
         await mqttServer.StopAsync();
     }
-    }
 
     public static async Task Run_Server_With_Logging()
     {
@@ -103,8 +97,7 @@ public static class Server_Simple_Samples
 
         var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         await mqttServer.StartAsync();
 
         Console.WriteLine("Press Enter to exit.");
@@ -113,7 +106,6 @@ public static class Server_Simple_Samples
         // Stop and dispose the MQTT server if it is no longer needed!
         await mqttServer.StopAsync();
     }
-    }
 
     public static async Task Validating_Connections()
     {
@@ -127,8 +119,7 @@ public static class Server_Simple_Samples
 
         var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         // Setup connection validation before starting the server so that there is
         // no change to connect without valid credentials.
         mqttServer.ValidatingConnectionAsync += e =>
@@ -158,7 +149,6 @@ public static class Server_Simple_Samples
 
         await mqttServer.StopAsync();
     }
-    }
 
     static async Task<MqttServer> StartMqttServer()
     {

Samples/Server/Server_TLS_Samples.cs

@@ -31,8 +31,7 @@ public static class Server_TLS_Samples
 
         var mqttServerOptions = new MqttServerOptionsBuilder().WithEncryptionCertificate(certificate).WithEncryptedEndpoint().Build();
 
-        using (var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions))
+        using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
-        {
         await mqttServer.StartAsync();
 
         Console.WriteLine("Press Enter to exit.");
@@ -41,7 +40,6 @@ public static class Server_TLS_Samples
         // Stop and dispose the MQTT server if it is no longer needed!
         await mqttServer.StopAsync();
     }
-    }
 
     static X509Certificate2 CreateSelfSignedCertificate(string oid)
     {
@@ -50,8 +48,7 @@ public static class Server_TLS_Samples
         sanBuilder.AddIpAddress(IPAddress.IPv6Loopback);
         sanBuilder.AddDnsName("localhost");
 
-        using (var rsa = RSA.Create())
+        using var rsa = RSA.Create();
-        {
         var certRequest = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA512, RSASignaturePadding.Pkcs1);
 
         certRequest.CertificateExtensions.Add(
@@ -61,8 +58,7 @@ public static class Server_TLS_Samples
 
         certRequest.CertificateExtensions.Add(sanBuilder.Build());
 
-            using (var certificate = certRequest.CreateSelfSigned(DateTimeOffset.Now.AddMinutes(-10), DateTimeOffset.Now.AddMinutes(10)))
+        using var certificate = certRequest.CreateSelfSigned(DateTimeOffset.Now.AddMinutes(-10), DateTimeOffset.Now.AddMinutes(10));
-            {
         var pfxCertificate = new X509Certificate2(
             certificate.Export(X509ContentType.Pfx),
             (string)null!,
@@ -71,5 +67,3 @@ public static class Server_TLS_Samples
         return pfxCertificate;
     }
 }
-    }
-}

Source/MQTTnet.AspTestApp/Pages/Index.cshtml.cs

@@ -4,11 +4,11 @@
 
 using Microsoft.AspNetCore.Mvc.RazorPages;
 
-namespace MQTTnet.AspTestApp.Pages
+namespace MQTTnet.AspTestApp.Pages;
-{
+
 public class IndexModel : PageModel
 {
-        private readonly ILogger<IndexModel> _logger;
+    readonly ILogger<IndexModel> _logger;
 
     public IndexModel(ILogger<IndexModel> logger)
     {
@@ -17,7 +17,6 @@ namespace MQTTnet.AspTestApp.Pages
 
     public void OnGet()
     {
-
+        _logger.LogDebug("OnGet");
-        }
     }
 }

Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs

@@ -33,10 +33,8 @@ public static class ApplicationBuilderExtensions
                 }
 
                 var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
-                using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol).ConfigureAwait(false))
+                using var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol).ConfigureAwait(false);
-                {
                 await adapter.RunWebSocketConnectionAsync(webSocket, context);
-                }
             });
 
         return app;

Source/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs

@@ -4,8 +4,8 @@
 
 using Microsoft.AspNetCore.Connections;
 
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
-{
+
 public static class ConnectionBuilderExtensions
 {
     public static IConnectionBuilder UseMqtt(this IConnectionBuilder builder)
@@ -13,4 +13,3 @@ namespace MQTTnet.AspNetCore
         return builder.UseConnectionHandler<MqttConnectionHandler>();
     }
 }
-}

Source/MQTTnet.AspnetCore/EndpointRouterExtensions.cs

@@ -6,8 +6,8 @@ using System;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Routing;
 
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
-{
+
 public static class EndpointRouterExtensions
 {
     public static void MapMqtt(this IEndpointRouteBuilder endpoints, string pattern)
@@ -20,5 +20,3 @@ namespace MQTTnet.AspNetCore
         });
     }
 }
-}
-

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -26,7 +26,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseExpression>MIT</PackageLicenseExpression>
+        <PackageLicenseFile>LICENSE</PackageLicenseFile>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
@@ -38,13 +38,14 @@
         <NuGetAuditLevel>low</NuGetAuditLevel>
         <NuGetAuditLevel>low</NuGetAuditLevel>
         <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <IsAotCompatible>true</IsAotCompatible>
+        <LangVersion>default</LangVersion>
+        <Nullable>enable</Nullable>
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
-            <Pack>True</Pack>
+        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
 
     <ItemGroup>

Source/MQTTnet.AspnetCore/MqttClientConnectionContextFactory.cs

@@ -7,8 +7,8 @@ using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Formatter;
 
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
-{
+
 public sealed class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
 {
     public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
@@ -31,4 +31,3 @@ namespace MQTTnet.AspNetCore
         }
     }
 }
-}

Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs

@@ -33,15 +33,13 @@ public sealed class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapte
         }
 
         var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax));
-        using (var adapter = new MqttConnectionContext(formatter, connection))
+        using var adapter = new MqttConnectionContext(formatter, connection);
-        {
         var clientHandler = ClientHandler;
         if (clientHandler != null)
         {
             await clientHandler(adapter).ConfigureAwait(false);
         }
     }
-    }
 
     public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
     {

Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs

@@ -43,12 +43,10 @@ public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter
                 var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535));
                 var channel = new MqttWebSocketChannel(webSocket, remoteEndPoint, isSecureConnection, clientCertificate);
 
-                using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger))
+                using var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger);
-                {
                 await clientHandler(channelAdapter).ConfigureAwait(false);
             }
         }
-        }
         finally
         {
             clientCertificate?.Dispose();

Source/MQTTnet.AspnetCore/SocketAwaitable.cs

@@ -12,34 +12,28 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore;
 
-public class SocketAwaitable : ICriticalNotifyCompletion
+public class SocketAwaitable(PipeScheduler ioScheduler) : ICriticalNotifyCompletion
 {
-    static readonly Action _callbackCompleted = () =>
+    static readonly Action CallbackCompleted = () =>
     {
     };
 
-    readonly PipeScheduler _ioScheduler;
     int _bytesTransferred;
 
     Action _callback;
     SocketError _error;
 
-    public SocketAwaitable(PipeScheduler ioScheduler)
+    public bool IsCompleted => ReferenceEquals(_callback, CallbackCompleted);
-    {
-        _ioScheduler = ioScheduler;
-    }
-
-    public bool IsCompleted => ReferenceEquals(_callback, _callbackCompleted);
 
     public void Complete(int bytesTransferred, SocketError socketError)
     {
         _error = socketError;
         _bytesTransferred = bytesTransferred;
-        var continuation = Interlocked.Exchange(ref _callback, _callbackCompleted);
+        var continuation = Interlocked.Exchange(ref _callback, CallbackCompleted);
 
         if (continuation != null)
         {
-            _ioScheduler.Schedule(state => ((Action)state)(), continuation);
+            ioScheduler.Schedule(state => ((Action)state)(), continuation);
         }
     }
 
@@ -50,7 +44,7 @@ public class SocketAwaitable : ICriticalNotifyCompletion
 
     public int GetResult()
     {
-        Debug.Assert(ReferenceEquals(_callback, _callbackCompleted));
+        Debug.Assert(ReferenceEquals(_callback, CallbackCompleted));
 
         _callback = null;
 
@@ -64,7 +58,7 @@ public class SocketAwaitable : ICriticalNotifyCompletion
 
     public void OnCompleted(Action continuation)
     {
-        if (ReferenceEquals(_callback, _callbackCompleted) || ReferenceEquals(Interlocked.CompareExchange(ref _callback, continuation, null), _callbackCompleted))
+        if (ReferenceEquals(_callback, CallbackCompleted) || ReferenceEquals(Interlocked.CompareExchange(ref _callback, continuation, null), CallbackCompleted))
         {
             Task.Run(continuation);
         }

Source/MQTTnet.AspnetCore/SocketConnection.cs

@@ -18,11 +18,11 @@ namespace MQTTnet.AspNetCore;
 public sealed class SocketConnection : ConnectionContext
 {
     readonly EndPoint _endPoint;
+
     volatile bool _aborted;
     IDuplexPipe _application;
     SocketReceiver _receiver;
     SocketSender _sender;
-
     Socket _socket;
 
     public SocketConnection(EndPoint endPoint)
@@ -40,10 +40,13 @@ public sealed class SocketConnection : ConnectionContext
     }
 
     public override string ConnectionId { get; set; }
+
     public override IFeatureCollection Features { get; }
 
     public bool IsConnected { get; private set; }
+
     public override IDictionary<object, object> Items { get; set; }
+
     public override IDuplexPipe Transport { get; set; }
 
     public override ValueTask DisposeAsync()
@@ -78,7 +81,7 @@ public sealed class SocketConnection : ConnectionContext
         IsConnected = true;
     }
 
-    Exception ConnectionAborted()
+    static Exception ConnectionAborted()
     {
         return new MqttCommunicationException("Connection Aborted");
     }
@@ -95,8 +98,7 @@ public sealed class SocketConnection : ConnectionContext
         {
             error = new MqttCommunicationException(ex);
         }
-        catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted || ex.SocketErrorCode == SocketError.ConnectionAborted ||
+        catch (SocketException ex) when (ex.SocketErrorCode is SocketError.OperationAborted or SocketError.ConnectionAborted or SocketError.Interrupted or SocketError.InvalidArgument)
-                                         ex.SocketErrorCode == SocketError.Interrupted || ex.SocketErrorCode == SocketError.InvalidArgument)
         {
             if (!_aborted)
             {
@@ -123,10 +125,10 @@ public sealed class SocketConnection : ConnectionContext
         {
             if (_aborted)
             {
-                error = error ?? ConnectionAborted();
+                error ??= ConnectionAborted();
             }
 
-            _application.Output.Complete(error);
+            await _application.Output.CompleteAsync(error);
         }
     }
 

Source/MQTTnet.Benchmarks/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+[assembly: DoNotParallelize]

Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs

@@ -9,8 +9,8 @@ using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [MemoryDiagnoser]
 public class AsyncLockBenchmark : BaseBenchmark
@@ -52,8 +52,7 @@ namespace MQTTnet.Benchmarks
     {
         var asyncLock = new AsyncLock();
 
-            using (var cancellationToken = new CancellationTokenSource())
+        using var cancellationToken = new CancellationTokenSource();
-            {
         for (var i = 0; i < 100000; i++)
         {
             using (await asyncLock.EnterAsync(cancellationToken.Token).ConfigureAwait(false))
@@ -62,5 +61,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-    }
-}

Source/MQTTnet.Benchmarks/BaseBenchmark.cs

@@ -2,9 +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.
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 public abstract class BaseBenchmark
 {
 }
-}

Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -12,8 +12,8 @@ using MQTTnet.Formatter;
 using MQTTnet.Packets;
 using MQTTnet.Tests.Mockups;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [MemoryDiagnoser]
 public sealed class ChannelAdapterBenchmark : BaseBenchmark
 {
@@ -87,4 +87,3 @@ namespace MQTTnet.Benchmarks
         return buffer.ToArray();
     }
 }
-}

Source/MQTTnet.Benchmarks/LoggerBenchmark.cs

@@ -2,12 +2,13 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Diagnostics;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Diagnostics.Logger;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter]
 [MemoryDiagnoser]
@@ -42,7 +43,7 @@ namespace MQTTnet.Benchmarks
     {
         if (_useHandler)
         {
-                eventArgs.LogMessage.ToString();
+            Debug.WriteLine(eventArgs.LogMessage.ToString());
         }
     }
 
@@ -88,4 +89,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-}

Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -18,7 +18,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="BenchmarkDotNet" Version="0.14.0" />
+        <PackageReference Include="BenchmarkDotNet" Version="0.15.4" />
     </ItemGroup>
 
     <ItemGroup>

Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs

@@ -3,16 +3,17 @@ using BenchmarkDotNet.Jobs;
 using System;
 
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter, RankColumn]
 [MemoryDiagnoser]
 public class MemoryCopyBenchmark
 {
-        const int max_length = 1024 * 8;
+    const int MaxLength = 1024 * 8;
-        private byte[] source;
+
-        private byte[] target;
+    byte[] _source;
+    byte[] _target;
 
     [Params(64 - 1, 128 - 1, 256 - 1, 512 - 1, 1024 - 1, 2048 - 1, 5096 - 1)]
     public int Length { get; set; }
@@ -20,21 +21,20 @@ namespace MQTTnet.Benchmarks
     [GlobalSetup]
     public void Setup()
     {
-            source = new byte[max_length];
+        _source = new byte[MaxLength];
-            target = new byte[max_length];
+        _target = new byte[MaxLength];
     }
 
     [Benchmark(Baseline = true)]
     public void Array_Copy()
     {
-            Array.Copy(source, 0, target, 0, Length);
+        Array.Copy(_source, 0, _target, 0, Length);
     }
 
     [Benchmark]
     public void Memory_Copy()
     {
-            MQTTnet.Internal.MqttMemoryHelper.Copy(source, 0, target, 0, Length);
+        Internal.MqttMemoryHelper.Copy(_source, 0, _target, 0, Length);
     }
 
 }
-}

Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs

@@ -4,7 +4,6 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
@@ -23,22 +22,19 @@ public class MessageDeliveryBenchmark : BaseBenchmark
     int _messagesExpectedCount;
     int _messagesReceivedCount;
     Dictionary<string, IMqttClient> _mqttPublisherClientsByPublisherName;
-
     MqttServer _mqttServer;
     List<IMqttClient> _mqttSubscriberClients;
+
+    [Params(1000, 10000)] public int _numPublishers;
+
+    [Params(5, 10, 20, 50)] public int _numSubscribedTopicsPerSubscriber;
+
+    [Params(10)] public int _numSubscribers;
+
+    [Params(1, 5)] public int _numTopicsPerPublisher;
     Dictionary<string, string> _publisherByTopic;
-    List<MqttApplicationMessage> _topicPublishMessages;
-
     Dictionary<string, List<string>> _topicsByPublisher;
 
-    [Params(1000, 10000)] public int NumPublishers;
-
-    [Params(5, 10, 20, 50)] public int NumSubscribedTopicsPerSubscriber;
-
-    [Params(10)] public int NumSubscribers;
-
-    [Params(1, 5)] public int NumTopicsPerPublisher;
-
     [GlobalCleanup]
     public void Cleanup()
     {
@@ -64,14 +60,11 @@ public class MessageDeliveryBenchmark : BaseBenchmark
         _mqttServer = null;
     }
 
-    /// <summary>
-    ///     Publish messages and wait for messages sent to subscribers
-    /// </summary>
     [Benchmark]
     public void DeliverMessages()
     {
         // There should be one message received per publish for each subscribed topic
-        _messagesExpectedCount = NumSubscribedTopicsPerSubscriber * NumSubscribers;
+        _messagesExpectedCount = _numSubscribedTopicsPerSubscriber * _numSubscribers;
 
         // Loop for a while and exchange messages
 
@@ -82,8 +75,6 @@ public class MessageDeliveryBenchmark : BaseBenchmark
         // same payload for all messages
         var payload = new byte[] { 1, 2, 3, 4 };
 
-        var tasks = new List<Task>();
-
         // publish a message for each subscribed topic
         foreach (var topic in _allSubscribedTopics)
         {
@@ -101,35 +92,23 @@ public class MessageDeliveryBenchmark : BaseBenchmark
         }
         catch
         {
+            // Ignore all errors.
         }
 
         _cancellationTokenSource.Dispose();
 
         if (_messagesReceivedCount < _messagesExpectedCount)
         {
-            throw new Exception(string.Format("Messages Received Count mismatch, expected {0}, received {1}", _messagesExpectedCount, _messagesReceivedCount));
+            throw new Exception($"Messages Received Count mismatch, expected {_messagesExpectedCount}, received {_messagesReceivedCount}");
         }
     }
 
-
     [GlobalSetup]
     public void Setup()
     {
         _lockMsgCount = new object();
 
-        Dictionary<string, List<string>> singleWildcardTopicsByPublisher;
+        TopicGenerator.Generate(_numPublishers, _numTopicsPerPublisher, out _topicsByPublisher, out _, out _);
-        Dictionary<string, List<string>> multiWildcardTopicsByPublisher;
-
-        TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
-
-        var topics = _topicsByPublisher.First().Value;
-        _topicPublishMessages = new List<MqttApplicationMessage>();
-        // Prepare messages, same for each publisher
-        foreach (var topic in topics)
-        {
-            var message = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[] { 1, 2, 3, 4, 5, 6, 7, 8 }).Build();
-            _topicPublishMessages.Add(message);
-        }
 
         // Create server
         var serverFactory = new MqttServerFactory();
@@ -139,7 +118,7 @@ public class MessageDeliveryBenchmark : BaseBenchmark
         _mqttServer.StartAsync().GetAwaiter().GetResult();
 
         // Create publisher clients
-        _mqttPublisherClientsByPublisherName = new Dictionary<string, IMqttClient>();
+        _mqttPublisherClientsByPublisherName = [];
         foreach (var pt in _topicsByPublisher)
         {
             var publisherName = pt.Key;
@@ -151,13 +130,13 @@ public class MessageDeliveryBenchmark : BaseBenchmark
 
         // Create subscriber clients
         _mqttSubscriberClients = new List<IMqttClient>();
-        for (var i = 0; i < NumSubscribers; i++)
+        for (var i = 0; i < _numSubscribers; i++)
         {
             var mqttSubscriberClient = clientFactory.CreateMqttClient();
             _mqttSubscriberClients.Add(mqttSubscriberClient);
 
             var subscriberOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").WithClientId("subscriber" + i).Build();
-            mqttSubscriberClient.ApplicationMessageReceivedAsync += r =>
+            mqttSubscriberClient.ApplicationMessageReceivedAsync += _ =>
             {
                 // count messages and signal cancellation when expected message count is reached
                 lock (_lockMsgCount)
@@ -171,12 +150,13 @@ public class MessageDeliveryBenchmark : BaseBenchmark
 
                 return Task.CompletedTask;
             };
+
             mqttSubscriberClient.ConnectAsync(subscriberOptions).GetAwaiter().GetResult();
         }
 
 
         var allTopics = new List<string>();
-        _publisherByTopic = new Dictionary<string, string>();
+        _publisherByTopic = [];
         foreach (var t in _topicsByPublisher)
         {
             foreach (var topic in t.Value)
@@ -189,21 +169,18 @@ public class MessageDeliveryBenchmark : BaseBenchmark
         // Subscribe to NumSubscribedTopics topics spread across all topics
         _allSubscribedTopics = new List<string>();
 
-        var totalNumTopics = NumPublishers * NumTopicsPerPublisher;
+        var totalNumTopics = _numPublishers * _numTopicsPerPublisher;
-        var topicIndexStep = totalNumTopics / (NumSubscribedTopicsPerSubscriber * NumSubscribers);
+        var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers);
-        if (topicIndexStep * NumSubscribedTopicsPerSubscriber * NumSubscribers != totalNumTopics)
+        if (topicIndexStep * _numSubscribedTopicsPerSubscriber * _numSubscribers != totalNumTopics)
         {
             throw new Exception(
-                string.Format(
+                $"The total number of topics must be divisible by the number of subscribed topics across all subscribers. Total number of topics: {totalNumTopics}, topic step: {topicIndexStep}");
-                    "The total number of topics must be divisible by the number of subscribed topics across all subscribers. Total number of topics: {0}, topic step: {1}",
-                    totalNumTopics,
-                    topicIndexStep));
         }
 
         var topicIndex = 0;
         foreach (var mqttSubscriber in _mqttSubscriberClients)
         {
-            for (var i = 0; i < NumSubscribedTopicsPerSubscriber; ++i, topicIndex += topicIndexStep)
+            for (var i = 0; i < _numSubscribedTopicsPerSubscriber; ++i, topicIndex += topicIndexStep)
             {
                 var topic = allTopics[topicIndex];
                 _allSubscribedTopics.Add(topic);

Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs

@@ -9,8 +9,8 @@ using Microsoft.AspNetCore.Hosting;
 using MQTTnet.AspNetCore;
 using MQTTnet.Diagnostics.Logger;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [MemoryDiagnoser]
 public class MessageProcessingMqttConnectionContextBenchmark : BaseBenchmark
@@ -30,7 +30,7 @@ namespace MQTTnet.Benchmarks
                     .AddMqttConnectionHandler();
             })
             .Configure(app => {
-                       app.UseMqttServer(s => {
+                app.UseMqttServer(_ => {
 
                 });
             })
@@ -70,4 +70,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-}

Source/MQTTnet.Benchmarks/MqttBufferReaderBenchmark.cs

@@ -8,8 +8,8 @@ using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Formatter;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [MemoryDiagnoser]
 public class MqttBufferReaderBenchmark
@@ -40,4 +40,3 @@ namespace MQTTnet.Benchmarks
         Encoding.UTF8.GetString(_buffer, 0, _bufferLength);
     }
 }
-}

Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs

@@ -8,8 +8,8 @@ using BenchmarkDotNet.Jobs;
 using MQTTnet.Formatter;
 using MQTTnet.Tests.Mockups;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [MemoryDiagnoser]
 public class MqttPacketReaderWriterBenchmark : BaseBenchmark
@@ -93,4 +93,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-}

Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -2,7 +2,6 @@
 // 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.Buffers;
 using System.Reflection;
 using System.Threading;

Source/MQTTnet.Benchmarks/Program.cs

@@ -5,12 +5,10 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using BenchmarkDotNet.Loggers;
 using BenchmarkDotNet.Running;
-using MQTTnet.Diagnostics;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 public static class Program
 {
     static int _selectedBenchmarkIndex;
@@ -139,4 +137,3 @@ namespace MQTTnet.Benchmarks
         Console.WriteLine();
     }
 }
-}

Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs

@@ -1,31 +1,27 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
-using MQTTnet.Adapter;
 using MQTTnet.AspNetCore;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
-using MQTTnet.Packets;
-using System;
 using System.Buffers;
 using System.IO;
 using System.IO.Pipelines;
-using System.Linq;
 using System.Threading.Tasks;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter, RankColumn]
 [MemoryDiagnoser]
 public class ReaderExtensionsBenchmark
 {
-        MqttPacketFormatterAdapter mqttPacketFormatter;
+    MqttPacketFormatterAdapter _mqttPacketFormatter;
-        MemoryStream stream;
+    MemoryStream _stream;
 
     [GlobalSetup]
     public void GlobalSetup()
     {
-            mqttPacketFormatter = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
+        _mqttPacketFormatter = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
         var mqttMessage = new MqttApplicationMessageBuilder()
             .WithTopic("topic")
             .WithPayload(new byte[10 * 1024])
@@ -33,66 +29,18 @@ namespace MQTTnet.Benchmarks
 
         var packet = MqttPublishPacketFactory.Create(mqttMessage);
 
-            var buffer = mqttPacketFormatter.Encode(packet);
+        var buffer = _mqttPacketFormatter.Encode(packet);
-            stream = new MemoryStream();
+        _stream = new MemoryStream();
-            stream.Write(buffer.Packet);
+        _stream.Write(buffer.Packet);
-            stream.Write(buffer.Payload.ToArray());
+        _stream.Write(buffer.Payload.ToArray());
-            mqttPacketFormatter.Cleanup();
+        _mqttPacketFormatter.Cleanup();
-        }
-
-        [Benchmark(Baseline = true)]
-        public async Task Before()
-        {
-            stream.Position = 0;
-            var input = PipeReader.Create(stream);
-
-            while (true)
-            {
-                ReadResult readResult;
-                var readTask = input.ReadAsync();
-                if (readTask.IsCompleted)
-                {
-                    readResult = readTask.Result;
-                }
-                else
-                {
-                    readResult = await readTask.ConfigureAwait(false);
-                }
-
-                var buffer = readResult.Buffer;
-
-                var consumed = buffer.Start;
-                var observed = buffer.Start;
-
-                try
-                {
-                    if (!buffer.IsEmpty)
-                    {
-                        if (ReaderExtensions_Before.TryDecode(mqttPacketFormatter, buffer, out var packet, out consumed, out observed, out var received))
-                        {
-                            break;
-                        }
-                    }
-                    else if (readResult.IsCompleted)
-                    {
-                        throw new MqttCommunicationException("Connection Aborted");
-                    }
-                }
-                finally
-                {
-                    // The buffer was sliced up to where it was consumed, so we can just advance to the start.
-                    // We mark examined as buffer.End so that if we didn't receive a full frame, we'll wait for more data
-                    // before yielding the read again.
-                    input.AdvanceTo(consumed, observed);
-                }
-            }
     }
 
     [Benchmark]
     public async Task After()
     {
-            stream.Position = 0;
+        _stream.Position = 0;
-            var input = PipeReader.Create(stream);
+        var input = PipeReader.Create(_stream);
 
         while (true)
         {
@@ -116,7 +64,7 @@ namespace MQTTnet.Benchmarks
             {
                 if (!buffer.IsEmpty)
                 {
-                        if (ReaderExtensions.TryDecode(mqttPacketFormatter, buffer, out var packet, out consumed, out observed, out var received))
+                    if (_mqttPacketFormatter.TryDecode(buffer, out _, out consumed, out observed, out _))
                     {
                         break;
                     }
@@ -135,106 +83,4 @@ namespace MQTTnet.Benchmarks
             }
         }
     }
-
-        public static class ReaderExtensions_Before
-        {
-            public static bool TryDecode(MqttPacketFormatterAdapter formatter,
-                in ReadOnlySequence<byte> input,
-                out MqttPacket packet,
-                out SequencePosition consumed,
-                out SequencePosition observed,
-                out int bytesRead)
-            {
-                ArgumentNullException.ThrowIfNull(formatter);
-
-                packet = null;
-                consumed = input.Start;
-                observed = input.End;
-                bytesRead = 0;
-                var copy = input;
-
-                if (copy.Length < 2)
-                {
-                    return false;
-                }
-
-                var fixedHeader = copy.First.Span[0];
-                if (!TryReadBodyLength(ref copy, out int headerLength, out var bodyLength))
-                {
-                    return false;
-                }
-
-                if (copy.Length < bodyLength)
-                {
-                    return false;
-                }
-
-                var bodySlice = copy.Slice(0, bodyLength);
-                var buffer = GetMemory(bodySlice).ToArray();
-
-                var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, new ArraySegment<byte>(buffer, 0, buffer.Length), buffer.Length + 2);
-
-                if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
-                {
-                    formatter.DetectProtocolVersion(receivedMqttPacket);
-                }
-
-                packet = formatter.Decode(receivedMqttPacket);
-                consumed = bodySlice.End;
-                observed = bodySlice.End;
-                bytesRead = headerLength + bodyLength;
-                return true;
-            }
-
-            static ReadOnlyMemory<byte> GetMemory(in ReadOnlySequence<byte> input)
-            {
-                if (input.IsSingleSegment)
-                {
-                    return input.First;
-                }
-
-                // Should be rare
-                return input.ToArray();
-            }
-
-            static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int headerLength, out int bodyLength)
-            {
-                // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
-                var multiplier = 1;
-                var value = 0;
-                byte encodedByte;
-                var index = 1;
-                headerLength = 0;
-                bodyLength = 0;
-
-                var temp = GetMemory(input.Slice(0, Math.Min(5, input.Length))).Span;
-
-                do
-                {
-                    if (index == temp.Length)
-                    {
-                        return false;
-                    }
-
-                    encodedByte = temp[index];
-                    index++;
-
-                    value += (byte)(encodedByte & 127) * multiplier;
-                    if (multiplier > 128 * 128 * 128)
-                    {
-                        throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())}).");
-                    }
-
-                    multiplier *= 128;
-                } while ((encodedByte & 128) != 0);
-
-                input = input.Slice(index);
-
-                headerLength = index;
-                bodyLength = value;
-                return true;
-            }
-        }
-
-    }
 }

Source/MQTTnet.Benchmarks/RoundtripProcessingBenchmark.cs

@@ -3,8 +3,8 @@ using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
 using MQTTnet.Tests.Server;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter, RankColumn]
 [MemoryDiagnoser]
@@ -24,7 +24,6 @@ namespace MQTTnet.Benchmarks
     [Benchmark]
     public void Handle_100_000_Messages_In_Receiving_Client()
     {
-            new Load_Tests().Handle_100_000_Messages_In_Receiving_Client().GetAwaiter().GetResult();
+        //new Load_Tests().Handle_100_000_Messages_In_Receiving_Client().GetAwaiter().GetResult();
-        }
     }
 }

Source/MQTTnet.Benchmarks/SendPacketAsyncBenchmark.cs

@@ -7,52 +7,51 @@ using System.IO;
 using System.IO.Pipelines;
 using System.Threading.Tasks;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter, RankColumn]
 [MemoryDiagnoser]
 public class SendPacketAsyncBenchmark : BaseBenchmark
 {
-        MemoryStream stream;
+    MemoryStream _stream;
-        MqttPacketBuffer buffer;
+    MqttPacketBuffer _buffer;
 
     [GlobalSetup]
     public void GlobalSetup()
     {
-            stream = new MemoryStream(1024);
+        _stream = new MemoryStream(1024);
         var packet = new ArraySegment<byte>(new byte[10]);
-            buffer = new MqttPacketBuffer(packet);
+        _buffer = new MqttPacketBuffer(packet);
     }
 
     [Benchmark(Baseline = true)]
     public async ValueTask Before()
     {
-            stream.Position = 0;
+        _stream.Position = 0;
-            var output = PipeWriter.Create(stream);
+        var output = PipeWriter.Create(_stream);
 
-            WritePacketBuffer(output, buffer);
+        WritePacketBuffer(output, _buffer);
         await output.FlushAsync();
     }
 
     [Benchmark]
     public async ValueTask After()
     {
-            stream.Position = 0;
+        _stream.Position = 0;
-            var output = PipeWriter.Create(stream);
+        var output = PipeWriter.Create(_stream);
 
-            if (buffer.Payload.Length == 0)
+        if (_buffer.Payload.Length == 0)
         {
-                await output.WriteAsync(buffer.Packet).ConfigureAwait(false);
+            await output.WriteAsync(_buffer.Packet).ConfigureAwait(false);
         }
         else
         {
-                WritePacketBuffer(output, buffer);
+            WritePacketBuffer(output, _buffer);
             await output.FlushAsync().ConfigureAwait(false);
         }
     }
 
-
     static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
     {
         // copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter
@@ -61,9 +60,8 @@ namespace MQTTnet.Benchmarks
         var span = output.GetSpan(buffer.Length);
 
         buffer.Packet.AsSpan().CopyTo(span);
-            buffer.Payload.CopyTo(span.Slice(buffer.Packet.Count));
+        buffer.Payload.CopyTo(span[buffer.Packet.Count..]);
 
         output.Advance(buffer.Length);
     }
 }
-}

Source/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -17,8 +17,8 @@ using MQTTnet.Diagnostics.Logger;
 using System.Buffers;
 using System.Net;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter]
 [MemoryDiagnoser]
@@ -62,7 +62,7 @@ namespace MQTTnet.Benchmarks
         {
             channel.Reset();
 
-                var header = reader.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
+            reader.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
         }
     }
 
@@ -81,7 +81,7 @@ namespace MQTTnet.Benchmarks
 
         public bool IsSecureConnection { get; } = false;
 
-            public X509Certificate2 ClientCertificate { get; }
+        public X509Certificate2 ClientCertificate { get; set; }
 
         public void Reset()
         {
@@ -100,7 +100,7 @@ namespace MQTTnet.Benchmarks
 
         public Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-                Array.Copy(_buffer.Array, _position, buffer, offset, count);
+            Array.Copy(_buffer!.Array!, _position, buffer, offset, count);
             _position += count;
 
             return Task.FromResult(count);
@@ -116,4 +116,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-}

Source/MQTTnet.Benchmarks/ServerProcessingBenchmark.cs

@@ -7,8 +7,8 @@ using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
 using MQTTnet.Tests.Server;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter, RankColumn]
 [MemoryDiagnoser]
@@ -28,13 +28,12 @@ namespace MQTTnet.Benchmarks
     [Benchmark]
     public void Handle_100_000_Messages_In_Server_MqttClient()
     {
-            new Load_Tests().Handle_100_000_Messages_In_Server().GetAwaiter().GetResult();
+        //new Load_Tests().Handle_100_000_Messages_In_Server().GetAwaiter().GetResult();
     }
 
-        //[Benchmark]
+    [Benchmark]
     public void Handle_100_000_Messages_In_Server_LowLevelMqttClient()
     {
-            new Load_Tests().Handle_100_000_Messages_In_Low_Level_Client().GetAwaiter().GetResult();
+        //new Load_Tests().Handle_100_000_Messages_In_Low_Level_Client().GetAwaiter().GetResult();
-        }
     }
 }

Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs

@@ -30,7 +30,7 @@ public class SubscribeBenchmark : BaseBenchmark
     [GlobalSetup]
     public void Setup()
     {
-        TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
+        TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out _, out _);
         _topics = topicsByPublisher.Values.First();
 
         var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();

Source/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -11,8 +11,8 @@ using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.AspNetCore;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [MemoryDiagnoser]
 public class TcpPipesBenchmark : BaseBenchmark
@@ -77,4 +77,3 @@ namespace MQTTnet.Benchmarks
         }
     }
 }
-}

Source/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs

@@ -2,19 +2,16 @@
 // 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 BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 [SimpleJob(RuntimeMoniker.Net60)]
 [RPlotExporter]
 [MemoryDiagnoser]
 public class TopicFilterComparerBenchmark : BaseBenchmark
 {
-        static readonly char[] TopicLevelSeparator = { '/' };
-
     readonly string _longTopic =
         "AAAAAAAAAAAAAssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssshhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkAAAAAAAAAAAAAAABBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
 
@@ -33,238 +30,4 @@ namespace MQTTnet.Benchmarks
             MqttTopicFilterComparer.Compare(_longTopic, _longTopic);
         }
     }
-
-        [Benchmark]
-        public void MqttTopicFilterComparer_10000_LoopMethod_Without_Pointer()
-        {
-            for (var i = 0; i < 100000; i++)
-            {
-                MqttTopicFilterComparerWithoutPointer.Compare("sport/tennis/player1", "sport/#");
-                MqttTopicFilterComparerWithoutPointer.Compare("sport/tennis/player1/ranking", "sport/#/ranking");
-                MqttTopicFilterComparerWithoutPointer.Compare("sport/tennis/player1/score/wimbledon", "sport/+/player1/#");
-                MqttTopicFilterComparerWithoutPointer.Compare("sport/tennis/player1", "sport/tennis/+");
-                MqttTopicFilterComparerWithoutPointer.Compare("/finance", "+/+");
-                MqttTopicFilterComparerWithoutPointer.Compare("/finance", "/+");
-                MqttTopicFilterComparerWithoutPointer.Compare("/finance", "+");
-                MqttTopicFilterComparerWithoutPointer.Compare(_longTopic, _longTopic);
-            }
-        }
-
-        [Benchmark]
-        public void MqttTopicFilterComparer_10000_StringSplitMethod()
-        {
-            for (var i = 0; i < 100000; i++)
-            {
-                LegacyMethodByStringSplit("sport/tennis/player1", "sport/#");
-                LegacyMethodByStringSplit("sport/tennis/player1/ranking", "sport/#/ranking");
-                LegacyMethodByStringSplit("sport/tennis/player1/score/wimbledon", "sport/+/player1/#");
-                LegacyMethodByStringSplit("sport/tennis/player1", "sport/tennis/+");
-                LegacyMethodByStringSplit("/finance", "+/+");
-                LegacyMethodByStringSplit("/finance", "/+");
-                LegacyMethodByStringSplit("/finance", "+");
-                MqttTopicFilterComparer.Compare(_longTopic, _longTopic);
-            }
-        }
-
-        [GlobalSetup]
-        public void Setup()
-        {
-        }
-
-        static bool LegacyMethodByStringSplit(string topic, string filter)
-        {
-            ArgumentNullException.ThrowIfNull(topic);
-            ArgumentNullException.ThrowIfNull(filter);
-
-            if (string.Equals(topic, filter, StringComparison.Ordinal))
-            {
-                return true;
-            }
-
-            var fragmentsTopic = topic.Split(TopicLevelSeparator, StringSplitOptions.None);
-            var fragmentsFilter = filter.Split(TopicLevelSeparator, StringSplitOptions.None);
-
-            // # > In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2].
-            for (var i = 0; i < fragmentsFilter.Length; i++)
-            {
-                if (fragmentsFilter[i] == "+")
-                {
-                    continue;
-                }
-
-                if (fragmentsFilter[i] == "#")
-                {
-                    return true;
-                }
-
-                if (i >= fragmentsTopic.Length)
-                {
-                    return false;
-                }
-
-                if (!string.Equals(fragmentsFilter[i], fragmentsTopic[i], StringComparison.Ordinal))
-                {
-                    return false;
-                }
-            }
-
-            return fragmentsTopic.Length == fragmentsFilter.Length;
-        }
-
-        public static class MqttTopicFilterComparerWithoutPointer
-        {
-            public const char LevelSeparator = '/';
-            public const char MultiLevelWildcard = '#';
-            public const char SingleLevelWildcard = '+';
-            public const char ReservedTopicPrefix = '$';
-
-            public static MqttTopicFilterCompareResult Compare(string topic, string filter)
-            {
-                if (string.IsNullOrEmpty(topic))
-                {
-                    return MqttTopicFilterCompareResult.TopicInvalid;
-                }
-
-                if (string.IsNullOrEmpty(filter))
-                {
-                    return MqttTopicFilterCompareResult.FilterInvalid;
-                }
-
-                var filterOffset = 0;
-                var filterLength = filter.Length;
-
-                var topicOffset = 0;
-                var topicLength = topic.Length;
-
-                var topicPointer = topic;
-                var filterPointer = filter;
-
-                var isMultiLevelFilter = filterPointer[filterLength - 1] == MultiLevelWildcard;
-                var isReservedTopic = topicPointer[0] == ReservedTopicPrefix;
-
-                if (isReservedTopic && filterLength == 1 && isMultiLevelFilter)
-                {
-                    // It is not allowed to receive i.e. '$foo/bar' with filter '#'.
-                    return MqttTopicFilterCompareResult.NoMatch;
-                }
-
-                if (isReservedTopic && filterPointer[0] == SingleLevelWildcard)
-                {
-                    // It is not allowed to receive i.e. '$SYS/monitor/Clients' with filter '+/monitor/Clients'.
-                    return MqttTopicFilterCompareResult.NoMatch;
-                }
-
-                if (filterLength == 1 && isMultiLevelFilter)
-                {
-                    // Filter '#' matches basically everything.
-                    return MqttTopicFilterCompareResult.IsMatch;
-                }
-
-                // Go through the filter char by char.
-                while (filterOffset < filterLength && topicOffset < topicLength)
-                {
-                    // Check if the current char is a multi level wildcard. The char is only allowed
-                    // at the very las position.
-                    if (filterPointer[filterOffset] == MultiLevelWildcard && filterOffset != filterLength - 1)
-                    {
-                        return MqttTopicFilterCompareResult.FilterInvalid;
-                    }
-
-                    if (filterPointer[filterOffset] == topicPointer[topicOffset])
-                    {
-                        if (topicOffset == topicLength - 1)
-                        {
-                            // Check for e.g. "foo" matching "foo/#"
-                            if (filterOffset == filterLength - 3 && filterPointer[filterOffset + 1] == LevelSeparator && isMultiLevelFilter)
-                            {
-                                return MqttTopicFilterCompareResult.IsMatch;
-                            }
-
-                            // Check for e.g. "foo/" matching "foo/#"
-                            if (filterOffset == filterLength - 2 && filterPointer[filterOffset] == LevelSeparator && isMultiLevelFilter)
-                            {
-                                return MqttTopicFilterCompareResult.IsMatch;
-                            }
-                        }
-
-                        filterOffset++;
-                        topicOffset++;
-
-                        // Check if the end was reached and i.e. "foo/bar" matches "foo/bar"
-                        if (filterOffset == filterLength && topicOffset == topicLength)
-                        {
-                            return MqttTopicFilterCompareResult.IsMatch;
-                        }
-
-                        var endOfTopic = topicOffset == topicLength;
-
-                        if (endOfTopic && filterOffset == filterLength - 1 && filterPointer[filterOffset] == SingleLevelWildcard)
-                        {
-                            if (filterOffset > 0 && filterPointer[filterOffset - 1] != LevelSeparator)
-                            {
-                                return MqttTopicFilterCompareResult.FilterInvalid;
-                            }
-
-                            return MqttTopicFilterCompareResult.IsMatch;
-                        }
-                    }
-                    else
-                    {
-                        if (filterPointer[filterOffset] == SingleLevelWildcard)
-                        {
-                            // Check for invalid "+foo" or "a/+foo" subscription
-                            if (filterOffset > 0 && filterPointer[filterOffset - 1] != LevelSeparator)
-                            {
-                                return MqttTopicFilterCompareResult.FilterInvalid;
-                            }
-
-                            // Check for bad "foo+" or "foo+/a" subscription
-                            if (filterOffset < filterLength - 1 && filterPointer[filterOffset + 1] != LevelSeparator)
-                            {
-                                return MqttTopicFilterCompareResult.FilterInvalid;
-                            }
-
-                            filterOffset++;
-                            while (topicOffset < topicLength && topicPointer[topicOffset] != LevelSeparator)
-                            {
-                                topicOffset++;
-                            }
-
-                            if (topicOffset == topicLength && filterOffset == filterLength)
-                            {
-                                return MqttTopicFilterCompareResult.IsMatch;
-                            }
-                        }
-                        else if (filterPointer[filterOffset] == MultiLevelWildcard)
-                        {
-                            if (filterOffset > 0 && filterPointer[filterOffset - 1] != LevelSeparator)
-                            {
-                                return MqttTopicFilterCompareResult.FilterInvalid;
-                            }
-
-                            if (filterOffset + 1 != filterLength)
-                            {
-                                return MqttTopicFilterCompareResult.FilterInvalid;
-                            }
-
-                            return MqttTopicFilterCompareResult.IsMatch;
-                        }
-                        else
-                        {
-                            // Check for e.g. "foo/bar" matching "foo/+/#".
-                            if (filterOffset > 0 && filterOffset + 2 == filterLength && topicOffset == topicLength && filterPointer[filterOffset - 1] == SingleLevelWildcard &&
-                                filterPointer[filterOffset] == LevelSeparator && isMultiLevelFilter)
-                            {
-                                return MqttTopicFilterCompareResult.IsMatch;
-                            }
-
-                            return MqttTopicFilterCompareResult.NoMatch;
-                        }
-                    }
-                }
-
-                return MqttTopicFilterCompareResult.NoMatch;
-            }
-        }
-    }
 }

Source/MQTTnet.Benchmarks/TopicGenerator.cs

@@ -1,8 +1,8 @@
 using System;
 using System.Collections.Generic;
 
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
-{
+
 public class TopicGenerator
 {
     public static void Generate(
@@ -81,4 +81,3 @@ namespace MQTTnet.Benchmarks
         topicList.Add(topic);
     }
 }
-}

Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs

@@ -14,15 +14,15 @@ public class UnsubscribeBenchmark : BaseBenchmark
 {
     const int NumPublishers = 1;
     const int NumTopicsPerPublisher = 10000;
+
     IMqttClient _mqttClient;
     MqttServer _mqttServer;
-
     List<string> _topics;
 
     [GlobalSetup]
     public void Setup()
     {
-        TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
+        TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out _, out _);
         _topics = topicsByPublisher.Values.First();
 
         var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();

Source/MQTTnet.Extensions.Rpc/DefaultMqttRpcClientTopicGenerationStrategy.cs

@@ -4,8 +4,8 @@
 
 using System;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
 {
     public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
@@ -27,4 +27,3 @@ namespace MQTTnet.Extensions.Rpc
         };
     }
 }
-}

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

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

Source/MQTTnet.Extensions.Rpc/IMqttRpcClientTopicGenerationStrategy.cs

@@ -2,10 +2,9 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public interface IMqttRpcClientTopicGenerationStrategy
 {
     MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context);
 }
-}

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

@@ -26,7 +26,7 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseExpression>MIT</PackageLicenseExpression>
+        <PackageLicenseFile>LICENSE</PackageLicenseFile>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -36,13 +36,14 @@
         <NuGetAudit>true</NuGetAudit>
         <NuGetAuditLevel>low</NuGetAuditLevel>
         <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <IsAotCompatible>true</IsAotCompatible>
+        <LangVersion>default</LangVersion>
+        <Nullable>enable</Nullable>
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\Images\nuget.png">
+        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
-            <Pack>True</Pack>
+        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
 
     <ItemGroup>

Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs

@@ -4,8 +4,8 @@
 
 using System;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public static class MqttFactoryExtensions
 {
     public static IMqttRpcClient CreateMqttRpcClient(this MqttClientFactory clientFactory, IMqttClient mqttClient)
@@ -26,4 +26,3 @@ namespace MQTTnet.Extensions.Rpc
         return new MqttRpcClient(mqttClient, rpcClientOptions);
     }
 }
-}

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -6,7 +6,6 @@ using System;
 using System.Buffers;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Exceptions;
@@ -14,14 +13,14 @@ using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class MqttRpcClient : IMqttRpcClient
 {
     readonly IMqttClient _mqttClient;
     readonly MqttRpcClientOptions _options;
 
-        readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>>();
+    readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new();
 
     public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options)
     {
@@ -43,10 +42,14 @@ namespace MQTTnet.Extensions.Rpc
         _waitingCalls.Clear();
     }
 
-        public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null)
+    public async Task<byte[]> ExecuteAsync(
-        {
+        TimeSpan timeout,
-            using (var timeoutToken = new CancellationTokenSource(timeout))
+        string methodName,
+        byte[] payload,
+        MqttQualityOfServiceLevel qualityOfServiceLevel,
+        IDictionary<string, object>? parameters = null)
     {
+        using var timeoutToken = new CancellationTokenSource(timeout);
         try
         {
             return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, parameters, timeoutToken.Token).ConfigureAwait(false);
@@ -61,9 +64,13 @@ namespace MQTTnet.Extensions.Rpc
             throw;
         }
     }
-        }
 
-        public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
+    public async Task<byte[]> ExecuteAsync(
+        string methodName,
+        byte[] payload,
+        MqttQualityOfServiceLevel qualityOfServiceLevel,
+        IDictionary<string, object>? parameters = null,
+        CancellationToken cancellationToken = default)
     {
         ArgumentNullException.ThrowIfNull(methodName);
 
@@ -85,7 +92,7 @@ namespace MQTTnet.Extensions.Rpc
 
         var requestMessageBuilder = new MqttApplicationMessageBuilder().WithTopic(requestTopic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel);
 
-            if (_mqttClient.Options.ProtocolVersion == MqttProtocolVersion.V500)
+        if (_mqttClient.Options!.ProtocolVersion == MqttProtocolVersion.V500)
         {
             requestMessageBuilder.WithResponseTopic(responseTopic);
         }
@@ -106,8 +113,7 @@ namespace MQTTnet.Extensions.Rpc
             await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
             await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
 
-                using (cancellationToken.Register(
+            using (cancellationToken.Register(() =>
-                           () =>
                    {
                        awaitable.TrySetCanceled();
                    }))
@@ -138,4 +144,3 @@ namespace MQTTnet.Extensions.Rpc
         return CompletedTask.Instance;
     }
 }
-}

Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs

@@ -8,11 +8,11 @@ using System.Text;
 using System.Threading.Tasks;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public static class MqttRpcClientExtensions
 {
-        public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null)
+    public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object>? parameters = null)
     {
         if (client == null) throw new ArgumentNullException(nameof(client));
 
@@ -21,4 +21,3 @@ namespace MQTTnet.Extensions.Rpc
         return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel, parameters);
     }
 }
-}

Source/MQTTnet.Extensions.Rpc/MqttRpcClientOptions.cs

@@ -2,10 +2,9 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class MqttRpcClientOptions
 {
     public IMqttRpcClientTopicGenerationStrategy TopicGenerationStrategy { get; set; } = new DefaultMqttRpcClientTopicGenerationStrategy();
 }
-}

Source/MQTTnet.Extensions.Rpc/MqttRpcClientOptionsBuilder.cs

@@ -4,8 +4,8 @@
 
 using System;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class MqttRpcClientOptionsBuilder
 {
     IMqttRpcClientTopicGenerationStrategy _topicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy();
@@ -25,4 +25,3 @@ namespace MQTTnet.Extensions.Rpc
         return this;
     }
 }
-}

Source/MQTTnet.Extensions.Rpc/MqttRpcTopicPair.cs

@@ -2,12 +2,11 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class MqttRpcTopicPair
 {
-        public string RequestTopic { get; set; }
+    public required string RequestTopic { get; init; }
 
-        public string ResponseTopic { get; set; }
+    public required string ResponseTopic { get; init; }
-    }
 }

Source/MQTTnet.Extensions.Rpc/TopicGenerationContext.cs

@@ -6,12 +6,16 @@ using System;
 using System.Collections.Generic;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Extensions.Rpc
+namespace MQTTnet.Extensions.Rpc;
-{
+
 public sealed class TopicGenerationContext
 {
-        public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName,
+    public TopicGenerationContext(
-            IDictionary<string, object> parameters, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        IMqttClient mqttClient,
+        MqttRpcClientOptions options,
+        string methodName,
+        IDictionary<string, object>? parameters,
+        MqttQualityOfServiceLevel qualityOfServiceLevel)
     {
         MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
         Parameters = parameters;
@@ -22,12 +26,12 @@ namespace MQTTnet.Extensions.Rpc
 
     public string MethodName { get; }
 
-        public IDictionary<string, object> Parameters { get; }
-
     public IMqttClient MqttClient { get; }
 
     public MqttRpcClientOptions Options { get; }
 
+    public IDictionary<string, object>? Parameters { get; }
+
     /// <summary>
     ///     Gets or sets the quality of service level.
     ///     The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message
@@ -39,4 +43,3 @@ namespace MQTTnet.Extensions.Rpc
     /// </summary>
     public MqttQualityOfServiceLevel QualityOfServiceLevel { get; }
 }
-}

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

@@ -28,10 +28,13 @@
         <PackageIcon>nuget.png</PackageIcon>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseExpression>MIT</PackageLicenseExpression>
+        <PackageLicenseFile>LICENSE</PackageLicenseFile>
         <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <NoWarn>1591;NETSDK1138</NoWarn>
+        <IsAotCompatible>true</IsAotCompatible>
+        <LangVersion>default</LangVersion>
+        <Nullable>enable</Nullable>
     </PropertyGroup>
 
     <ItemGroup>
@@ -39,15 +42,16 @@
             <Pack>True</Pack>
             <PackagePath>\</PackagePath>
         </None>
-        <None Include="..\..\Images\nuget.png">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
+        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
-        <PackageReference Include="System.ValueTuple" Version="4.5.0" />
+        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All" />
+        <PackageReference Include="System.ValueTuple" Version="4.6.1" />
     </ItemGroup>
 
     <ItemGroup>

Source/MQTTnet.Extensions.TopicTemplate/MqttTopicTemplate.cs

@@ -230,7 +230,7 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
             throw new ArgumentException("the topic has to match this template", nameof(topic));
         }
 
-        return parseParameterValuesInternal(topic);
+        return ParseParameterValuesInternal(topic);
     }
 
     /// <summary>
@@ -330,7 +330,7 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
         return parameters.Aggregate(this, (t, p) => t.TrySetParameter(p.parameter, p.value));
     }
 
-    IEnumerable<(string parameter, int index, string value)> parseParameterValuesInternal(string topic)
+    IEnumerable<(string parameter, int index, string value)> ParseParameterValuesInternal(string topic)
     {
         // because we have a match, we know the segment array is at least the template's length
         var segments = topic.Split(MqttTopicFilterComparer.LevelSeparator);

Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptions.cs

@@ -5,8 +5,8 @@
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class MqttServerClientDisconnectOptions
 {
     public MqttDisconnectReasonCode ReasonCode { get; set; } = MqttDisconnectReasonCode.NormalDisconnection;
@@ -29,4 +29,3 @@ namespace MQTTnet.Server
     /// </summary>
     public List<MqttUserProperty> UserProperties { get; set; }
 }
-}

Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs

@@ -5,8 +5,8 @@
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class MqttServerClientDisconnectOptionsBuilder
 {
     readonly MqttServerClientDisconnectOptions _options = new MqttServerClientDisconnectOptions();
@@ -51,4 +51,3 @@ namespace MQTTnet.Server
         return this;
     }
 }
-}

Source/MQTTnet.Server/EnhancedAuthentication/ExchangeEnhancedAuthenticationOptions.cs

@@ -8,9 +8,9 @@ namespace MQTTnet.Server.EnhancedAuthentication;
 
 public sealed class ExchangeEnhancedAuthenticationOptions
 {
-    public byte[] AuthenticationData { get; set; }
+    public byte[]? AuthenticationData { get; set; }
 
-    public string ReasonString { get; set; }
+    public string? ReasonString { get; set; }
 
-    public List<MqttUserProperty> UserProperties { get; set; }
+    public List<MqttUserProperty>? UserProperties { get; set; }
 }

Source/MQTTnet.Server/EnhancedAuthentication/ExchangeEnhancedAuthenticationResult.cs

@@ -4,9 +4,9 @@ namespace MQTTnet.Server.EnhancedAuthentication;
 
 public sealed class ExchangeEnhancedAuthenticationResult
 {
-    public string ReasonString { get; set; }
+    public string? ReasonString { get; init; }
 
-    public List<MqttUserProperty> UserProperties { get; set; }
+    public List<MqttUserProperty>? UserProperties { get; init; }
 
-    public byte[] AuthenticationData { get; set; }
+    public byte[]? AuthenticationData { get; init; }
 }

Source/MQTTnet.Server/Events/ApplicationMessageEnqueuedEventArgs.cs

@@ -2,10 +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.
 
-using System;
+namespace MQTTnet.Server;
 
-namespace MQTTnet.Server
-{
 public sealed class ApplicationMessageEnqueuedEventArgs : EventArgs
 {
     public ApplicationMessageEnqueuedEventArgs(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, bool isDropped)
@@ -24,4 +22,3 @@ namespace MQTTnet.Server
 
     public MqttApplicationMessage ApplicationMessage { get; }
 }
-}

Source/MQTTnet.Server/Events/ApplicationMessageNotConsumedEventArgs.cs

@@ -2,10 +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.
 
-using System;
+namespace MQTTnet.Server;
 
-namespace MQTTnet.Server
-{
 public sealed class ApplicationMessageNotConsumedEventArgs : EventArgs
 {
     public ApplicationMessageNotConsumedEventArgs(MqttApplicationMessage applicationMessage, string senderId)
@@ -24,4 +22,3 @@ namespace MQTTnet.Server
     /// </summary>
     public string SenderId { get; }
 }
-}

Source/MQTTnet.Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs

@@ -2,12 +2,11 @@
 // 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;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
 {
     public ClientAcknowledgedPublishPacketEventArgs(string clientId, string userName, IDictionary sessionItems, MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
@@ -49,4 +48,3 @@ namespace MQTTnet.Server
     /// </summary>
     public IDictionary SessionItems { get; }
 }
-}

Source/MQTTnet.Server/Events/ClientConnectedEventArgs.cs

@@ -2,16 +2,14 @@
 // 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;
-using System.Collections.Generic;
 using System.Net;
 using System.Text;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class ClientConnectedEventArgs : EventArgs
 {
     readonly MqttConnectPacket _connectPacket;
@@ -24,15 +22,15 @@ namespace MQTTnet.Server
         SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
     }
 
-        public byte[] AuthenticationData => _connectPacket.AuthenticationData;
+    public byte[]? AuthenticationData => _connectPacket.AuthenticationData;
 
-        public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
+    public string? AuthenticationMethod => _connectPacket.AuthenticationMethod;
 
     /// <summary>
     ///     Gets the client identifier of the connected client.
     ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
     /// </summary>
-        public string ClientId => _connectPacket.ClientId;
+    public string ClientId => _connectPacket.ClientId!;
 
     /// <summary>
     ///     Gets the endpoint of the connected client.
@@ -40,7 +38,7 @@ namespace MQTTnet.Server
     public EndPoint RemoteEndPoint { get; }
 
     [Obsolete("Use RemoteEndPoint instead.")]
-        public string Endpoint => RemoteEndPoint?.ToString();
+    public string? Endpoint => RemoteEndPoint?.ToString();
 
     /// <summary>
     ///     Gets the protocol version which is used by the connected client.
@@ -55,17 +53,16 @@ namespace MQTTnet.Server
     /// <summary>
     ///     Gets the user name of the connected client.
     /// </summary>
-        public string UserName => _connectPacket.Username;
+    public string? UserName => _connectPacket.Username;
 
     /// <summary>
     ///     Gets the password of the connected client.
     /// </summary>
-        public string Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
+    public string? Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
 
     /// <summary>
     ///     Gets the user properties sent by the client.
     ///     <remarks>MQTT 5.0.0+ feature.</remarks>
     /// </summary>
-        public List<MqttUserProperty> UserProperties => _connectPacket?.UserProperties;
+    public List<MqttUserProperty>? UserProperties => _connectPacket?.UserProperties;
-    }
 }

Source/MQTTnet.Server/Events/ClientDisconnectedEventArgs.cs

@@ -2,16 +2,14 @@
 // 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;
-using System.Collections.Generic;
 using System.Net;
 using System.Text;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class ClientDisconnectedEventArgs : EventArgs
 {
     readonly MqttConnectPacket _connectPacket;
@@ -37,24 +35,24 @@ namespace MQTTnet.Server
     ///     Gets the client identifier.
     ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
     /// </summary>
-        public string ClientId => _connectPacket.ClientId;
+    public string ClientId => _connectPacket.ClientId!;
 
     /// <summary>
     ///     Gets the user name of the client.
     /// </summary>
-        public string UserName => _connectPacket.Username;
+    public string? UserName => _connectPacket.Username;
 
     /// <summary>
     ///     Gets the password of the client.
     /// </summary>
-        public string Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
+    public string? Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
 
     public MqttClientDisconnectType DisconnectType { get; }
 
     public EndPoint RemoteEndPoint { get; }
 
     [Obsolete("Use RemoteEndPoint instead.")]
-        public string Endpoint => RemoteEndPoint?.ToString();
+    public string? Endpoint => RemoteEndPoint?.ToString();
 
     /// <summary>
     ///     Gets the reason code sent by the client.
@@ -68,7 +66,7 @@ namespace MQTTnet.Server
     ///     Only available for clean disconnects.
     ///     <remarks>MQTT 5.0.0+ feature.</remarks>
     /// </summary>
-        public string ReasonString => _disconnectPacket?.ReasonString;
+    public string? ReasonString => _disconnectPacket?.ReasonString;
 
     /// <summary>
     ///     Gets the session expiry interval sent by the client.
@@ -87,6 +85,5 @@ namespace MQTTnet.Server
     ///     Only available for clean disconnects.
     ///     <remarks>MQTT 5.0.0+ feature.</remarks>
     /// </summary>
-        public List<MqttUserProperty> UserProperties => _disconnectPacket?.UserProperties;
+    public List<MqttUserProperty>? UserProperties => _disconnectPacket?.UserProperties;
-    }
 }

Source/MQTTnet.Server/Events/ClientSubscribedTopicEventArgs.cs

@@ -2,12 +2,11 @@
 // 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;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class ClientSubscribedTopicEventArgs : EventArgs
 {
     public ClientSubscribedTopicEventArgs(string clientId, string userName, MqttTopicFilter topicFilter, IDictionary sessionItems)
@@ -40,4 +39,3 @@ namespace MQTTnet.Server
     /// </summary>
     public MqttTopicFilter TopicFilter { get; }
 }
-}

Source/MQTTnet.Server/Events/ClientUnsubscribedTopicEventArgs.cs

@@ -2,11 +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;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class ClientUnsubscribedTopicEventArgs : EventArgs
 {
     public ClientUnsubscribedTopicEventArgs(string clientId, string userName, string topicFilter, IDictionary sessionItems)
@@ -39,4 +38,3 @@ namespace MQTTnet.Server
     /// </summary>
     public string TopicFilter { get; }
 }
-}

Source/MQTTnet.Server/Events/InterceptingClientApplicationMessageEnqueueEventArgs.cs

@@ -2,10 +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.
 
-using System;
+namespace MQTTnet.Server;
 
-namespace MQTTnet.Server
-{
 public sealed class InterceptingClientApplicationMessageEnqueueEventArgs : EventArgs
 {
     public InterceptingClientApplicationMessageEnqueueEventArgs(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage)
@@ -32,4 +30,3 @@ namespace MQTTnet.Server
 
     public string SenderClientId { get; }
 }
-}

Source/MQTTnet.Server/Events/InterceptingPacketEventArgs.cs

@@ -2,14 +2,12 @@
 // 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;
 using System.Net;
-using System.Threading;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class InterceptingPacketEventArgs : EventArgs
 {
     public InterceptingPacketEventArgs(CancellationToken cancellationToken, string clientId, string userName, EndPoint remoteEndPoint, MqttPacket packet, IDictionary sessionItems)
@@ -61,4 +59,3 @@ namespace MQTTnet.Server
     /// </summary>
     public IDictionary SessionItems { get; }
 }
-}

Source/MQTTnet.Server/Events/InterceptingPublishEventArgs.cs

@@ -2,12 +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;
-using System.Threading;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class InterceptingPublishEventArgs : EventArgs
 {
     public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems)
@@ -54,4 +52,3 @@ namespace MQTTnet.Server
     /// </summary>
     public IDictionary SessionItems { get; }
 }
-}

Source/MQTTnet.Server/Events/InterceptingSubscriptionEventArgs.cs

@@ -2,14 +2,11 @@
 // 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;
-using System.Collections.Generic;
-using System.Threading;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class InterceptingSubscriptionEventArgs : EventArgs
 {
     public InterceptingSubscriptionEventArgs(
@@ -88,4 +85,3 @@ namespace MQTTnet.Server
     /// </summary>
     public List<MqttUserProperty> UserProperties { get; set; }
 }
-}

Source/MQTTnet.Server/Events/InterceptingUnsubscriptionEventArgs.cs

@@ -2,14 +2,11 @@
 // 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;
-using System.Collections.Generic;
-using System.Threading;
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
 {
     public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties)
@@ -75,4 +72,3 @@ namespace MQTTnet.Server
     /// </summary>
     public List<MqttUserProperty> UserProperties { get; set; }
 }
-}

Source/MQTTnet.Server/Events/LoadingRetainedMessagesEventArgs.cs

@@ -2,13 +2,9 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
+namespace MQTTnet.Server;
-using System.Collections.Generic;
 
-namespace MQTTnet.Server
-{
 public sealed class LoadingRetainedMessagesEventArgs : EventArgs
 {
     public List<MqttApplicationMessage> LoadedRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
 }
-}

Source/MQTTnet.Server/Events/PreparingSessionEventArgs.cs

@@ -2,13 +2,11 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
-using System.Collections.Generic;
 using MQTTnet.Packets;
 using MQTTnet.Server.Internal;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class PreparingSessionEventArgs : EventArgs
 {
     public string Id { get; set; }
@@ -48,4 +46,3 @@ namespace MQTTnet.Server
     // TODO: Use single properties. No entire will message.
     //MqttApplicationMessage WillMessage { get; set; }
 }
-}

Source/MQTTnet.Server/Events/QueueMessageOverwrittenEventArgs.cs

@@ -2,11 +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 MQTTnet.Packets;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class QueueMessageOverwrittenEventArgs : EventArgs
 {
     public QueueMessageOverwrittenEventArgs(string receiverClientId, MqttPacket packet)
@@ -19,4 +18,3 @@ namespace MQTTnet.Server
 
     public string ReceiverClientId { get; }
 }
-}

Source/MQTTnet.Server/Events/RetainedMessageChangedEventArgs.cs

@@ -2,11 +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.
 
-using System;
+namespace MQTTnet.Server;
-using System.Collections.Generic;
 
-namespace MQTTnet.Server
-{
 public sealed class RetainedMessageChangedEventArgs : EventArgs
 {
     public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List<MqttApplicationMessage> storedRetainedMessages)
@@ -22,4 +19,3 @@ namespace MQTTnet.Server
 
     public List<MqttApplicationMessage> StoredRetainedMessages { get; }
 }
-}

Source/MQTTnet.Server/Events/SessionDeletedEventArgs.cs

@@ -2,11 +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;
 
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
-{
+
 public sealed class SessionDeletedEventArgs : EventArgs
 {
     public SessionDeletedEventArgs(string id, string userName, IDictionary sessionItems)
@@ -31,4 +30,3 @@ namespace MQTTnet.Server
     /// </summary>
     public IDictionary SessionItems { get; }
 }
-}

Source/MQTTnet.Server/Exceptions/MqttPendingMessagesOverflowException.cs

@@ -0,0 +1,19 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Server.Exceptions;
+
+public class MqttPendingMessagesOverflowException : Exception
+{
+    public MqttPendingMessagesOverflowException(string sessionId, MqttPendingMessagesOverflowStrategy overflowStrategy) : base(
+        $"Send buffer max pending messages overflow occurred for session '{sessionId}'. Strategy: {overflowStrategy}.")
+    {
+        SessionId = sessionId;
+        OverflowStrategy = overflowStrategy;
+    }
+
+    public MqttPendingMessagesOverflowStrategy OverflowStrategy { get; }
+
+    public string SessionId { get; }
+}

Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerAdapter.cs

@@ -8,11 +8,11 @@ using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Server.Internal.Adapter
+namespace MQTTnet.Server.Internal.Adapter;
-{
+
 public sealed class MqttTcpServerAdapter : IMqttServerAdapter
 {
-        readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
+    readonly List<MqttTcpServerListener> _listeners = new();
 
     CancellationTokenSource _cancellationTokenSource;
 
@@ -122,4 +122,3 @@ namespace MQTTnet.Server.Internal.Adapter
         }
     }
 }
-}

Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerListener.cs

@@ -12,8 +12,8 @@ using MQTTnet.Formatter;
 using MQTTnet.Implementations;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Server.Internal.Adapter
+namespace MQTTnet.Server.Internal.Adapter;
-{
+
 public sealed class MqttTcpServerListener : IDisposable
 {
     readonly MqttNetSourceLogger _logger;
@@ -151,8 +151,7 @@ namespace MQTTnet.Server.Internal.Adapter
             {
                 if (exception is SocketException socketException)
                 {
-                        if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
+                    if (socketException.SocketErrorCode is SocketError.ConnectionAborted or SocketError.OperationAborted)
-                            socketException.SocketErrorCode == SocketError.OperationAborted)
                     {
                         continue;
                     }
@@ -166,8 +165,8 @@ namespace MQTTnet.Server.Internal.Adapter
 
     async Task TryHandleClientConnectionAsync(CrossPlatformSocket clientSocket)
     {
-            Stream stream = null;
+        Stream? stream = null;
-            EndPoint remoteEndPoint = null;
+        EndPoint? remoteEndPoint = null;
 
         try
         {
@@ -216,13 +215,11 @@ namespace MQTTnet.Server.Internal.Adapter
                 var bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax);
                 var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter);
 
-                    using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger))
+                using var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger);
-                    {
                 clientAdapter.AllowPacketFragmentation = _options.AllowPacketFragmentation;
                 await clientHandler(clientAdapter).ConfigureAwait(false);
             }
         }
-            }
         catch (Exception exception)
         {
             if (exception is ObjectDisposedException)
@@ -243,8 +240,11 @@ namespace MQTTnet.Server.Internal.Adapter
         {
             try
             {
-                    // ReSharper disable once MethodHasAsyncOverload
+                if (stream != null)
-                    stream?.Dispose();
+                {
+                    await stream.DisposeAsync().ConfigureAwait(false);
+                }
+
                 clientSocket?.Dispose();
             }
             catch (Exception disposeException)
@@ -256,4 +256,3 @@ namespace MQTTnet.Server.Internal.Adapter
         _logger.Verbose("TCP client '{0}' disconnected (Local endpoint={1})", remoteEndPoint, _localEndPoint);
     }
 }
-}

Source/MQTTnet.Server/Internal/CheckSubscriptionsResult.cs

@@ -4,8 +4,8 @@
 
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class CheckSubscriptionsResult
 {
     public static CheckSubscriptionsResult NotSubscribed { get; } = new CheckSubscriptionsResult();
@@ -18,4 +18,3 @@ namespace MQTTnet.Server.Internal
 
     public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
 }
-}

Source/MQTTnet.Server/Internal/DispatchApplicationMessageResult.cs

@@ -4,8 +4,8 @@
 
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class DispatchApplicationMessageResult
 {
     public DispatchApplicationMessageResult(int reasonCode, bool closeConnection, string reasonString, List<MqttUserProperty> userProperties)
@@ -24,4 +24,3 @@ namespace MQTTnet.Server.Internal
 
     public List<MqttUserProperty> UserProperties { get; }
 }
-}

Source/MQTTnet.Server/Internal/EnqueueDataPacketResult.cs

@@ -2,11 +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.
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public enum EnqueueDataPacketResult
 {
     Enqueued,
     Dropped
 }
-}

Source/MQTTnet.Server/Internal/Formatter/MqttUnsubAckPacketFactory.cs

@@ -4,8 +4,8 @@
 
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server.Internal.Formatter
+namespace MQTTnet.Server.Internal.Formatter;
-{
+
 public static class MqttUnsubAckPacketFactory
 {
     public static MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult)
@@ -24,4 +24,3 @@ namespace MQTTnet.Server.Internal.Formatter
         return unsubAckPacket;
     }
 }
-}

Source/MQTTnet.Server/Internal/ISubscriptionChangedNotification.cs

@@ -1,9 +1,8 @@
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public interface ISubscriptionChangedNotification
 {
     void OnSubscriptionsAdded(MqttSession clientSession, List<string> subscriptionsTopics);
 
     void OnSubscriptionsRemoved(MqttSession clientSession, List<string> subscriptionTopics);
 }
-}

Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs

@@ -189,17 +189,10 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
                         continue;
                     }
 
-                    if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
+                    if (await _eventContainer.ShouldSkipEnqueue(senderId, session.Id, applicationMessage))
                     {
-                        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
-                        await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-
-                        if (!eventArgs.AcceptEnqueue)
-                        {
-                            // Continue checking the other subscriptions
                         continue;
                     }
-                    }
 
                     var publishPacketCopy = MqttPublishPacketFactory.Create(applicationMessage);
                     publishPacketCopy.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
@@ -311,25 +304,6 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         return Task.FromResult((IList<MqttClientStatus>)result);
     }
 
-    public Task<MqttSessionStatus> GetSessionStatus(string id)
-    {
-        _sessionsManagementLock.EnterReadLock();
-        try
-        {
-            if (!_sessionsStorage.TryGetSession(id, out var session))
-            {
-                throw new InvalidOperationException($"Session with ID '{id}' not found.");
-            }
-
-            var sessionStatus = new MqttSessionStatus(session);
-            return Task.FromResult(sessionStatus);
-        }
-        finally
-        {
-            _sessionsManagementLock.ExitReadLock();
-        }
-    }
-
     public Task<IList<MqttSessionStatus>> GetSessionsStatus()
     {
         var result = new List<MqttSessionStatus>();
@@ -351,6 +325,25 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         return Task.FromResult((IList<MqttSessionStatus>)result);
     }
 
+    public Task<MqttSessionStatus> GetSessionStatus(string id)
+    {
+        _sessionsManagementLock.EnterReadLock();
+        try
+        {
+            if (!_sessionsStorage.TryGetSession(id, out var session))
+            {
+                throw new InvalidOperationException($"Session with ID '{id}' not found.");
+            }
+
+            var sessionStatus = new MqttSessionStatus(session);
+            return Task.FromResult(sessionStatus);
+        }
+        finally
+        {
+            _sessionsManagementLock.ExitReadLock();
+        }
+    }
+
     public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
     {
         MqttConnectedClient connectedClient = null;
@@ -439,12 +432,10 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
                 }
             }
 
-            using (var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
+            using var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout);
-            {
             await channelAdapter.DisconnectAsync(timeout.Token).ConfigureAwait(false);
         }
     }
-    }
 
     public void OnSubscriptionsAdded(MqttSession clientSession, List<string> topics)
     {
@@ -510,15 +501,22 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
 
         var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
 
-        if (subscribeResult.RetainedMessages != null)
+        if (subscribeResult.RetainedMessages == null)
         {
+            return;
+        }
+
         foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
         {
+            if (await _eventContainer.ShouldSkipEnqueue(string.Empty, clientId, retainedMessageMatch.ApplicationMessage))
+            {
+                continue;
+            }
+
             var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
             clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
         }
     }
-    }
 
     public Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters)
     {
@@ -669,20 +667,18 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         }
     }
 
-    async Task<MqttConnectPacket> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+    async Task<MqttConnectPacket?> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
     {
         try
         {
-            using (var timeoutToken = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
+            using var timeoutToken = new CancellationTokenSource(_options.DefaultCommunicationTimeout);
-            using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, cancellationToken))
+            using var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, cancellationToken);
-            {
             var firstPacket = await channelAdapter.ReceivePacketAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
             if (firstPacket is MqttConnectPacket connectPacket)
             {
                 return connectPacket;
             }
         }
-        }
         catch (OperationCanceledException)
         {
             _logger.Warning("Client '{0}': Connected but did not sent a CONNECT packet.", channelAdapter.RemoteEndPoint);
@@ -740,6 +736,19 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         }
     }
 
+    async Task<bool> ShouldSkipEnqueue(string senderId, string clientId, MqttApplicationMessage applicationMessage)
+    {
+        if (!_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
+        {
+            return false;
+        }
+
+        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, clientId, applicationMessage);
+        await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
+        return !eventArgs.AcceptEnqueue;
+    }
+
     async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
     {
         // TODO: Load session items from persisted sessions in the future.

Source/MQTTnet.Server/Internal/MqttClientStatistics.cs

@@ -4,8 +4,8 @@
 
 using MQTTnet.Packets;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class MqttClientStatistics
 {
     Statistics _statistics = new Statistics     // mutable struct, don't make readonly!
@@ -101,4 +101,3 @@ namespace MQTTnet.Server.Internal
         }
     }
 }
-}

Source/MQTTnet.Server/Internal/MqttClientSubscriptionsManager.cs

@@ -5,8 +5,8 @@
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class MqttClientSubscriptionsManager : IDisposable
 {
     static readonly List<uint> EmptySubscriptionIdentifiers = new List<uint>();
@@ -513,4 +513,3 @@ namespace MQTTnet.Server.Internal
         public MqttSubscription Subscription { get; set; }
     }
 }
-}

Source/MQTTnet.Server/Internal/MqttConnectedClient.cs

@@ -284,15 +284,22 @@ public sealed class MqttConnectedClient : IDisposable
             return;
         }
 
-        if (subscribeResult.RetainedMessages != null)
+        if (subscribeResult.RetainedMessages == null)
         {
+            return;
+        }
+
         foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
         {
+            if (await _eventContainer.ShouldSkipEnqueue(string.Empty, Id, retainedMessageMatch.ApplicationMessage))
+            {
+                continue;
+            }
+
             var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
             Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
         }
     }
-    }
 
     async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
     {

Source/MQTTnet.Server/Internal/MqttRetainedMessagesManager.cs

@@ -2,16 +2,15 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System.Collections.ObjectModel;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class MqttRetainedMessagesManager
 {
-        readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>(4096);
+    readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096);
-        readonly AsyncLock _storageAccessLock = new AsyncLock();
+    readonly AsyncLock _storageAccessLock = new();
 
     readonly MqttServerEventContainer _eventContainer;
     readonly MqttNetSourceLogger _logger;
@@ -62,6 +61,7 @@ namespace MQTTnet.Server.Internal
 
             lock (_messages)
             {
+                var hasHandlers = _eventContainer.RetainedMessageChangedEvent.HasHandlers;
                 var payload = applicationMessage.Payload;
                 var hasPayload = payload.Length > 0;
 
@@ -90,13 +90,15 @@ namespace MQTTnet.Server.Internal
                     _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
                 }
 
-                    if (saveIsRequired)
+                if (saveIsRequired && hasHandlers)
                 {
                     messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
                 }
             }
 
             if (saveIsRequired)
+            {
+                if (_eventContainer.RetainedMessageChangedEvent.HasHandlers)
                 {
                     using (await _storageAccessLock.EnterAsync().ConfigureAwait(false))
                     {
@@ -105,6 +107,7 @@ namespace MQTTnet.Server.Internal
                     }
                 }
             }
+        }
         catch (Exception exception)
         {
             _logger.Error(exception, "Unhandled exception while handling retained messages.");
@@ -146,4 +149,3 @@ namespace MQTTnet.Server.Internal
         }
     }
 }
-}

Source/MQTTnet.Server/Internal/MqttServerEventContainer.cs

@@ -4,52 +4,64 @@
 
 using MQTTnet.Internal;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
+
+public class MqttServerEventContainer
 {
-    public sealed class MqttServerEventContainer
+    public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new();
+
+    public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new();
+
+    public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new();
+
+    public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new();
+
+    public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new();
+
+    public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new();
+
+    public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new();
+
+    public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new();
+
+    public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new();
+
+    public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new();
+
+    public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new();
+
+    public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new();
+
+    public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new();
+
+    public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new();
+
+    public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new();
+
+    public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new();
+
+    public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new();
+
+    public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new();
+
+    public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new();
+
+    public AsyncEvent<EventArgs> StartedEvent { get; } = new();
+
+    public AsyncEvent<EventArgs> StoppedEvent { get; } = new();
+
+    public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new();
+
+    public async Task<bool> ShouldSkipEnqueue(string senderClientId, string clientId, MqttApplicationMessage applicationMessage)
     {
-        public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
+        if (!InterceptingClientEnqueueEvent.HasHandlers)
+        {
+            return false;
+        }
 
-        public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new AsyncEvent<ClientAcknowledgedPublishPacketEventArgs>();
+        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderClientId, clientId, applicationMessage);
+        await InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
 
-        public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
+        return !eventArgs.AcceptEnqueue;
-
-        public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
-
-        public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
-
-        public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
-
-        public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs>();
-        
-        public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new AsyncEvent<ApplicationMessageEnqueuedEventArgs>();
-
-        public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new AsyncEvent<QueueMessageOverwrittenEventArgs>();
-
-        public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
-
-        public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
-
-        public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
-
-        public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
-
-        public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
-
-        public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
-
-        public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
-
-        public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
-
-        public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
-
-        public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
-
-        public AsyncEvent<EventArgs> StartedEvent { get; } = new AsyncEvent<EventArgs>();
-
-        public AsyncEvent<EventArgs> StoppedEvent { get; } = new AsyncEvent<EventArgs>();
-
-        public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
     }
 }

Source/MQTTnet.Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -6,8 +6,8 @@ using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class MqttServerKeepAliveMonitor
 {
     static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
@@ -132,4 +132,3 @@ namespace MQTTnet.Server.Internal
         }
     }
 }
-}

Source/MQTTnet.Server/Internal/MqttSession.cs

@@ -6,6 +6,7 @@ using System.Collections;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server.Exceptions;
 
 namespace MQTTnet.Server.Internal;
 
@@ -111,10 +112,11 @@ public sealed class MqttSession : IDisposable
 
     public EnqueueDataPacketResult EnqueueDataPacket(MqttPacketBusItem packetBusItem)
     {
-        if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
+        if (PendingDataPacketsCount >= _serverOptions.MaxPendingMessagesPerClient)
         {
             if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
             {
+                packetBusItem.Fail(new MqttPendingMessagesOverflowException(Id, _serverOptions.PendingMessagesOverflowStrategy));
                 return EnqueueDataPacketResult.Dropped;
             }
 
@@ -123,13 +125,18 @@ public sealed class MqttSession : IDisposable
                 // Only drop from the data partition. Dropping from control partition might break the connection
                 // because the client does not receive PINGREQ packets etc. any longer.
                 var firstItem = _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
-                if (firstItem != null && _eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)
+                if (firstItem != null)
+                {
+                    firstItem.Fail(new MqttPendingMessagesOverflowException(Id, _serverOptions.PendingMessagesOverflowStrategy));
+
+                    if (_eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)
                     {
                         var eventArgs = new QueueMessageOverwrittenEventArgs(Id, firstItem.Packet);
                         _eventContainer.QueuedApplicationMessageOverwrittenEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
                 }
             }
+        }
 
         var publishPacket = (MqttPublishPacket)packetBusItem.Packet;
 

Source/MQTTnet.Server/Internal/MqttSessionsStorage.cs

@@ -4,8 +4,8 @@
 
 using System.Collections.ObjectModel;
 
-namespace MQTTnet.Server.Internal
+namespace MQTTnet.Server.Internal;
-{
+
 public sealed class MqttSessionsStorage
 {
     readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
@@ -78,4 +78,3 @@ namespace MQTTnet.Server.Internal
         _sessions[id] = session;
     }
 }
-}