Validate MQTT 5 SUBSCRIBE identifiers (#2262)

Observe packet task failures to prevent unobserved overflow exceptions

Server dispatch and session enqueue operations create packet bus items without
waiting for delivery. When the pending-message queue overflows, Fail faults
the item's completion task, which can later raise UnobservedTaskException
even if the application handles the dropped-message notification.

Observe the completion task's exception inside MqttPacketBusItem.Fail. This
keeps the task faulted and preserves the original exception for existing and
future WaitAsync callers, including DeliverApplicationMessageAsync.

Add a garbage-collection regression test for failures without a waiter and
verify that requesting the task before or after failure preserves the original
overflow exception. Re-enable unobserved-exception tracking in the existing
overflow integration tests for both drop strategies.

Before the fix, the focused run failed the new unobserved-exception test and
both enqueue overflow tests. After the fix, all nine focused cases and all
489 tests in the .NET 10 Release test suite pass.

Fixes https://github.com/dotnet/MQTTnet/issues/2271

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "5.1.0.${{github.run_number}}"
+  VERSION: "5.2.0.${{github.run_number}}"
   PACKAGE_SUFFIX: ""
 
 jobs:
@@ -17,19 +17,26 @@ jobs:
         with:
           dotnet-version: |
             8.0.x
+            10.0.x
 
       - name: Checkout code
         uses: actions/checkout@v4
 
       - name: Load strong name certificate
+        if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }}
         run: |
           echo "$SNC_BASE64" | base64 --decode > "${{ github.workspace }}/certificate.snk"
         shell: bash
         env:
           SNC_BASE64: ${{ secrets.SNC_BASE64 }}
 
-      - name: Build package
+      - name: Build package (signed)
-        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
+        if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }}
+        run: dotnet build MQTTnet.slnx --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: Build package (unsigned)
+        if: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository }}
+        run: dotnet build MQTTnet.slnx --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }}
 
       - name: Upload nuget packages
         uses: actions/upload-artifact@v4
@@ -48,24 +55,19 @@ jobs:
         with:
           dotnet-version: |
             8.0.x
+            10.0.x
 
       - name: Checkout code
         uses: actions/checkout@v4
 
       - name: Execute tests
-        run: dotnet test --framework net8.0 --configuration Release Source/MQTTnet.Tests/MQTTnet.Tests.csproj
+        run: dotnet test --framework net10.0 --configuration Release --project Source/MQTTnet.Tests/MQTTnet.Tests.csproj
 
   sign:
     needs: build
     runs-on: windows-latest # Code signing must run on a Windows agent for Authenticode signing (dll/exe)
-    if: github.repository == 'dotnet/MQTTnet'
+    if: github.repository == 'dotnet/MQTTnet' && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository)
     steps:
-      - name: Setup .NET SDK
-        uses: actions/setup-dotnet@v4
-        with:
-          dotnet-version: |
-            6.0.x
-
       - name: Download nuget packages
         uses: actions/download-artifact@v4
         with:
@@ -73,7 +75,7 @@ jobs:
           path: nugets
 
       - name: Install sign CLI tool
-        run: dotnet tool install --tool-path . sign --version 0.9.0-beta.23127.3
+        run: dotnet tool install --tool-path . sign --version 0.9.1-beta.25379.1
 
       - name: Sign nugets
         shell: pwsh
@@ -114,7 +116,7 @@ jobs:
         run: dotnet nuget push ${{ github.workspace }}/nugets/**/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate
 
   publish-nuget:
-    # Tag schema is 'v5.0.0-rc' etc.
+    # Tag schema is 'v5.1.0-rc' etc.
     if: startsWith(github.ref, 'refs/tags/v')
     needs: [sign, test]
     runs-on: ubuntu-latest

Directory.Build.props

@@ -1,6 +1,6 @@
 <Project>
     <PropertyGroup>
-        <TargetFramework>net8.0</TargetFramework>
+        <TargetFrameworks>net8.0;net10.0</TargetFrameworks>
         <Product>MQTTnet</Product>
         <Authors>The contributors of MQTTnet</Authors>
         <Company>The contributors of MQTTnet</Company>
@@ -11,6 +11,7 @@
         <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
         <EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
         <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>disable</Nullable>
 
         <NeutralLanguage>en-US</NeutralLanguage>
         <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
@@ -24,35 +25,10 @@
         <AnalysisLevel>latest</AnalysisLevel>
         <AnalysisMode>recommended</AnalysisMode>
         <AnalysisModeStyle>default</AnalysisModeStyle>
-
-        <TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
-        <TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
-        <EnableMSTestRunner>true</EnableMSTestRunner>
     </PropertyGroup>
 
     <PropertyGroup Condition="'$(GITHUB_ACTIONS)'=='true'">
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
     </PropertyGroup>
 
-    <PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
-        <PackageIcon>nuget.png</PackageIcon>
-        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
-        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
-        <RepositoryType>git</RepositoryType>
-        <IncludeSymbols>true</IncludeSymbols>
-        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <PublishRepositoryUrl>true</PublishRepositoryUrl>
-        <EmbedUntrackedSources>true</EmbedUntrackedSources>
-    </PropertyGroup>
-
-    <ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
-        <None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
-        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
-        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
-    </ItemGroup>
-
-    <ItemGroup Condition="'$(IncludeSymbols)'=='true'">
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
-    </ItemGroup>
 </Project>

Directory.Build.targets

@@ -0,0 +1,24 @@
+<Project>
+
+    <PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
+        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageIcon>nuget.png</PackageIcon>
+        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
+        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
+        <RepositoryType>git</RepositoryType>
+        <IncludeSymbols>true</IncludeSymbols>
+        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+        <PublishRepositoryUrl>true</PublishRepositoryUrl>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+    </PropertyGroup>
+
+    <ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
+        <None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
+        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
+        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
+    </ItemGroup>
+
+    <ItemGroup Condition="'$(IncludeSymbols)'=='true'">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
+    </ItemGroup>
+</Project>

MQTTnet.sln

@@ -1,95 +0,0 @@
-
-Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 17
-VisualStudioVersion = 17.0.31919.166
-MinimumVisualStudioVersion = 10.0.40219.1
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet", "Source\MQTTnet\MQTTnet.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
-	ProjectSection(SolutionItems) = preProject
-		.github\workflows\ci.yml = .github\workflows\ci.yml
-		CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
-		Directory.Build.props = Directory.Build.props
-		LICENSE = LICENSE
-		README.md = README.md
-		Source\ReleaseNotes.md = Source\ReleaseNotes.md
-	EndProjectSection
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Samples", "Samples\MQTTnet.Samples.csproj", "{71CF35F5-3327-4A91-AAF4-5340F6701771}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Source\MQTTnet.Tests\MQTTnet.Tests.csproj", "{B270F32A-9F3E-42EE-A989-813E35E29ADB}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Benchmarks", "Source\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp", "Source\MQTTnet.TestApp\MQTTnet.TestApp.csproj", "{175D5340-CC5B-4542-939D-4E7D15A0BC8D}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspTestApp", "Source\MQTTnet.AspTestApp\MQTTnet.AspTestApp.csproj", "{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.TopicTemplate", "Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj", "{374D5861-F7A6-42A2-8CBE-040846000EF9}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.PowerShell", "Source\MQTTnet.PowerShell\MQTTnet.PowerShell.csproj", "{37B129C5-0231-4EB5-AF13-77959068911C}"
-EndProject
-Global
-	GlobalSection(SolutionConfigurationPlatforms) = preSolution
-		Debug|Any CPU = Debug|Any CPU
-		Release|Any CPU = Release|Any CPU
-	EndGlobalSection
-	GlobalSection(ProjectConfigurationPlatforms) = postSolution
-		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.Build.0 = Release|Any CPU
-		{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|Any CPU.Build.0 = Release|Any CPU
-		{C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{C444E9C8-95FA-430E-9126-274129DE16CD}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{C444E9C8-95FA-430E-9126-274129DE16CD}.Release|Any CPU.Build.0 = Release|Any CPU
-		{71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|Any CPU.Build.0 = Release|Any CPU
-		{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|Any CPU.Build.0 = Release|Any CPU
-		{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|Any CPU.Build.0 = Release|Any CPU
-		{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|Any CPU.Build.0 = Release|Any CPU
-		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.Build.0 = Release|Any CPU
-		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.Build.0 = Release|Any CPU
-		{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.Build.0 = Release|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.Build.0 = Release|Any CPU
-	EndGlobalSection
-	GlobalSection(SolutionProperties) = preSolution
-		HideSolutionNode = FALSE
-	EndGlobalSection
-	GlobalSection(ExtensibilityGlobals) = postSolution
-		SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
-	EndGlobalSection
-EndGlobal

MQTTnet.slnx

@@ -0,0 +1,28 @@
+<Solution>
+  <Configurations>
+    <Platform Name="Any CPU" />
+    <Platform Name="x64" />
+    <Platform Name="x86" />
+  </Configurations>
+  <Folder Name="/Solution Items/">
+    <File Path=".github/workflows/ci.yml" />
+    <File Path="CODE-OF-CONDUCT.md" />
+    <File Path="Directory.Build.props" />
+    <File Path="Directory.Build.targets" />
+    <File Path="LICENSE" />
+    <File Path="README.md" />
+    <File Path="Source/ReleaseNotes.md" />
+  </Folder>
+  <Project Path="Samples/MQTTnet.Samples.csproj" />
+  <Project Path="Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj" />
+  <Project Path="Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj" />
+  <Project Path="Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj" />
+  <Project Path="Source/MQTTnet.Server/MQTTnet.Server.csproj" />
+  <Project Path="Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj" />
+  <Project Path="Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj" />
+  <Project Path="Source/MQTTnet.Tests/MQTTnet.Tests.csproj" />
+  <Project Path="Source/MQTTnet/MQTTnet.csproj" />
+</Solution>

README.md

@@ -12,9 +12,9 @@
 
 # MQTTnet
 
-MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (
+MQTTnet is a high performance .NET library for MQTT based communication.
-broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
+It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
-version and CPU architecture.
+It is compatible with mostly any supported .NET Framework version and CPU architecture.
 
 ## Features
 

Samples/Client/Client_Publish_Samples.cs

@@ -6,6 +6,9 @@
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
 
+using System;
+using System.Text;
+
 namespace MQTTnet.Samples.Client;
 
 public static class Client_Publish_Samples
@@ -84,4 +87,21 @@ public static class Client_Publish_Samples
 
         Console.WriteLine("MQTT application message is published.");
     }
+
+    public static MqttApplicationMessage Create_Message_With_Binary_User_Property()
+    {
+        /*
+         * MQTT v5 user properties are encoded as UTF-8 strings. When the UTF-8 payload is already available
+         * as a byte buffer, the builder APIs can avoid creating intermediate strings.
+         */
+
+        var encodedValue = Encoding.UTF8.GetBytes("sensor-01");
+
+        return new MqttApplicationMessageBuilder()
+            .WithTopic("samples/metadata/binary")
+            .WithUserProperty("client-id", encodedValue.AsMemory())
+            .WithUserProperty("checksum", new ArraySegment<byte>(encodedValue))
+            .WithPayload("metadata")
+            .Build();
+    }
 }

Samples/Client/Client_Socks5_Samples.cs

@@ -0,0 +1,86 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
+using MQTTnet.Extensions.Socks5;
+using MQTTnet.Samples.Helpers;
+
+namespace MQTTnet.Samples.Client;
+
+public static class Client_Socks5_Samples
+{
+    public static async Task Connect_Tcp_Via_Socks5_No_Auth()
+    {
+        /*
+         * This sample connects to an MQTT broker through a SOCKS5 proxy that does not
+         * require authentication. The SOCKS5 stream provider performs the proxy handshake
+         * (RFC 1928) and returns a connected stream that the MQTT TCP channel can use as if
+         * it were a direct TCP connection.
+         *
+         * Host name resolution is performed by the proxy by default (ATYP=DOMAINNAME), which
+         * is the desired behavior when the MQTT broker hostname is only reachable from inside
+         * the network the proxy lives in.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+
+        var mqttClientOptions = new MqttClientOptionsBuilder()
+            .WithTcpServer("broker.internal.example", 1883)
+            .WithSocks5Proxy(socks => socks
+                .WithHost("proxy.example.com")
+                .WithPort(1080))
+            .Build();
+
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+
+        Console.WriteLine("The MQTT client is connected through SOCKS5.");
+        response.DumpToConsole();
+
+        var disconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+        await mqttClient.DisconnectAsync(disconnectOptions, CancellationToken.None);
+    }
+
+    public static async Task Connect_Tcp_With_Tls_Via_Socks5_With_Credentials()
+    {
+        /*
+         * This sample connects to a TLS-protected MQTT broker through a SOCKS5 proxy that
+         * requires username/password authentication (RFC 1929).
+         *
+         * Note that the TLS handshake is layered on top of the SOCKS5 tunnel: the SNI used
+         * by SslStream is derived from the broker hostname (or the value passed to
+         * WithTargetHost), not from the proxy hostname.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+
+        var mqttClientOptions = new MqttClientOptionsBuilder()
+            .WithTcpServer("broker.internal.example", 8883)
+            .WithTlsOptions(tls => tls
+                .WithTargetHost("broker.internal.example"))
+            .WithSocks5Proxy(socks => socks
+                .WithHost("proxy.example.com")
+                .WithPort(1080)
+                .WithCredentials("alice", "s3cr3t")
+                .WithResolveDnsRemotely(true)
+                .WithHandshakeTimeout(TimeSpan.FromSeconds(10)))
+            .Build();
+
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+
+        Console.WriteLine("The MQTT client is connected through SOCKS5 with TLS.");
+        response.DumpToConsole();
+
+        var disconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+        await mqttClient.DisconnectAsync(disconnectOptions, CancellationToken.None);
+    }
+}

Samples/Client/Client_Subscribe_Samples.cs

@@ -7,7 +7,7 @@
 // ReSharper disable InconsistentNaming
 // ReSharper disable UnusedMember.Local
 
-using MQTTnet.Extensions.TopicTemplate;
+using System.Text;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Samples.Helpers;
@@ -16,8 +16,6 @@ namespace MQTTnet.Samples.Client;
 
 public static class Client_Subscribe_Samples
 {
-    static readonly MqttTopicTemplate sampleTemplate = new("mqttnet/samples/topic/{id}");
-
     public static async Task Handle_Received_Application_Message()
     {
         /*
@@ -42,7 +40,7 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "2")).Build();
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("mqttnet/samples/topic/2").Build();
 
         await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 
@@ -55,7 +53,7 @@ public static class Client_Subscribe_Samples
     public static async Task Send_Responses()
     {
         /*
-         * This sample subscribes to a topic and sends a response to the broker. This requires at least QoS level 1 to work!
+         * This sample subscribes to a topic and sends a detailed response to the client. This requires at least QoS level 1 to work!
          */
 
         var mqttFactory = new MqttClientFactory();
@@ -70,7 +68,7 @@ public static class Client_Subscribe_Samples
             args.ResponseReasonString = "That did not work!";
 
             // User properties require MQTT v5!
-            args.ResponseUserProperties.Add(new MqttUserProperty("My", "Data"));
+            args.ResponseUserProperties.Add(new MqttUserProperty("My", Encoding.UTF8.GetBytes("Data")));
 
             // Now the broker will resend the message again.
             return Task.CompletedTask;
@@ -80,7 +78,7 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("topic/test").Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 
@@ -106,9 +104,10 @@ public static class Client_Subscribe_Samples
         // Create the subscribe options including several topics with different options.
         // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
         var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
+            .WithTopicFilter(t => t.WithTopic("topic/1").WithAtLeastOnceQoS())
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "2"), noLocal: true)
+            .WithTopicFilter(t => t.WithTopic("topic/2").WithAtMostOnceQoS())
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
+            .WithTopicFilter(t => t.WithTopic("topic/3").WithNoLocal())
+            .WithTopicFilter(t => t.WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
             .Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -132,7 +131,9 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
+        var topicFilter = mqttFactory.CreateTopicFilterBuilder().WithTopic("mqttnet/samples/topic/2").WithAtLeastOnceQoS();
+
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter(topicFilter).Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 

Samples/Client/Client_Subscribe_TopicTemplate_Samples.cs

@@ -0,0 +1,75 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Extensions.TopicTemplate;
+using MQTTnet.Protocol;
+using MQTTnet.Samples.Helpers;
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+// ReSharper disable UnusedMember.Local
+
+namespace MQTTnet.Samples.Client;
+
+public static class Client_Subscribe_TopicTemplate_Samples
+{
+    static readonly MqttTopicTemplate SampleTemplate = new("mqttnet/samples/topic/{id}");
+
+    public static async Task Subscribe_Multiple_Topics_With_Template()
+    {
+        /*
+         * This sample subscribes to several topics in a single request.
+         *
+         * This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+        var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+        // Create the subscribe options including several topics with different options.
+        // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "1"))
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "2"), noLocal: true)
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
+            .Build();
+
+        var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+
+        Console.WriteLine("MQTT client subscribed to topics.");
+
+        // The response contains additional data sent by the server after subscribing.
+        response.DumpToConsole();
+    }
+
+    public static async Task Subscribe_Topic_With_Template()
+    {
+        /*
+         * This sample subscribes to a topic.
+         *
+         * This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+        var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(SampleTemplate.WithParameter("id", "1")).Build();
+
+        var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+
+        Console.WriteLine("MQTT client subscribed to topic.");
+
+        // The response contains additional data sent by the server after subscribing.
+        response.DumpToConsole();
+    }
+}

Samples/MQTTnet.Samples.csproj

@@ -12,6 +12,7 @@
 
         <ProjectReference Include="..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj"/>
         <ProjectReference Include="..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj"/>
+        <ProjectReference Include="..\Source\MQTTnet.Extensions.Socks5\MQTTnet.Extensions.Socks5.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Server\MQTTnet.Server.csproj"/>
         <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj"/>

Samples/Server/Server_TLS_Samples.cs

@@ -59,10 +59,12 @@ public static class Server_TLS_Samples
         certRequest.CertificateExtensions.Add(sanBuilder.Build());
 
         using var certificate = certRequest.CreateSelfSigned(DateTimeOffset.Now.AddMinutes(-10), DateTimeOffset.Now.AddMinutes(10));
+#pragma warning disable SYSLIB0057
         var pfxCertificate = new X509Certificate2(
             certificate.Export(X509ContentType.Pfx),
             (string)null!,
             X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
+#pragma warning restore SYSLIB0057
 
         return pfxCertificate;
     }

Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.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 Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.DependencyInjection;
 using MQTTnet.Server;

Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.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 MQTTnet.Server;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/BufferExtensions.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.Runtime.InteropServices;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/EndpointRouterExtensions.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 Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Routing;
 

Source/MQTTnet.AspnetCore/MqttClientConnectionContextFactory.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 MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Formatter;

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -2,13 +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.Buffers;
 using System.IO.Pipelines;
 using System.Net;
 using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;

Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs

@@ -2,8 +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.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using MQTTnet.Adapter;

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -2,10 +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.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;

Source/MQTTnet.AspnetCore/MqttSubProtocolSelector.cs

@@ -2,9 +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.Collections.Generic;
-using System.Linq;
 using Microsoft.AspNetCore.Http;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.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;
 using System.Net;
 using System.Net.WebSockets;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;

Source/MQTTnet.AspnetCore/ReaderExtensions.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.Runtime.InteropServices;
 using MQTTnet.Adapter;

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.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 Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;

Source/MQTTnet.AspnetCore/SocketAwaitable.cs

@@ -2,13 +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.Diagnostics;
 using System.IO.Pipelines;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
-using System.Threading;
-using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore;
 

Source/MQTTnet.AspnetCore/SocketConnection.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;
-using System.Collections.Generic;
-using System.IO;
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Features;
 using MQTTnet.Exceptions;

Source/MQTTnet.AspnetCore/SocketReceiver.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.IO.Pipelines;
 using System.Net.Sockets;
 

Source/MQTTnet.AspnetCore/SocketSender.cs

@@ -2,9 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
 using System.Buffers;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Net.Sockets;

Source/MQTTnet.Benchmarks/AssemblyInfo.cs

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

Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs

@@ -2,9 +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.Threading;
-using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Internal;

Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -2,9 +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.IO;
-using System.Threading;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;

Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs

@@ -1,6 +1,5 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
-using System;
 
 
 namespace MQTTnet.Benchmarks;

Source/MQTTnet.Benchmarks/MqttBufferReaderBenchmark.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.Text;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;

Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.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 BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Formatter;

Source/MQTTnet.Benchmarks/Program.cs

@@ -2,9 +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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Running;
 
 namespace MQTTnet.Benchmarks;

Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs

@@ -4,9 +4,7 @@ using MQTTnet.AspNetCore;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using System.Buffers;
-using System.IO;
 using System.IO.Pipelines;
-using System.Threading.Tasks;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/RoundtripProcessingBenchmark.cs

@@ -1,7 +1,6 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
-using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -4,10 +4,7 @@
 
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Packets;
-using System;
 using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Channel;
 using MQTTnet.Formatter;

Source/MQTTnet.Benchmarks/ServerProcessingBenchmark.cs

@@ -5,7 +5,6 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
-using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs

@@ -2,8 +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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Server;
 

Source/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -5,8 +5,6 @@
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
-using System.Threading;
-using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.AspNetCore;

Source/MQTTnet.Benchmarks/TopicGenerator.cs

@@ -1,6 +1,3 @@
-using System;
-using System.Collections.Generic;
-
 namespace MQTTnet.Benchmarks;
 
 public class TopicGenerator

Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs

@@ -2,8 +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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Server;
 

Source/MQTTnet.Extensions.Rpc/DefaultMqttRpcClientTopicGenerationStrategy.cs

@@ -2,8 +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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

@@ -2,10 +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.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs

@@ -2,8 +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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public static class MqttFactoryExtensions

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -2,12 +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;
 using System.Buffers;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;

Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs

@@ -2,10 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
-using System.Collections.Generic;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Rpc/MqttRpcClientOptionsBuilder.cs

@@ -2,8 +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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public sealed class MqttRpcClientOptionsBuilder

Source/MQTTnet.Extensions.Rpc/TopicGenerationContext.cs

@@ -2,8 +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.Collections.Generic;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <AssemblyName>MQTTnet.Extensions.Socks5</AssemblyName>
+        <RootNamespace>MQTTnet.Extensions.Socks5</RootNamespace>
+        <IsAotCompatible>true</IsAotCompatible>
+
+        <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
+        <Description>This is an extension library which adds SOCKS5 proxy support (RFC 1928 / RFC 1929) to the MQTTnet TCP transport.</Description>
+        <PackageId>MQTTnet.Extensions.Socks5</PackageId>
+        <PackageTags>MQTT Message Queue Telemetry Transport MQTTClient SOCKS SOCKS5 Proxy Tunnel Networking IoT InternetOfThings</PackageTags>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
+        <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
+    </ItemGroup>
+
+</Project>

Source/MQTTnet.Extensions.Socks5/MqttClientOptionsBuilderExtensions.cs

@@ -0,0 +1,39 @@
+// 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.Extensions.Socks5;
+
+/// <summary>
+///     Sugar extensions for configuring a SOCKS5 proxy on the MQTT client options builder.
+/// </summary>
+public static class MqttClientOptionsBuilderExtensions
+{
+    /// <summary>
+    ///     Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+    ///     proxy. A TCP transport (via <c>WithTcpServer(...)</c> or <c>WithEndPoint(...)</c>)
+    ///     must be configured first.
+    /// </summary>
+    public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Socks5ProxyOptions options)
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(options);
+
+        return builder.WithStreamProvider(new Socks5StreamProvider(options));
+    }
+
+    /// <summary>
+    ///     Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+    ///     proxy using the fluent <see cref="Socks5ProxyOptionsBuilder"/>.
+    /// </summary>
+    public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Action<Socks5ProxyOptionsBuilder> configure)
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(configure);
+
+        var optionsBuilder = new Socks5ProxyOptionsBuilder();
+        configure(optionsBuilder);
+
+        return WithSocks5Proxy(builder, optionsBuilder.Build());
+    }
+}

Source/MQTTnet.Extensions.Socks5/Socks5NetworkStream.cs

@@ -0,0 +1,109 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net;
+using System.Net.Sockets;
+using MQTTnet.Channel;
+
+namespace MQTTnet.Extensions.Socks5;
+
+/// <summary>
+///     Wraps a <see cref="NetworkStream"/> opened by the SOCKS5 stream provider so
+///     <c>MqttTcpChannel</c> can read the original socket's endpoints via
+///     <see cref="IMqttClientStreamEndPoints"/> while the rest of the I/O delegates to the
+///     underlying stream.
+/// </summary>
+sealed class Socks5NetworkStream : Stream, IMqttClientStreamEndPoints
+{
+    readonly NetworkStream _inner;
+
+    public Socks5NetworkStream(NetworkStream inner, EndPoint localEndPoint, EndPoint remoteEndPoint)
+    {
+        _inner = inner ?? throw new ArgumentNullException(nameof(inner));
+        LocalEndPoint = localEndPoint;
+        RemoteEndPoint = remoteEndPoint;
+    }
+
+    public EndPoint LocalEndPoint { get; }
+
+    public EndPoint RemoteEndPoint { get; }
+
+    public override bool CanRead => _inner.CanRead;
+
+    public override bool CanSeek => _inner.CanSeek;
+
+    public override bool CanWrite => _inner.CanWrite;
+
+    public override bool CanTimeout => _inner.CanTimeout;
+
+    public override long Length => _inner.Length;
+
+    public override long Position
+    {
+        get => _inner.Position;
+        set => _inner.Position = value;
+    }
+
+    public override int ReadTimeout
+    {
+        get => _inner.ReadTimeout;
+        set => _inner.ReadTimeout = value;
+    }
+
+    public override int WriteTimeout
+    {
+        get => _inner.WriteTimeout;
+        set => _inner.WriteTimeout = value;
+    }
+
+    public override void Flush() => _inner.Flush();
+
+    public override Task FlushAsync(CancellationToken cancellationToken) => _inner.FlushAsync(cancellationToken);
+
+    public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count);
+
+    public override int Read(Span<byte> buffer) => _inner.Read(buffer);
+
+    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => _inner.ReadAsync(buffer, offset, count, cancellationToken);
+
+    public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        => _inner.ReadAsync(buffer, cancellationToken);
+
+    public override long Seek(long offset, SeekOrigin origin) => _inner.Seek(offset, origin);
+
+    public override void SetLength(long value) => _inner.SetLength(value);
+
+    public override void Write(byte[] buffer, int offset, int count) => _inner.Write(buffer, offset, count);
+
+    public override void Write(ReadOnlySpan<byte> buffer) => _inner.Write(buffer);
+
+    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => _inner.WriteAsync(buffer, offset, count, cancellationToken);
+
+    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+        => _inner.WriteAsync(buffer, cancellationToken);
+
+    public override void Close()
+    {
+        _inner.Close();
+        base.Close();
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _inner.Dispose();
+        }
+
+        base.Dispose(disposing);
+    }
+
+    public override async ValueTask DisposeAsync()
+    {
+        await _inner.DisposeAsync().ConfigureAwait(false);
+        await base.DisposeAsync().ConfigureAwait(false);
+    }
+}

Source/MQTTnet.Extensions.Socks5/Socks5Protocol.cs

@@ -0,0 +1,39 @@
+// 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.Extensions.Socks5;
+
+/// <summary>
+///     Protocol constants for SOCKS5 (RFC 1928) and the username/password sub-negotiation
+///     (RFC 1929). Internal use only.
+/// </summary>
+internal static class Socks5Protocol
+{
+    public const byte Version = 0x05;
+    public const byte AuthSubNegotiationVersion = 0x01;
+
+    public const byte MethodNoAuth = 0x00;
+    public const byte MethodGssApi = 0x01;
+    public const byte MethodUsernamePassword = 0x02;
+    public const byte MethodNoneAcceptable = 0xFF;
+
+    public const byte CommandConnect = 0x01;
+    public const byte Reserved = 0x00;
+
+    public const byte AddressTypeIPv4 = 0x01;
+    public const byte AddressTypeDomainName = 0x03;
+    public const byte AddressTypeIPv6 = 0x04;
+
+    public const byte ReplySucceeded = 0x00;
+    public const byte ReplyGeneralFailure = 0x01;
+    public const byte ReplyConnectionNotAllowed = 0x02;
+    public const byte ReplyNetworkUnreachable = 0x03;
+    public const byte ReplyHostUnreachable = 0x04;
+    public const byte ReplyConnectionRefused = 0x05;
+    public const byte ReplyTtlExpired = 0x06;
+    public const byte ReplyCommandNotSupported = 0x07;
+    public const byte ReplyAddressTypeNotSupported = 0x08;
+
+    public const int MaxStringLength = 255;
+}

Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptions.cs

@@ -0,0 +1,65 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net.Sockets;
+
+namespace MQTTnet.Extensions.Socks5;
+
+/// <summary>
+///     Options for the SOCKS5 stream provider (RFC 1928 / RFC 1929).
+///     A new provider can be constructed from these options via
+///     <see cref="MqttClientOptionsBuilderExtensions.WithSocks5Proxy(MqttClientOptionsBuilder, Socks5ProxyOptions)"/>.
+/// </summary>
+public sealed class Socks5ProxyOptions
+{
+    /// <summary>
+    ///     The SOCKS5 proxy host name or IP literal. Must be set.
+    /// </summary>
+    public string Host { get; set; }
+
+    /// <summary>
+    ///     The SOCKS5 proxy port. Defaults to 1080.
+    /// </summary>
+    public int Port { get; set; } = 1080;
+
+    /// <summary>
+    ///     The user name used for the username/password authentication sub-negotiation (RFC 1929).
+    ///     When <c>null</c> or empty, the client will only offer the "no authentication" method.
+    /// </summary>
+    public string Username { get; set; }
+
+    /// <summary>
+    ///     The password (raw bytes per RFC 1929) used for the username/password authentication
+    ///     sub-negotiation. Must be 1..255 bytes when <see cref="Username"/> is set.
+    /// </summary>
+    public byte[] Password { get; set; }
+
+    /// <summary>
+    ///     When <c>true</c> (the default) the broker hostname is forwarded to the proxy as a
+    ///     SOCKS5 <c>DOMAINNAME</c> address type, so DNS resolution happens at the proxy
+    ///     (typical when the broker is behind the proxy in a private network).
+    ///     When <c>false</c>, the hostname is resolved locally before sending the CONNECT request
+    ///     and an IPv4 / IPv6 address type is used.
+    /// </summary>
+    public bool ResolveDnsRemotely { get; set; } = true;
+
+    /// <summary>
+    ///     The maximum time allowed for the full SOCKS5 handshake (TCP connect to proxy + greeting
+    ///     + optional auth + CONNECT). The handshake will be cancelled and a
+    ///     <see cref="MQTTnet.Exceptions.MqttProxyException"/> will be thrown when this elapses.
+    ///     Defaults to 30 seconds.
+    /// </summary>
+    public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    ///     Address family preference for the socket connecting to the proxy and for local DNS
+    ///     resolution when <see cref="ResolveDnsRemotely"/> is <c>false</c>.
+    /// </summary>
+    public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
+
+    public override string ToString()
+    {
+        return $"{Host}:{Port}";
+    }
+}

Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptionsBuilder.cs

@@ -0,0 +1,94 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net.Sockets;
+using System.Text;
+
+namespace MQTTnet.Extensions.Socks5;
+
+/// <summary>
+///     Fluent builder for <see cref="Socks5ProxyOptions"/>.
+/// </summary>
+public sealed class Socks5ProxyOptionsBuilder
+{
+    readonly Socks5ProxyOptions _options = new();
+
+    public Socks5ProxyOptions Build()
+    {
+        if (string.IsNullOrEmpty(_options.Host))
+        {
+            throw new InvalidOperationException("A SOCKS5 proxy host must be configured via WithHost.");
+        }
+
+        if (_options.Port <= 0 || _options.Port > 65535)
+        {
+            throw new InvalidOperationException("The SOCKS5 proxy port must be in the range 1..65535.");
+        }
+
+        if (_options.HandshakeTimeout <= TimeSpan.Zero)
+        {
+            throw new InvalidOperationException("The SOCKS5 handshake timeout must be greater than zero.");
+        }
+
+        if (string.IsNullOrEmpty(_options.Username) && _options.Password is { Length: > 0 })
+        {
+            throw new InvalidOperationException("A SOCKS5 password cannot be configured without a username.");
+        }
+
+        if (!string.IsNullOrEmpty(_options.Username) && (_options.Password == null || _options.Password.Length == 0))
+        {
+            throw new InvalidOperationException("A SOCKS5 password must be configured when a username is set.");
+        }
+
+        return _options;
+    }
+
+    public Socks5ProxyOptionsBuilder WithHost(string host)
+    {
+        _options.Host = host ?? throw new ArgumentNullException(nameof(host));
+        return this;
+    }
+
+    public Socks5ProxyOptionsBuilder WithPort(int port)
+    {
+        _options.Port = port;
+        return this;
+    }
+
+    public Socks5ProxyOptionsBuilder WithCredentials(string username, string password)
+    {
+        _options.Username = username;
+        _options.Password = password != null ? Encoding.UTF8.GetBytes(password) : null;
+        return this;
+    }
+
+    public Socks5ProxyOptionsBuilder WithCredentials(string username, byte[] password)
+    {
+        _options.Username = username;
+        _options.Password = password;
+        return this;
+    }
+
+    /// <summary>
+    ///     Controls whether the broker hostname is forwarded to the proxy (remote DNS, default) or
+    ///     resolved locally before issuing the CONNECT request.
+    /// </summary>
+    public Socks5ProxyOptionsBuilder WithResolveDnsRemotely(bool resolveDnsRemotely = true)
+    {
+        _options.ResolveDnsRemotely = resolveDnsRemotely;
+        return this;
+    }
+
+    public Socks5ProxyOptionsBuilder WithHandshakeTimeout(TimeSpan timeout)
+    {
+        _options.HandshakeTimeout = timeout;
+        return this;
+    }
+
+    public Socks5ProxyOptionsBuilder WithAddressFamily(AddressFamily addressFamily)
+    {
+        _options.AddressFamily = addressFamily;
+        return this;
+    }
+}

Source/MQTTnet.Extensions.Socks5/Socks5StreamProvider.cs

@@ -0,0 +1,507 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using MQTTnet.Channel;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Extensions.Socks5;
+
+/// <summary>
+///     <see cref="IMqttClientStreamProvider"/> implementation that tunnels the connection through
+///     a SOCKS5 proxy (RFC 1928) with optional username/password authentication (RFC 1929).
+/// </summary>
+public sealed class Socks5StreamProvider : IMqttClientStreamProvider
+{
+    readonly Socks5ProxyOptions _options;
+
+    public Socks5StreamProvider(Socks5ProxyOptions options)
+    {
+        _options = options ?? throw new ArgumentNullException(nameof(options));
+
+        if (string.IsNullOrEmpty(_options.Host))
+        {
+            throw new ArgumentException("Socks5ProxyOptions.Host must be set.", nameof(options));
+        }
+
+        if (_options.Port <= 0 || _options.Port > 65535)
+        {
+            throw new ArgumentOutOfRangeException(nameof(options), _options.Port, "Socks5ProxyOptions.Port must be in the range 1..65535.");
+        }
+
+        if (_options.HandshakeTimeout <= TimeSpan.Zero)
+        {
+            throw new ArgumentOutOfRangeException(nameof(options), _options.HandshakeTimeout, "Socks5ProxyOptions.HandshakeTimeout must be greater than zero.");
+        }
+
+        if (string.IsNullOrEmpty(_options.Username) && _options.Password is { Length: > 0 })
+        {
+            throw new ArgumentException("Socks5ProxyOptions.Password cannot be set without Socks5ProxyOptions.Username.", nameof(options));
+        }
+
+        if (!string.IsNullOrEmpty(_options.Username) && (_options.Password == null || _options.Password.Length == 0))
+        {
+            throw new ArgumentException("Socks5ProxyOptions.Password must be set when Socks5ProxyOptions.Username is set.", nameof(options));
+        }
+    }
+
+    public async Task<Stream> ConnectAsync(EndPoint brokerEndPoint, CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(brokerEndPoint);
+
+        using var timeoutCts = new CancellationTokenSource(_options.HandshakeTimeout);
+        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token);
+        var handshakeToken = linkedCts.Token;
+
+        var addressFamily = _options.AddressFamily;
+        Socket socket = addressFamily == AddressFamily.Unspecified
+            ? new Socket(SocketType.Stream, ProtocolType.Tcp)
+            : new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
+
+        NetworkStream networkStream = null;
+        try
+        {
+            try
+            {
+                await socket.ConnectAsync(_options.Host, _options.Port, handshakeToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyUnreachable,
+                    $"Timed out connecting to SOCKS5 proxy after {_options.HandshakeTimeout}.",
+                    _options.ToString());
+            }
+            catch (Exception ex) when (ex is SocketException or IOException)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyUnreachable,
+                    $"Could not connect to SOCKS5 proxy: {ex.Message}",
+                    _options.ToString(),
+                    ex);
+            }
+
+            networkStream = new NetworkStream(socket, ownsSocket: true);
+
+            try
+            {
+                await PerformGreetingAndAuthAsync(networkStream, handshakeToken).ConfigureAwait(false);
+                await PerformConnectAsync(networkStream, brokerEndPoint, handshakeToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyProtocolError,
+                    $"Timed out during SOCKS5 handshake after {_options.HandshakeTimeout}.",
+                    _options.ToString());
+            }
+            catch (EndOfStreamException ex)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyProtocolError,
+                    "The SOCKS5 proxy closed the connection during the handshake.",
+                    _options.ToString(),
+                    ex);
+            }
+            catch (Exception ex) when (ex is IOException or SocketException)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyProtocolError,
+                    $"I/O error during SOCKS5 handshake: {ex.Message}",
+                    _options.ToString(),
+                    ex);
+            }
+
+            // Hand off the stream — ownership transfers to the caller.
+            var result = new Socks5NetworkStream(networkStream, socket.LocalEndPoint, socket.RemoteEndPoint);
+            networkStream = null;
+            socket = null;
+            return result;
+        }
+        finally
+        {
+            networkStream?.Dispose();
+            socket?.Dispose();
+        }
+    }
+
+    async Task PerformGreetingAndAuthAsync(NetworkStream stream, CancellationToken cancellationToken)
+    {
+        var hasCredentials = !string.IsNullOrEmpty(_options.Username);
+
+        // VER NMETHODS METHOD[1..n]
+        // We always offer no-auth; we additionally offer user/pass when credentials are configured.
+        byte[] greeting;
+        if (hasCredentials)
+        {
+            greeting = [Socks5Protocol.Version, 0x02, Socks5Protocol.MethodNoAuth, Socks5Protocol.MethodUsernamePassword];
+        }
+        else
+        {
+            greeting = [Socks5Protocol.Version, 0x01, Socks5Protocol.MethodNoAuth];
+        }
+
+        await stream.WriteAsync(greeting, cancellationToken).ConfigureAwait(false);
+
+        // VER METHOD
+        var greetingResponse = new byte[2];
+        await ReadExactlyAsync(stream, greetingResponse, cancellationToken).ConfigureAwait(false);
+
+        if (greetingResponse[0] != Socks5Protocol.Version)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyProtocolError,
+                $"Unexpected SOCKS version in greeting response: 0x{greetingResponse[0]:X2}.",
+                _options.ToString());
+        }
+
+        var chosenMethod = greetingResponse[1];
+        switch (chosenMethod)
+        {
+            case Socks5Protocol.MethodNoAuth:
+                if (hasCredentials)
+                {
+                    // Server accepted no-auth even though we offered credentials: fine, continue.
+                }
+
+                return;
+
+            case Socks5Protocol.MethodUsernamePassword:
+                if (!hasCredentials)
+                {
+                    throw new MqttProxyException(
+                        MqttProxyErrorCode.ProxyProtocolError,
+                        "The SOCKS5 server selected username/password authentication but no credentials are configured.",
+                        _options.ToString());
+                }
+
+                await PerformUsernamePasswordAuthAsync(stream, cancellationToken).ConfigureAwait(false);
+                return;
+
+            case Socks5Protocol.MethodNoneAcceptable:
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyAuthMethodRejected,
+                    "The SOCKS5 server rejected all offered authentication methods.",
+                    _options.ToString());
+
+            default:
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyProtocolError,
+                    $"The SOCKS5 server selected an unsupported authentication method: 0x{chosenMethod:X2}.",
+                    _options.ToString());
+        }
+    }
+
+    async Task PerformUsernamePasswordAuthAsync(NetworkStream stream, CancellationToken cancellationToken)
+    {
+        var usernameBytes = Encoding.UTF8.GetBytes(_options.Username ?? string.Empty);
+        var passwordBytes = _options.Password ?? [];
+
+        if (usernameBytes.Length > Socks5Protocol.MaxStringLength)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAuthFailed,
+                $"Username exceeds {Socks5Protocol.MaxStringLength} bytes (RFC 1929).",
+                _options.ToString());
+        }
+
+        if (usernameBytes.Length == 0)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAuthFailed,
+                "Username must be 1..255 bytes (RFC 1929).",
+                _options.ToString());
+        }
+
+        if (passwordBytes.Length > Socks5Protocol.MaxStringLength)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAuthFailed,
+                $"Password exceeds {Socks5Protocol.MaxStringLength} bytes (RFC 1929).",
+                _options.ToString());
+        }
+
+        if (passwordBytes.Length == 0)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAuthFailed,
+                "Password must be 1..255 bytes (RFC 1929).",
+                _options.ToString());
+        }
+
+        var request = new byte[3 + usernameBytes.Length + passwordBytes.Length];
+        var offset = 0;
+        request[offset++] = Socks5Protocol.AuthSubNegotiationVersion;
+        request[offset++] = (byte)usernameBytes.Length;
+        Buffer.BlockCopy(usernameBytes, 0, request, offset, usernameBytes.Length);
+        offset += usernameBytes.Length;
+        request[offset++] = (byte)passwordBytes.Length;
+        Buffer.BlockCopy(passwordBytes, 0, request, offset, passwordBytes.Length);
+
+        await stream.WriteAsync(request, cancellationToken).ConfigureAwait(false);
+
+        var response = new byte[2];
+        await ReadExactlyAsync(stream, response, cancellationToken).ConfigureAwait(false);
+
+        if (response[0] != Socks5Protocol.AuthSubNegotiationVersion)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyProtocolError,
+                $"Unexpected SOCKS5 auth sub-negotiation version: 0x{response[0]:X2}.",
+                _options.ToString());
+        }
+
+        if (response[1] != 0x00)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAuthFailed,
+                $"The SOCKS5 server rejected the supplied credentials (status 0x{response[1]:X2}).",
+                _options.ToString());
+        }
+    }
+
+    async Task PerformConnectAsync(NetworkStream stream, EndPoint brokerEndPoint, CancellationToken cancellationToken)
+    {
+        var request = BuildConnectRequest(brokerEndPoint, out var targetDescription);
+
+        await stream.WriteAsync(request, cancellationToken).ConfigureAwait(false);
+
+        // Reply: VER REP RSV ATYP BND.ADDR BND.PORT
+        var header = new byte[4];
+        await ReadExactlyAsync(stream, header, cancellationToken).ConfigureAwait(false);
+
+        if (header[0] != Socks5Protocol.Version)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyProtocolError,
+                $"Unexpected SOCKS version in CONNECT response: 0x{header[0]:X2}.",
+                _options.ToString());
+        }
+
+        if (header[2] != Socks5Protocol.Reserved)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyProtocolError,
+                $"Unexpected SOCKS5 reserved byte in CONNECT response: 0x{header[2]:X2}.",
+                _options.ToString());
+        }
+
+        var rep = header[1];
+        if (rep != Socks5Protocol.ReplySucceeded)
+        {
+            var (code, description) = MapReplyCode(rep);
+            throw new MqttProxyException(
+                code,
+                $"SOCKS5 CONNECT to '{targetDescription}' failed: {description}.",
+                _options.ToString());
+        }
+
+        // Discard BND.ADDR / BND.PORT. The length depends on ATYP and we tolerate unexpected lengths.
+        var atyp = header[3];
+        switch (atyp)
+        {
+            case Socks5Protocol.AddressTypeIPv4:
+                await SkipAsync(stream, 4 + 2, cancellationToken).ConfigureAwait(false);
+                break;
+
+            case Socks5Protocol.AddressTypeIPv6:
+                await SkipAsync(stream, 16 + 2, cancellationToken).ConfigureAwait(false);
+                break;
+
+            case Socks5Protocol.AddressTypeDomainName:
+                var lenBuf = new byte[1];
+                await ReadExactlyAsync(stream, lenBuf, cancellationToken).ConfigureAwait(false);
+                await SkipAsync(stream, lenBuf[0] + 2, cancellationToken).ConfigureAwait(false);
+                break;
+
+            default:
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyProtocolError,
+                    $"Unexpected SOCKS5 address type in CONNECT response: 0x{atyp:X2}.",
+                    _options.ToString());
+        }
+    }
+
+    byte[] BuildConnectRequest(EndPoint brokerEndPoint, out string targetDescription)
+    {
+        // Layout: VER CMD RSV ATYP DST.ADDR DST.PORT
+        if (brokerEndPoint is DnsEndPoint dns)
+        {
+            if (_options.ResolveDnsRemotely)
+            {
+                return BuildDomainConnectRequest(dns.Host, dns.Port, out targetDescription);
+            }
+
+            var addresses = ResolveDns(dns.Host, dns.Port);
+            return BuildIpConnectRequest(addresses, dns.Port, out targetDescription);
+        }
+
+        if (brokerEndPoint is IPEndPoint ip)
+        {
+            return BuildIpConnectRequest([ip.Address], ip.Port, out targetDescription);
+        }
+
+        throw new MqttProxyException(
+            MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+            $"Broker endpoint of type '{brokerEndPoint.GetType().Name}' is not supported by the SOCKS5 stream provider.",
+            _options.ToString());
+    }
+
+    IPAddress[] ResolveDns(string host, int port)
+    {
+        try
+        {
+            var addresses = _options.AddressFamily == AddressFamily.Unspecified
+                ? Dns.GetHostAddresses(host)
+                : Dns.GetHostAddresses(host, _options.AddressFamily);
+
+            if (addresses == null || addresses.Length == 0)
+            {
+                throw new MqttProxyException(
+                    MqttProxyErrorCode.ProxyHostUnreachable,
+                    $"DNS resolution for '{host}:{port}' returned no addresses.",
+                    _options.ToString());
+            }
+
+            return addresses;
+        }
+        catch (SocketException ex)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyHostUnreachable,
+                $"Local DNS resolution failed for '{host}:{port}': {ex.Message}",
+                _options.ToString(),
+                ex);
+        }
+    }
+
+    byte[] BuildIpConnectRequest(IPAddress[] addresses, int port, out string targetDescription)
+    {
+        var preferredFamily = _options.AddressFamily == AddressFamily.Unspecified
+            ? AddressFamily.InterNetwork
+            : _options.AddressFamily;
+
+        IPAddress chosen = null;
+        foreach (var address in addresses)
+        {
+            if (address.AddressFamily == preferredFamily)
+            {
+                chosen = address;
+                break;
+            }
+        }
+
+        chosen ??= addresses[0];
+
+        targetDescription = $"{chosen}:{port}";
+
+        if (chosen.AddressFamily == AddressFamily.InterNetwork)
+        {
+            var addr = chosen.GetAddressBytes();
+            var request = new byte[4 + 4 + 2];
+            request[0] = Socks5Protocol.Version;
+            request[1] = Socks5Protocol.CommandConnect;
+            request[2] = Socks5Protocol.Reserved;
+            request[3] = Socks5Protocol.AddressTypeIPv4;
+            Buffer.BlockCopy(addr, 0, request, 4, 4);
+            BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(8, 2), (ushort)port);
+            return request;
+        }
+
+        if (chosen.AddressFamily == AddressFamily.InterNetworkV6)
+        {
+            var addr = chosen.GetAddressBytes();
+            var request = new byte[4 + 16 + 2];
+            request[0] = Socks5Protocol.Version;
+            request[1] = Socks5Protocol.CommandConnect;
+            request[2] = Socks5Protocol.Reserved;
+            request[3] = Socks5Protocol.AddressTypeIPv6;
+            Buffer.BlockCopy(addr, 0, request, 4, 16);
+            BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(20, 2), (ushort)port);
+            return request;
+        }
+
+        throw new MqttProxyException(
+            MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+            $"IP address family '{chosen.AddressFamily}' is not supported by SOCKS5.",
+            _options.ToString());
+    }
+
+    byte[] BuildDomainConnectRequest(string host, int port, out string targetDescription)
+    {
+        var hostBytes = Encoding.UTF8.GetBytes(host ?? string.Empty);
+        if (hostBytes.Length == 0)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyHostUnreachable,
+                "Broker host is empty.",
+                _options.ToString());
+        }
+
+        if (hostBytes.Length > Socks5Protocol.MaxStringLength)
+        {
+            throw new MqttProxyException(
+                MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+                $"Broker host exceeds the SOCKS5 DOMAINNAME limit of {Socks5Protocol.MaxStringLength} bytes.",
+                _options.ToString());
+        }
+
+        targetDescription = $"{host}:{port}";
+
+        var request = new byte[4 + 1 + hostBytes.Length + 2];
+        request[0] = Socks5Protocol.Version;
+        request[1] = Socks5Protocol.CommandConnect;
+        request[2] = Socks5Protocol.Reserved;
+        request[3] = Socks5Protocol.AddressTypeDomainName;
+        request[4] = (byte)hostBytes.Length;
+        Buffer.BlockCopy(hostBytes, 0, request, 5, hostBytes.Length);
+        BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(5 + hostBytes.Length, 2), (ushort)port);
+        return request;
+    }
+
+    static (MqttProxyErrorCode Code, string Description) MapReplyCode(byte rep)
+    {
+        return rep switch
+        {
+            Socks5Protocol.ReplyGeneralFailure => (MqttProxyErrorCode.ProxyGeneralFailure, "general SOCKS server failure"),
+            Socks5Protocol.ReplyConnectionNotAllowed => (MqttProxyErrorCode.ProxyConnectionNotAllowed, "connection not allowed by ruleset"),
+            Socks5Protocol.ReplyNetworkUnreachable => (MqttProxyErrorCode.ProxyNetworkUnreachable, "network unreachable"),
+            Socks5Protocol.ReplyHostUnreachable => (MqttProxyErrorCode.ProxyHostUnreachable, "host unreachable"),
+            Socks5Protocol.ReplyConnectionRefused => (MqttProxyErrorCode.ProxyConnectionRefused, "connection refused"),
+            Socks5Protocol.ReplyTtlExpired => (MqttProxyErrorCode.ProxyTtlExpired, "TTL expired"),
+            Socks5Protocol.ReplyCommandNotSupported => (MqttProxyErrorCode.ProxyCommandNotSupported, "command not supported"),
+            Socks5Protocol.ReplyAddressTypeNotSupported => (MqttProxyErrorCode.ProxyAddressTypeNotSupported, "address type not supported"),
+            _ => (MqttProxyErrorCode.ProxyUnknownError, $"unknown reply code 0x{rep:X2}")
+        };
+    }
+
+    static async Task ReadExactlyAsync(NetworkStream stream, byte[] buffer, CancellationToken cancellationToken)
+    {
+        var offset = 0;
+        while (offset < buffer.Length)
+        {
+            var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset), cancellationToken).ConfigureAwait(false);
+            if (read <= 0)
+            {
+                throw new EndOfStreamException();
+            }
+
+            offset += read;
+        }
+    }
+
+    static async Task SkipAsync(NetworkStream stream, int count, CancellationToken cancellationToken)
+    {
+        if (count <= 0)
+        {
+            return;
+        }
+
+        var buffer = new byte[count];
+        await ReadExactlyAsync(stream, buffer, cancellationToken).ConfigureAwait(false);
+    }
+}

Source/MQTTnet.Extensions.TopicTemplate/MqttTopicTemplate.cs

@@ -2,11 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Text.RegularExpressions;
-using System.Threading;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.TopicTemplate;

Source/MQTTnet.Extensions.TopicTemplate/TopicTemplateExtensions.cs

@@ -2,8 +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.Linq;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 

Source/MQTTnet.PowerShell/Cmdlets/ConnectMqttSessionCmdlet.cs

@@ -1,120 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Connect, "MqttSession")]
-[OutputType(typeof(MqttClientConnectResult))]
-public class ConnectMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter]
-    public SwitchParameter CleanSession { get; set; } = true;
-
-    [Parameter]
-    public string? ClientId { get; set; } = Guid.NewGuid().ToString();
-
-    [Parameter(Mandatory = true, ParameterSetName = "Tcp")]
-    [Parameter(Mandatory = true, ParameterSetName = "TcpWithTls")]
-    public new string? Host { get; set; }
-
-    [Parameter]
-    public string? Password { get; set; }
-
-    [Parameter(ParameterSetName = "Tcp")]
-    [Parameter(ParameterSetName = "TcpWithTls")]
-    public int Port { get; set; } = 1883;
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public string? Username { get; set; }
-
-    [Parameter(ParameterSetName = "TcpWithTls")]
-    public SwitchParameter UseTls { get; set; }
-
-    [Parameter(ParameterSetName = "TcpWithTls")]
-    public SwitchParameter AllowUntrustedCertificates { get; set; }
-
-    [Parameter(Mandatory = true, ParameterSetName = "WebSocket")]
-    public string? WebSocketUri { get; set; }
-
-    [Parameter]
-    public int KeepAlivePeriod { get; set; } = 15;
-
-    [Parameter]
-    public int Timeout { get; set; } = 10;
-
-    [Parameter]
-    public string? WillTopic { get; set; }
-
-    [Parameter]
-    public string? WillPayload { get; set; }
-
-    [Parameter]
-    public int WillQoS { get; set; } = 0;
-
-    [Parameter]
-    public SwitchParameter WillRetain { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var clientOptionsBuilder = new MqttClientOptionsBuilder();
-        
-        // Configure transport
-        if (ParameterSetName == "WebSocket")
-        {
-            clientOptionsBuilder.WithWebSocketServer(o => o.WithUri(WebSocketUri!));
-        }
-        else
-        {
-            clientOptionsBuilder.WithTcpServer(Host!, Port);
-            
-            if (UseTls)
-            {
-                clientOptionsBuilder.WithTlsOptions(o =>
-                {
-                    if (AllowUntrustedCertificates)
-                    {
-                        o.WithCertificateValidationHandler(_ => true);
-                    }
-                });
-            }
-        }
-
-        // Configure authentication
-        if (Username != null)
-        {
-            clientOptionsBuilder.WithCredentials(Username, Password);
-        }
-
-        // Configure session
-        clientOptionsBuilder.WithCleanSession(CleanSession);
-        
-        if (!string.IsNullOrEmpty(ClientId))
-        {
-            clientOptionsBuilder.WithClientId(ClientId);
-        }
-
-        // Configure keep alive
-        clientOptionsBuilder.WithKeepAlivePeriod(TimeSpan.FromSeconds(KeepAlivePeriod));
-        clientOptionsBuilder.WithTimeout(TimeSpan.FromSeconds(Timeout));
-
-        // Configure Will message
-        if (!string.IsNullOrEmpty(WillTopic))
-        {
-            clientOptionsBuilder.WithWillTopic(WillTopic);
-            
-            if (!string.IsNullOrEmpty(WillPayload))
-            {
-                clientOptionsBuilder.WithWillPayload(System.Text.Encoding.UTF8.GetBytes(WillPayload));
-            }
-            
-            clientOptionsBuilder.WithWillQualityOfServiceLevel((Protocol.MqttQualityOfServiceLevel)WillQoS);
-            clientOptionsBuilder.WithWillRetain(WillRetain);
-        }
-
-        var response = Session.GetClient().ConnectAsync(clientOptionsBuilder.Build()).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/DisconnectMqttSessionCmdlet.cs

@@ -1,31 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Disconnect, "MqttSession")]
-public class DisconnectMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter]
-    public MqttClientDisconnectOptionsReason Reason { get; set; } = MqttClientDisconnectOptionsReason.NormalDisconnection;
-
-    [Parameter]
-    public string? ReasonString { get; set; }
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public uint SessionExpiryInterval { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        if (Session.GetClient().IsConnected)
-        {
-            var options = new MqttClientDisconnectOptionsBuilder().WithSessionExpiryInterval(SessionExpiryInterval).WithReason(Reason).WithReasonString(ReasonString).Build();
-
-            Session.GetClient().DisconnectAsync(options).GetAwaiter().GetResult();
-        }
-
-        WriteObject("Disconnected.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/GetMqttSessionStatusCmdlet.cs

@@ -1,27 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.Get, "MqttSessionStatus")]
-[OutputType(typeof(MqttClientConnectionStatus))]
-public class GetMqttSessionStatusCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var client = Session.GetClient();
-        var status = new
-        {
-            IsConnected = client.IsConnected,
-            Options = client.Options
-        };
-
-        WriteObject(status);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/NewMqttSessionCmdlet.cs

@@ -1,13 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.New, "MqttSession")]
-[OutputType(typeof(PsMqttSession))]
-public class NewMqttSessionCmdlet : PSCmdlet
-{
-    protected override void ProcessRecord()
-    {
-        WriteObject(new PsMqttSession());
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/PublishMqttMessageCmdlet.cs

@@ -1,69 +0,0 @@
-using System.Collections;
-using System.Management.Automation;
-using System.Text;
-using MQTTnet.Protocol;
-using MQTTnet.Packets;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsData.Publish, "MqttMessage")]
-public class PublishMqttMessageCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true)]
-    public string? Payload { get; set; }
-
-    [Parameter]
-    public int QoS { get; set; } = 0;
-
-    [Parameter]
-    public SwitchParameter Retain { get; set; }
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    [Parameter]
-    public string? ContentType { get; set; }
-
-    [Parameter]
-    public string? ResponseTopic { get; set; }
-
-    [Parameter]
-    public ushort TopicAlias { get; set; }
-
-    [Parameter]
-    public uint MessageExpiryInterval { get; set; }
-
-    [Parameter]
-    public Hashtable? UserProperties { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var msgBuilder = new MqttApplicationMessageBuilder()
-            .WithTopic(Topic)
-            .WithPayload(Encoding.UTF8.GetBytes(Payload ?? string.Empty))
-            .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
-            .WithRetainFlag(Retain)
-            .WithContentType(ContentType)
-            .WithResponseTopic(ResponseTopic)
-            .WithTopicAlias(TopicAlias)
-            .WithMessageExpiryInterval(MessageExpiryInterval);
-
-        // Add user properties if provided
-        if (UserProperties != null && UserProperties.Count > 0)
-        {
-            foreach (var key in UserProperties.Keys)
-            {
-                var value = UserProperties[key]?.ToString() ?? string.Empty;
-                msgBuilder.WithUserProperty(key.ToString(), value);
-            }
-        }
-
-        var msg = msgBuilder.Build();
-        var response = Session.GetClient().PublishAsync(msg).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/ReceiveMqttMessageCmdlet.cs

@@ -1,74 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Receive, "MqttMessage")]
-[OutputType(typeof(PsMqttMessage))]
-public class ReceiveMqttMessageCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    [ValidateRange(0, int.MaxValue)]
-    public int TimeoutSeconds { get; set; } = 0;
-
-    protected override void ProcessRecord()
-    {
-        var tcs = new TaskCompletionSource<PsMqttMessage>();
-        EventHandler<PsMqttMessage>? handler = null;
-        CancellationTokenSource? cts = null;
-
-        try
-        {
-            handler = (s, e) =>
-            {
-                if (!tcs.Task.IsCompleted)
-                {
-                    tcs.TrySetResult(e);
-                }
-            };
-
-            Session.MessageReceived += handler;
-
-            // Optionaler Timeout
-            if (TimeoutSeconds > 0)
-            {
-                cts = new CancellationTokenSource();
-                cts.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
-                cts.Token.Register(() =>
-                {
-                    if (!tcs.Task.IsCompleted)
-                    {
-                        tcs.TrySetCanceled();
-                    }
-                });
-            }
-
-            WriteVerbose("Waiting for next MQTT message...");
-
-            try
-            {
-                var message = tcs.Task.GetAwaiter().GetResult();
-                WriteObject(message);
-            }
-            catch (TaskCanceledException)
-            {
-                WriteWarning($"Timeout after {TimeoutSeconds} seconds — no message received.");
-            }
-        }
-        finally
-        {
-            if (handler != null)
-            {
-                Session.MessageReceived -= handler;
-            }
-
-            cts?.Dispose();
-        }
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/RegisterMqttMessageHandlerCmdlet.cs

@@ -1,59 +0,0 @@
-using System.Management.Automation;
-using System.Management.Automation.Runspaces;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsLifecycle.Register, "MqttMessageHandler")]
-public class RegisterMqttMessageHandlerCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public ScriptBlock? Action { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        // Capture the ScriptBlock and current default runspace (available during cmdlet execution)
-        var scriptBlock = Action ?? throw new ArgumentNullException(nameof(Action));
-        var capturedRunspace = Runspace.DefaultRunspace ?? throw new InvalidOperationException("No default runspace available during cmdlet execution.");
-        var capturedHost = Host;
-
-        EventHandler<PsMqttMessage> handler = (sender, e) =>
-        {
-            // Run asynchronously to avoid blocking the event thread
-            Task.Run(() =>
-            {
-                // Temporarily set the default runspace for this thread
-                var originalRunspace = Runspace.DefaultRunspace;
-                Runspace.DefaultRunspace = capturedRunspace;
-
-                try
-                {
-                    // Invoke the ScriptBlock directly with arguments (returns Collection<PSObject>)
-                    var results = scriptBlock.Invoke(e.Topic, e.Payload, e); // Pass topic, payload, and full message
-
-                    // Handle output manually (e.g., write to host/console)
-                    foreach (var result in results)
-                    {
-                        capturedHost.UI.WriteLine(result?.ToString() ?? string.Empty);
-                    }
-                }
-                catch (Exception ex)
-                {
-                    // Handle errors by writing to host error stream
-                    capturedHost.UI.WriteErrorLine($"MQTT handler error: {ex.Message}");
-                    // Optionally: capturedHost.UI.WriteVerboseLine(ex.StackTrace); for more details
-                }
-                finally
-                {
-                    // Restore the original default runspace for this thread
-                    Runspace.DefaultRunspace = originalRunspace;
-                }
-            });
-        };
-
-        Session.MessageReceived += handler;
-        WriteObject($"Handler registered for MQTT messages on session.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/RemoveMqttSessionCmdlet.cs

@@ -1,16 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.Remove, "MqttSession")]
-public class RemoveMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        Session.Dispose();
-        WriteObject("Session removed.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/ShowMqttExamplesCmdlet.cs

@@ -1,78 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.Show, "MqttExamples")]
-public class ShowMqttExamplesCmdlet : PSCmdlet
-{
-    protected override void ProcessRecord()
-    {
-        var examples = @"
-MQTTnet PowerShell Module - Quick Examples
-==========================================
-
-1. Basic Connection and Publish:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
-   Publish-MqttMessage -Session $session -Topic 'test/topic' -Payload 'Hello'
-   Disconnect-MqttSession -Session $session
-   Remove-MqttSession -Session $session
-
-2. Subscribe and Receive Messages:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
-   Subscribe-MqttTopic -Session $session -Topic 'test/demo'
-   $msg = Receive-MqttMessage -Session $session -TimeoutSeconds 30
-   Write-Host ""Received: $($msg.Payload)""
-
-3. WebSocket Connection:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -WebSocketUri 'ws://broker.hivemq.com:8000/mqtt'
-   Publish-MqttMessage -Session $session -Topic 'test/ws' -Payload 'Hello WS'
-
-4. TLS/SSL Connection:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -Host 'broker.hivemq.com' -Port 8883 `
-       -UseTls -AllowUntrustedCertificates
-
-5. Async Message Handler:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -Host 'broker.hivemq.com'
-   Subscribe-MqttTopic -Session $session -Topic 'test/#'
-   Register-MqttMessageHandler -Session $session -Action {
-       param($topic, $payload, $message)
-       Write-Host ""Message on $topic : $payload""
-   }
-
-6. Last Will and Testament:
-   $session = New-MqttSession
-   Connect-MqttSession -Session $session -Host 'broker.hivemq.com' `
-       -WillTopic 'status/client' -WillPayload 'Offline' -WillQoS 1 -WillRetain
-
-7. Publish with User Properties:
-   $props = @{ 'source' = 'PowerShell'; 'version' = '1.0' }
-   Publish-MqttMessage -Session $session -Topic 'test/props' `
-       -Payload 'Data' -UserProperties $props
-
-8. High QoS Publishing:
-   Publish-MqttMessage -Session $session -Topic 'critical/data' `
-       -Payload 'Important' -QoS 2
-
-9. Test Connection:
-   $isAlive = Test-MqttConnection -Session $session
-   Write-Host ""Connection alive: $isAlive""
-
-10. Get Session Status:
-    $status = Get-MqttSessionStatus -Session $session
-    Write-Host ""Connected: $($status.IsConnected)""
-
-For more details, see: Source/MQTTnet.PowerShell/README.md
-";
-        
-        WriteObject(examples);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/SubscribeMqttTopicCmdlet.cs

@@ -1,55 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet("Subscribe", "MqttTopic")]
-[OutputType(typeof(MqttClientSubscribeResult))]
-public class SubscribeMqttTopicCmdlet : PSCmdlet
-{
-    [Parameter]
-    public bool NoLocal { get; set; }
-
-    [Parameter]
-    public int QoS { get; set; } = 0;
-
-    [Parameter]
-    public bool RetainAsPublished { get; set; }
-
-    [Parameter]
-    public MqttRetainHandling RetainHandling { get; set; } = MqttRetainHandling.SendAtSubscribe;
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public uint SubscriptionIdentifier { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var topicFilter = new MqttTopicFilterBuilder().WithTopic(Topic)
-            .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
-            .WithNoLocal(NoLocal)
-            .WithRetainAsPublished(RetainAsPublished)
-            .WithRetainHandling(RetainHandling)
-            .Build();
-
-        var options = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter);
-
-        if (SubscriptionIdentifier > 0)
-        {
-            options.WithSubscriptionIdentifier(SubscriptionIdentifier);
-        }
-
-        var response = Session.GetClient().SubscribeAsync(options.Build()).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/TestMqttConnectionCmdlet.cs

@@ -1,41 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsDiagnostic.Test, "MqttConnection")]
-[OutputType(typeof(bool))]
-public class TestMqttConnectionCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public int TimeoutSeconds { get; set; } = 5;
-
-    protected override void ProcessRecord()
-    {
-        try
-        {
-            var client = Session.GetClient();
-            
-            if (!client.IsConnected)
-            {
-                WriteObject(false);
-                return;
-            }
-
-            var cts = new CancellationTokenSource(TimeSpan.FromSeconds(TimeoutSeconds));
-            client.PingAsync(cts.Token).GetAwaiter().GetResult();
-            
-            WriteObject(true);
-        }
-        catch
-        {
-            WriteObject(false);
-        }
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/UnregisterMqttMessageHandlerCmdlet.cs

@@ -1,21 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsLifecycle.Unregister, "MqttMessageHandler")]
-public class UnregisterMqttMessageHandlerCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        // Clear all event handlers
-        Session.ClearMessageHandlers();
-        WriteObject($"All message handlers cleared for MQTT session.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/UnsubscribeMqttTopicCmdlet.cs

@@ -1,23 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet("Unsubscribe", "MqttTopic")]
-[OutputType(typeof(MqttClientUnsubscribeResult))]
-public class UnsubscribeMqttTopicCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var options = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(Topic).Build();
-
-        var response = Session.GetClient().UnsubscribeAsync(options).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/IMPLEMENTATION.md

@@ -1,197 +0,0 @@
-# PowerShell Module Extension - Implementation Summary
-
-## Overview
-This implementation extends the PowerShell support for the MQTTnet library as requested in issue #2215, implementing the proposal described in the ticket comments and exposing additional features of the library as PowerShell cmdlets.
-
-## Changes Implemented
-
-### 1. Enhanced RegisterMqttMessageHandler (Primary Requirement from #2215)
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/RegisterMqttMessageHandlerCmdlet.cs`
-
-Implemented the proposal from the issue comment by @ArakniD:
-- Proper ScriptBlock execution in the correct runspace
-- Asynchronous message handling using Task.Run to avoid blocking
-- Passes three parameters to the handler: topic (string), payload (string), and full message object
-- Proper error handling with try-catch and error output to host
-- Captures the runspace during cmdlet execution and restores it during handler execution
-
-**Usage Example:**
-```powershell
-Register-MqttMessageHandler -Session $session -Action {
-    param($topic, $payload, $message)
-    Write-Host "Received on $topic: $payload"
-    # Can access $message.QoS, $message.Retain, etc.
-}
-```
-
-### 2. Enhanced Connection Options
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/ConnectMqttSessionCmdlet.cs`
-
-Extended Connect-MqttSession with:
-- **WebSocket Support**: New `-WebSocketUri` parameter for WebSocket connections
-- **TLS/SSL Support**: New `-UseTls` and `-AllowUntrustedCertificates` parameters
-- **Last Will and Testament**: Parameters for `-WillTopic`, `-WillPayload`, `-WillQoS`, `-WillRetain`
-- **Connection Tuning**: Parameters for `-KeepAlivePeriod` and `-Timeout`
-- **Parameter Sets**: Organized into TCP, TcpWithTls, and WebSocket parameter sets
-
-**Usage Examples:**
-```powershell
-# TCP with TLS
-Connect-MqttSession -Session $session -Host "broker.com" -Port 8883 -UseTls
-
-# WebSocket
-Connect-MqttSession -Session $session -WebSocketUri "ws://broker.com:8000/mqtt"
-
-# With Last Will
-Connect-MqttSession -Session $session -Host "broker.com" `
-    -WillTopic "status" -WillPayload "Offline" -WillRetain
-```
-
-### 3. Enhanced Publishing
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/PublishMqttMessageCmdlet.cs`
-
-Added support for:
-- **User Properties**: New `-UserProperties` parameter accepting a Hashtable
-- MQTT 5 features already supported: ContentType, ResponseTopic, TopicAlias, MessageExpiryInterval
-
-**Usage Example:**
-```powershell
-$props = @{ "source" = "PowerShell"; "version" = "1.0" }
-Publish-MqttMessage -Session $session -Topic "data" -Payload "test" -UserProperties $props
-```
-
-### 4. New Utility Cmdlets
-
-#### Get-MqttSessionStatus
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/GetMqttSessionStatusCmdlet.cs`
-
-Returns connection status information.
-
-```powershell
-$status = Get-MqttSessionStatus -Session $session
-Write-Host "Connected: $($status.IsConnected)"
-```
-
-#### Test-MqttConnection
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/TestMqttConnectionCmdlet.cs`
-
-Pings the broker to verify connectivity.
-
-```powershell
-$isAlive = Test-MqttConnection -Session $session -TimeoutSeconds 5
-```
-
-#### Unregister-MqttMessageHandler
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/UnregisterMqttMessageHandlerCmdlet.cs`
-
-Clears all registered message handlers.
-
-```powershell
-Unregister-MqttMessageHandler -Session $session
-```
-
-#### Show-MqttExamples
-**File**: `Source/MQTTnet.PowerShell/Cmdlets/ShowMqttExamplesCmdlet.cs`
-
-Displays quick reference examples for common MQTT tasks.
-
-```powershell
-Show-MqttExamples
-```
-
-### 5. Enhanced PsMqttSession Class
-**File**: `Source/MQTTnet.PowerShell/PsMqttSession.cs`
-
-Added:
-- `ClearMessageHandlers()` method to remove all event handlers
-- Improved event handling infrastructure
-
-### 6. Documentation
-**Files**: 
-- `Source/MQTTnet.PowerShell/README.md` (new)
-- `Source/MQTTnet.PowerShell/Test.ps1` (updated)
-- `Source/MQTTnet.PowerShell/SmokeTest.ps1` (new)
-
-Created comprehensive documentation including:
-- Quick start guide
-- Detailed cmdlet documentation
-- Parameter descriptions
-- Usage examples for all features
-- Advanced scenarios (request-response, error handling)
-- Tips and best practices
-- Smoke tests for validation
-
-## Complete Cmdlet List (13 Total)
-
-1. **New-MqttSession** - Create MQTT session
-2. **Connect-MqttSession** - Connect with TCP/WebSocket/TLS support
-3. **Disconnect-MqttSession** - Clean disconnect
-4. **Remove-MqttSession** - Cleanup resources
-5. **Get-MqttSessionStatus** - Check connection status
-6. **Test-MqttConnection** - Ping broker
-7. **Publish-MqttMessage** - Publish with advanced features
-8. **Subscribe-MqttTopic** - Subscribe to topics
-9. **Unsubscribe-MqttTopic** - Unsubscribe from topics
-10. **Receive-MqttMessage** - Synchronous message receive
-11. **Register-MqttMessageHandler** - Async message handling ⭐ (Primary requirement)
-12. **Unregister-MqttMessageHandler** - Clear handlers
-13. **Show-MqttExamples** - Quick reference examples
-
-## Testing
-
-### Build Status
-✅ PowerShell module builds successfully
-✅ No compilation errors or warnings
-✅ Code passes CodeQL security analysis (0 alerts)
-
-### Smoke Tests
-✅ All 13 cmdlets load correctly
-✅ Session creation and disposal works
-✅ Show-MqttExamples returns help text
-✅ No runtime errors during basic operations
-
-### Test Files
-- `SmokeTest.ps1`: Quick validation of module loading and basic operations
-- `Test.ps1`: Comprehensive examples demonstrating all features
-
-## Features Not Implemented (Out of Scope)
-
-The following were considered but determined to be out of scope for this task:
-- MQTT broker/server cmdlets (only client features were requested)
-- Managed MQTT client wrapper (existing client is sufficient)
-- RPC extension support (specialized use case)
-- File-based certificate handling cmdlets (can be done with existing .NET capabilities)
-
-## Compatibility
-
-- Requires .NET 8.0 or later
-- Requires PowerShell 7.0 or later
-- Works on Windows, Linux, and macOS
-- Compatible with MQTT 3.1.1 and MQTT 5.0 protocols
-
-## Breaking Changes
-
-None. This is a new feature addition that doesn't modify existing code outside the PowerShell module.
-
-## Migration Guide
-
-For users who were using the initial implementation from PR #2215:
-- The RegisterMqttMessageHandler now actually works! Update your scripts to use the handler.
-- Connect-MqttSession now requires explicit parameter sets (-Host for TCP, -WebSocketUri for WebSocket)
-- No other breaking changes
-
-## Future Enhancements
-
-Potential future improvements that could be added:
-1. Pipeline support for batch publishing
-2. Managed client wrapper for automatic reconnection
-3. Certificate-based authentication helpers
-4. MQTT broker cmdlets for running local brokers
-5. Performance counters and statistics cmdlets
-6. Async/await support for better PowerShell integration
-
-## References
-
-- Issue #2215: https://github.com/dotnet/MQTTnet/pull/2215
-- Proposal comment by @ArakniD: https://github.com/dotnet/MQTTnet/pull/2215#issuecomment-3594492551
-- MQTTnet Documentation: https://github.com/dotnet/MQTTnet

Source/MQTTnet.PowerShell/MQTTnet.PowerShell.csproj

@@ -1,19 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <AssemblyName>MQTTnet.PowerShell</AssemblyName>
-    <LangVersion>default</LangVersion>
-    <Nullable>enable</Nullable>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="PowerShellStandard.Library" Version="5.1.1">
-      <PrivateAssets>All</PrivateAssets>
-    </PackageReference>
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
-  </ItemGroup>
-
-</Project>

Source/MQTTnet.PowerShell/PsMqttMessage.cs

@@ -1,30 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Management.Automation;
-using MQTTnet.Packets;
-
-namespace MQTTnet.PowerShell;
-
-public class PsMqttPublishResult
-{
-
-}
-
-public class PsMqttConnectResult
-{
-
-}
-
-public class PsMqttMessage
-{
-    public required string? ContentType { get; init; }
-    public required string Payload { get; init; }
-    public int QoS { get; init; }
-    public required byte[] RawPayload { get; init; }
-    public string? ResponseTopic { get; init; }
-    public bool Retain { get; init; }
-    public required string Topic { get; init; }
-    public required List<MqttUserProperty>? UserProperties { get; init; }
-}

Source/MQTTnet.PowerShell/PsMqttSession.cs

@@ -1,62 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System.Buffers;
-using System.Text;
-
-namespace MQTTnet.PowerShell;
-
-public sealed class PsMqttSession : IDisposable
-{
-    readonly IMqttClient _client;
-
-    public PsMqttSession()
-    {
-        var factory = new MqttClientFactory();
-        _client = factory.CreateMqttClient();
-
-        _client.ApplicationMessageReceivedAsync += e =>
-        {
-            MessageReceived?.Invoke(
-                this,
-                new PsMqttMessage
-                {
-                    Topic = e.ApplicationMessage.Topic,
-                    Payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
-                    RawPayload = e.ApplicationMessage.Payload.ToArray(),
-                    UserProperties = e.ApplicationMessage.UserProperties,
-                    QoS = (int)e.ApplicationMessage.QualityOfServiceLevel,
-                    ContentType = e.ApplicationMessage.ContentType,
-                    Retain = e.ApplicationMessage.Retain,
-                    ResponseTopic = e.ApplicationMessage.ResponseTopic
-                });
-
-            return Task.CompletedTask;
-        };
-    }
-
-    public event EventHandler<PsMqttMessage>? MessageReceived;
-
-    public void ClearMessageHandlers()
-    {
-        // Remove all event subscribers
-        if (MessageReceived != null)
-        {
-            foreach (var handler in MessageReceived.GetInvocationList())
-            {
-                MessageReceived -= (EventHandler<PsMqttMessage>)handler;
-            }
-        }
-    }
-
-    public void Dispose()
-    {
-        _client.Dispose();
-    }
-
-    public IMqttClient GetClient()
-    {
-        return _client;
-    }
-}

Source/MQTTnet.PowerShell/README.md

@@ -1,396 +0,0 @@
-# MQTTnet PowerShell Module
-
-A PowerShell module for MQTTnet that provides cmdlets for MQTT communication in PowerShell.
-
-## Features
-
-- **Full MQTT Protocol Support**: Connect, subscribe, publish, and manage MQTT connections
-- **TCP and WebSocket Transports**: Support for both TCP and WebSocket connections
-- **TLS/SSL Support**: Secure connections with TLS encryption
-- **Async Message Handling**: Register PowerShell ScriptBlocks as message handlers
-- **Advanced MQTT Features**: 
-  - Quality of Service (QoS) levels 0, 1, and 2
-  - Retained messages
-  - Last Will and Testament (LWT)
-  - User properties
-  - Message expiry intervals
-  - Response topics
-
-## Installation
-
-1. Build the MQTTnet.PowerShell project:
-   ```powershell
-   dotnet build Source/MQTTnet.PowerShell/MQTTnet.PowerShell.csproj
-   ```
-
-2. Import the module:
-   ```powershell
-   Import-Module ./Source/MQTTnet.PowerShell/bin/Debug/net8.0/MQTTnet.PowerShell.dll
-   ```
-
-## Quick Start
-
-### Basic Connection and Publish
-
-```powershell
-# Create a new MQTT session
-$session = New-MqttSession
-
-# Connect to a broker
-Connect-MqttSession -Session $session -Host "broker.hivemq.com" -Port 1883
-
-# Publish a message
-Publish-MqttMessage -Session $session -Topic "test/topic" -Payload "Hello MQTT!"
-
-# Disconnect and cleanup
-Disconnect-MqttSession -Session $session
-Remove-MqttSession -Session $session
-```
-
-### Subscribe and Receive Messages
-
-```powershell
-$session = New-MqttSession
-Connect-MqttSession -Session $session -Host "broker.hivemq.com"
-
-# Subscribe to a topic
-Subscribe-MqttTopic -Session $session -Topic "test/demo"
-
-# Wait for a message (with timeout)
-$message = Receive-MqttMessage -Session $session -TimeoutSeconds 30
-Write-Host "Received: $($message.Payload)"
-
-Disconnect-MqttSession -Session $session
-Remove-MqttSession -Session $session
-```
-
-### Async Message Handler
-
-```powershell
-$session = New-MqttSession
-Connect-MqttSession -Session $session -Host "broker.hivemq.com"
-
-Subscribe-MqttTopic -Session $session -Topic "test/#"
-
-# Register a message handler
-Register-MqttMessageHandler -Session $session -Action {
-    param($topic, $payload, $message)
-    Write-Host "Message on $topic : $payload"
-    Write-Host "QoS: $($message.QoS), Retain: $($message.Retain)"
-}
-
-# Messages will be handled asynchronously
-Start-Sleep -Seconds 60
-
-Disconnect-MqttSession -Session $session
-Remove-MqttSession -Session $session
-```
-
-## Available Cmdlets
-
-### Help and Examples
-
-#### Show-MqttExamples
-Displays quick examples for common MQTT tasks.
-
-```powershell
-Show-MqttExamples
-```
-
-### Session Management
-
-#### New-MqttSession
-Creates a new MQTT session.
-
-```powershell
-$session = New-MqttSession
-```
-
-#### Connect-MqttSession
-Connects to an MQTT broker with various options.
-
-**TCP Connection:**
-```powershell
-Connect-MqttSession -Session $session -Host "broker.hivemq.com" -Port 1883
-```
-
-**TCP with TLS:**
-```powershell
-Connect-MqttSession -Session $session -Host "broker.hivemq.com" -Port 8883 `
-    -UseTls -AllowUntrustedCertificates
-```
-
-**WebSocket Connection:**
-```powershell
-Connect-MqttSession -Session $session -WebSocketUri "ws://broker.hivemq.com:8000/mqtt"
-```
-
-**With Authentication:**
-```powershell
-Connect-MqttSession -Session $session -Host "broker.hivemq.com" `
-    -Username "user" -Password "pass"
-```
-
-**With Last Will and Testament:**
-```powershell
-Connect-MqttSession -Session $session -Host "broker.hivemq.com" `
-    -WillTopic "status/client" -WillPayload "Offline" -WillQoS 1 -WillRetain
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `Host`: Broker hostname (for TCP)
-- `Port`: Broker port (default: 1883)
-- `WebSocketUri`: WebSocket URI (for WebSocket connections)
-- `Username`: Username for authentication
-- `Password`: Password for authentication
-- `ClientId`: Client identifier (auto-generated if not specified)
-- `CleanSession`: Use clean session (default: true)
-- `UseTls`: Enable TLS encryption
-- `AllowUntrustedCertificates`: Allow untrusted certificates
-- `KeepAlivePeriod`: Keep-alive period in seconds (default: 15)
-- `Timeout`: Connection timeout in seconds (default: 10)
-- `WillTopic`: Last Will topic
-- `WillPayload`: Last Will payload
-- `WillQoS`: Last Will QoS level (0-2)
-- `WillRetain`: Retain Last Will message
-
-#### Get-MqttSessionStatus
-Gets the current status of the MQTT session.
-
-```powershell
-$status = Get-MqttSessionStatus -Session $session
-Write-Host "Connected: $($status.IsConnected)"
-```
-
-#### Disconnect-MqttSession
-Disconnects from the MQTT broker.
-
-```powershell
-Disconnect-MqttSession -Session $session
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `Reason`: Disconnect reason code
-- `ReasonString`: Human-readable reason
-- `SessionExpiryInterval`: Session expiry interval
-
-#### Remove-MqttSession
-Disposes the MQTT session and releases resources.
-
-```powershell
-Remove-MqttSession -Session $session
-```
-
-#### Test-MqttConnection
-Tests if the MQTT connection is alive by sending a PING packet.
-
-```powershell
-$isAlive = Test-MqttConnection -Session $session -TimeoutSeconds 5
-Write-Host "Connection is alive: $isAlive"
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `TimeoutSeconds`: Timeout for the ping operation (default: 5)
-
-### Publishing
-
-#### Publish-MqttMessage
-Publishes a message to a topic.
-
-**Basic:**
-```powershell
-Publish-MqttMessage -Session $session -Topic "test/topic" -Payload "Hello"
-```
-
-**With QoS and Retain:**
-```powershell
-Publish-MqttMessage -Session $session -Topic "test/topic" -Payload "Hello" `
-    -QoS 1 -Retain
-```
-
-**With User Properties:**
-```powershell
-$props = @{
-    "source" = "PowerShell"
-    "version" = "1.0"
-}
-Publish-MqttMessage -Session $session -Topic "test/topic" -Payload "Hello" `
-    -UserProperties $props
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `Topic`: Topic to publish to (mandatory)
-- `Payload`: Message payload (mandatory)
-- `QoS`: Quality of Service level (0-2, default: 0)
-- `Retain`: Retain flag
-- `ContentType`: Content type of the payload
-- `ResponseTopic`: Response topic for request/response pattern
-- `UserProperties`: Hashtable of user properties
-- `TopicAlias`: Topic alias
-- `MessageExpiryInterval`: Message expiry interval in seconds
-
-### Subscribing
-
-#### Subscribe-MqttTopic
-Subscribes to a topic.
-
-```powershell
-Subscribe-MqttTopic -Session $session -Topic "test/demo" -QoS 1
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `Topic`: Topic filter to subscribe to (mandatory)
-- `QoS`: Quality of Service level (0-2, default: 0)
-- `NoLocal`: No local flag
-- `RetainAsPublished`: Retain as published flag
-- `RetainHandling`: Retain handling option
-- `SubscriptionIdentifier`: Subscription identifier
-
-#### Unsubscribe-MqttTopic
-Unsubscribes from a topic.
-
-```powershell
-Unsubscribe-MqttTopic -Session $session -Topic "test/demo"
-```
-
-### Message Handling
-
-#### Receive-MqttMessage
-Waits for and receives a single message.
-
-```powershell
-# Wait indefinitely
-$message = Receive-MqttMessage -Session $session
-
-# Wait with timeout
-$message = Receive-MqttMessage -Session $session -TimeoutSeconds 30
-```
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `TimeoutSeconds`: Timeout in seconds (0 = no timeout)
-
-#### Register-MqttMessageHandler
-Registers an asynchronous message handler using a PowerShell ScriptBlock.
-
-```powershell
-Register-MqttMessageHandler -Session $session -Action {
-    param($topic, $payload, $message)
-    
-    Write-Host "Topic: $topic"
-    Write-Host "Payload: $payload"
-    Write-Host "QoS: $($message.QoS)"
-    
-    # Access full message properties
-    if ($message.UserProperties) {
-        foreach ($prop in $message.UserProperties) {
-            Write-Host "Property: $($prop.Name) = $($prop.Value)"
-        }
-    }
-}
-```
-
-The ScriptBlock receives three parameters:
-1. `$topic`: The message topic (string)
-2. `$payload`: The message payload as UTF-8 string
-3. `$message`: The full message object with properties:
-   - `Topic`: Message topic
-   - `Payload`: Payload as string
-   - `RawPayload`: Payload as byte array
-   - `QoS`: Quality of Service level
-   - `Retain`: Retain flag
-   - `ContentType`: Content type
-   - `ResponseTopic`: Response topic
-   - `UserProperties`: List of user properties
-
-**Parameters:**
-- `Session`: The MQTT session (mandatory)
-- `Action`: ScriptBlock to execute for each message (mandatory)
-
-#### Unregister-MqttMessageHandler
-Removes all registered message handlers.
-
-```powershell
-Unregister-MqttMessageHandler -Session $session
-```
-
-## Advanced Examples
-
-### Request-Response Pattern
-
-```powershell
-$session = New-MqttSession
-Connect-MqttSession -Session $session -Host "broker.hivemq.com"
-
-# Responder
-Subscribe-MqttTopic -Session $session -Topic "request/topic"
-Register-MqttMessageHandler -Session $session -Action {
-    param($topic, $payload, $message)
-    if ($message.ResponseTopic) {
-        $response = "Response to: $payload"
-        Publish-MqttMessage -Session $session -Topic $message.ResponseTopic `
-            -Payload $response
-    }
-}
-
-# Requestor (in another session)
-$responseSession = New-MqttSession
-Connect-MqttSession -Session $responseSession -Host "broker.hivemq.com"
-Subscribe-MqttTopic -Session $responseSession -Topic "response/topic"
-Publish-MqttMessage -Session $responseSession -Topic "request/topic" `
-    -Payload "Request data" -ResponseTopic "response/topic"
-
-# Wait for response
-$response = Receive-MqttMessage -Session $responseSession -TimeoutSeconds 10
-```
-
-### High QoS Publishing
-
-```powershell
-# Publish with QoS 2 (exactly once delivery)
-$result = Publish-MqttMessage -Session $session -Topic "critical/data" `
-    -Payload "Important data" -QoS 2
-
-Write-Host "Publish result: $($result.ReasonCode)"
-```
-
-### Error Handling in Message Handler
-
-```powershell
-Register-MqttMessageHandler -Session $session -Action {
-    param($topic, $payload, $message)
-    try {
-        # Process message
-        $data = ConvertFrom-Json $payload
-        # Do something with data
-    }
-    catch {
-        Write-Host "Error processing message: $_" -ForegroundColor Red
-    }
-}
-```
-
-## Tips and Best Practices
-
-1. **Always disconnect cleanly**: Use `Disconnect-MqttSession` before `Remove-MqttSession`
-2. **Use appropriate QoS levels**: QoS 0 for best performance, QoS 1 for reliability, QoS 2 for critical data
-3. **Handle errors in message handlers**: Wrap handler code in try-catch blocks
-4. **Use meaningful client IDs**: Set custom client IDs for debugging
-5. **Secure connections**: Use TLS for production environments
-6. **Topic design**: Use hierarchical topic structures like `device/sensor/temperature`
-7. **Clean session**: Use `-CleanSession:$false` to maintain subscriptions across reconnects
-
-## Requirements
-
-- .NET 8.0 or later
-- PowerShell 7.0 or later
-- MQTTnet library
-
-## License
-
-This project is licensed under the MIT License - see the LICENSE file in the repository root for details.

Source/MQTTnet.PowerShell/SmokeTest.ps1

@@ -1,25 +0,0 @@
-# Quick smoke test for MQTTnet PowerShell Module
-Import-Module ./bin/Debug/net8.0/MQTTnet.PowerShell.dll -WarningAction SilentlyContinue
-
-Write-Host "Testing module load..." -ForegroundColor Yellow
-$commands = Get-Command -Module MQTTnet.PowerShell
-Write-Host "Found $($commands.Count) cmdlets:" -ForegroundColor Green
-$commands | ForEach-Object { Write-Host "  - $($_.Name)" -ForegroundColor Cyan }
-
-Write-Host "`nTesting session creation..." -ForegroundColor Yellow
-$session = New-MqttSession
-Write-Host "Session created successfully!" -ForegroundColor Green
-
-Write-Host "`nTesting Show-MqttExamples..." -ForegroundColor Yellow
-$examples = Show-MqttExamples
-if ($examples) {
-    Write-Host "Examples retrieved successfully!" -ForegroundColor Green
-} else {
-    Write-Host "Failed to retrieve examples!" -ForegroundColor Red
-}
-
-Write-Host "`nTesting session cleanup..." -ForegroundColor Yellow
-Remove-MqttSession -Session $session
-Write-Host "Session removed successfully!" -ForegroundColor Green
-
-Write-Host "`n✓ All smoke tests passed!" -ForegroundColor Green

Source/MQTTnet.PowerShell/Test.ps1

@@ -1,117 +0,0 @@
-# MQTTnet PowerShell Module Test Script
-# This script demonstrates the usage of the MQTTnet PowerShell module
-
-Import-Module ./bin/Debug/net8.0/MQTTnet.PowerShell.dll
-
-Write-Host "=== MQTTnet PowerShell Module Test ===" -ForegroundColor Cyan
-
-# Example 1: Basic TCP connection
-Write-Host "`n1. Testing basic TCP connection..." -ForegroundColor Yellow
-$session = New-MqttSession
-
-try {
-    Connect-MqttSession -Session $session -Host "broker.hivemq.com" -Port 1883
-    Write-Host "Connected successfully!" -ForegroundColor Green
-    
-    # Get session status
-    Write-Host "`n2. Getting session status..." -ForegroundColor Yellow
-    $status = Get-MqttSessionStatus -Session $session
-    Write-Host "IsConnected: $($status.IsConnected)" -ForegroundColor Green
-    
-    # Example 2: Publish a simple message
-    Write-Host "`n3. Publishing a simple message..." -ForegroundColor Yellow
-    Publish-MqttMessage -Session $session -Topic "test/powershell" -Payload "Hello from PowerShell!"
-    Write-Host "Message published!" -ForegroundColor Green
-    
-    # Example 3: Publish with user properties
-    Write-Host "`n4. Publishing with user properties..." -ForegroundColor Yellow
-    $properties = @{
-        "source" = "PowerShell"
-        "version" = "1.0"
-    }
-    Publish-MqttMessage -Session $session -Topic "test/powershell/advanced" -Payload "Advanced message" -UserProperties $properties -QoS 1
-    Write-Host "Message with properties published!" -ForegroundColor Green
-    
-    # Example 4: Subscribe and receive messages
-    Write-Host "`n5. Subscribing to topics..." -ForegroundColor Yellow
-    Subscribe-MqttTopic -Session $session -Topic "test/demo"
-    Write-Host "Subscribed to test/demo" -ForegroundColor Green
-    
-    # Example 5: Register a message handler
-    Write-Host "`n6. Registering message handler..." -ForegroundColor Yellow
-    Register-MqttMessageHandler -Session $session -Action {
-        param($topic, $payload, $message)
-        Write-Host "Received message on topic: $topic" -ForegroundColor Cyan
-        Write-Host "Payload: $payload" -ForegroundColor White
-        Write-Host "QoS: $($message.QoS)" -ForegroundColor White
-    }
-    Write-Host "Handler registered!" -ForegroundColor Green
-    
-    # Publish a test message that the handler will receive
-    Write-Host "`n7. Publishing test message for handler..." -ForegroundColor Yellow
-    Publish-MqttMessage -Session $session -Topic "test/demo" -Payload "Test message for handler"
-    Start-Sleep -Seconds 2
-    
-    # Example 6: Unsubscribe
-    Write-Host "`n8. Unsubscribing from topic..." -ForegroundColor Yellow
-    Unsubscribe-MqttTopic -Session $session -Topic "test/demo"
-    Write-Host "Unsubscribed!" -ForegroundColor Green
-    
-    # Example 7: Clean disconnect
-    Write-Host "`n9. Disconnecting..." -ForegroundColor Yellow
-    Disconnect-MqttSession -Session $session
-    Write-Host "Disconnected!" -ForegroundColor Green
-}
-catch {
-    Write-Host "Error: $_" -ForegroundColor Red
-}
-finally {
-    # Cleanup
-    Remove-MqttSession -Session $session
-    Write-Host "`nSession cleaned up!" -ForegroundColor Green
-}
-
-Write-Host "`n=== WebSocket Connection Example ===" -ForegroundColor Cyan
-
-# Example 8: WebSocket connection
-Write-Host "`nTesting WebSocket connection..." -ForegroundColor Yellow
-$wsSession = New-MqttSession
-
-try {
-    Connect-MqttSession -Session $wsSession -WebSocketUri "ws://broker.hivemq.com:8000/mqtt"
-    Write-Host "Connected via WebSocket!" -ForegroundColor Green
-    
-    Publish-MqttMessage -Session $wsSession -Topic "test/websocket" -Payload "Hello via WebSocket!"
-    Write-Host "Message published via WebSocket!" -ForegroundColor Green
-    
-    Disconnect-MqttSession -Session $wsSession
-}
-catch {
-    Write-Host "WebSocket Error: $_" -ForegroundColor Red
-}
-finally {
-    Remove-MqttSession -Session $wsSession
-}
-
-Write-Host "`n=== Connection with Will Message Example ===" -ForegroundColor Cyan
-
-# Example 9: Connection with Last Will Testament
-Write-Host "`nTesting connection with Will message..." -ForegroundColor Yellow
-$willSession = New-MqttSession
-
-try {
-    Connect-MqttSession -Session $willSession -Host "broker.hivemq.com" `
-        -WillTopic "test/will" -WillPayload "Client disconnected unexpectedly" -WillQoS 1 -WillRetain
-    Write-Host "Connected with Will message configured!" -ForegroundColor Green
-    
-    Disconnect-MqttSession -Session $willSession
-}
-catch {
-    Write-Host "Will Error: $_" -ForegroundColor Red
-}
-finally {
-    Remove-MqttSession -Session $willSession
-}
-
-Write-Host "`n=== All tests completed ===" -ForegroundColor Cyan
-

Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs

@@ -40,6 +40,7 @@ public sealed class MqttServerClientDisconnectOptionsBuilder
         return this;
     }
 
+    [Obsolete("Please use more performance `WithUserProperty` with ArraySegment<byte> or ReadOnlyMemory<byte> for the value.")]
     public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, string value)
     {
         if (_options.UserProperties == null)
@@ -50,4 +51,26 @@ public sealed class MqttServerClientDisconnectOptionsBuilder
         _options.UserProperties.Add(new MqttUserProperty(name, value));
         return this;
     }
+
+    public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, ReadOnlyMemory<byte> value)
+    {
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+        return this;
+    }
+
+    public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, ArraySegment<byte> value)
+    {
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+        return this;
+    }
 }

Source/MQTTnet.Server/EnhancedAuthentication/ExchangeEnhancedAuthenticationOptionsFactory.cs

@@ -51,6 +51,7 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
         return this;
     }
 
+    [Obsolete("Please use more performance `WithUserProperty` with ArraySegment<byte> or ReadOnlyMemory<byte> for the value.")]
     public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value)
     {
         ArgumentNullException.ThrowIfNull(name);
@@ -64,4 +65,32 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
 
         return this;
     }
+
+    public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, ReadOnlyMemory<byte> value)
+    {
+        ArgumentNullException.ThrowIfNull(name);
+
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+
+        return this;
+    }
+
+    public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, ArraySegment<byte> value)
+    {
+        ArgumentNullException.ThrowIfNull(name);
+
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+
+        return this;
+    }
 }

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

@@ -202,9 +202,14 @@ public sealed class MqttTcpServerListener : IDisposable
 
                 clientCertificate = sslStream.RemoteCertificate as X509Certificate2;
 
+                // TODO: Check why this export is needed. Is there something else in the RemoteCertificate as a X509Certificate2???
                 if (clientCertificate == null && sslStream.RemoteCertificate != null)
                 {
+                    #if NET10_0_OR_GREATER
+                    clientCertificate = X509CertificateLoader.LoadCertificate(sslStream.RemoteCertificate.Export(X509ContentType.Cert));
+                    #else
                     clientCertificate = new X509Certificate2(sslStream.RemoteCertificate.Export(X509ContentType.Cert));
+                    #endif
                 }
             }
 

Source/MQTTnet.Server/Internal/MqttConnectedClient.cs

@@ -218,6 +218,8 @@ public sealed class MqttConnectedClient : IDisposable
         HandleTopicAlias(publishPacket);
 
         var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
+        // Topic aliases are scoped to this network connection and must not be forwarded to other clients.
+        applicationMessage.TopicAlias = 0;
 
         var dispatchApplicationMessageResult =
             await _sessionsManager.DispatchApplicationMessage(Id, UserName, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);

Source/MQTTnet.TestApp/AsyncLockTest.cs

@@ -2,9 +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.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Exceptions;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/ClientFlowTest.cs

@@ -2,8 +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.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/ClientTest.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;
 using System.Net;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -2,7 +2,6 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <DebugType>Full</DebugType>
         <IsPackable>false</IsPackable>
     </PropertyGroup>
 

Source/MQTTnet.TestApp/MessageThroughputTest.cs

@@ -1,10 +1,5 @@
 using MQTTnet.Server;
-using System.Collections.Generic;
-using System;
 using System.Diagnostics;
-using System.Threading;
-using System.Threading.Tasks;
-using System.Linq;
 using MQTTnet.Internal;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/MqttNetConsoleLogger.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.Text;
 using MQTTnet.Diagnostics.Logger;
 

Source/MQTTnet.TestApp/PerformanceTest.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;
 using System.Diagnostics;
-using System.Linq;
 using System.Net;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 

Source/MQTTnet.TestApp/Program.cs

@@ -2,10 +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.Threading;
-using System.Threading.Tasks;
-
 namespace MQTTnet.TestApp;
 
 public static class Program

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -2,10 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
 using System.Security.Authentication;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/ServerAndClientTest.cs

@@ -2,8 +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.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;
 

Source/MQTTnet.TestApp/ServerTest.cs

@@ -2,11 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
-using System.Collections.Generic;
-using System.IO;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/TopicGenerator.cs

@@ -1,5 +1,3 @@
-using System;
-using System.Collections.Generic;
 using MQTTnet.Extensions.TopicTemplate;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.Tests.Extensions.Socks5/AssemblyInfo.cs

@@ -0,0 +1,5 @@
+// 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.
+
+[assembly: DoNotParallelize]

Source/MQTTnet.Tests.Extensions.Socks5/FakeSocks5Server.cs

@@ -0,0 +1,529 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Buffers.Binary;
+using System.Collections.Generic;
+using System.IO;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Tests.Extensions.Socks5;
+
+/// <summary>
+///     Scriptable in-process SOCKS5 server used by the test suite.
+///     Listens on 127.0.0.1 on a random port and runs a single, configurable handshake per
+///     accepted connection. On a successful CONNECT, it opens a TCP connection to the destination
+///     described by the CONNECT request and bidirectionally relays bytes between the client and
+///     the destination, allowing tests to point the SOCKS5 client at the in-memory MQTTnet
+///     broker.
+/// </summary>
+internal sealed class FakeSocks5Server : IAsyncDisposable
+{
+    readonly TcpListener _listener;
+    readonly CancellationTokenSource _cts = new();
+    readonly Task _acceptLoop;
+    readonly List<Task> _connectionTasks = new();
+    readonly object _connectionsLock = new();
+    readonly List<RecordedConnect> _recordedConnects = new();
+    readonly object _recordedLock = new();
+
+    public FakeSocks5Server(FakeSocks5ServerOptions options)
+    {
+        Options = options ?? throw new ArgumentNullException(nameof(options));
+
+        _listener = new TcpListener(IPAddress.Loopback, 0);
+        _listener.Start();
+
+        Port = ((IPEndPoint)_listener.LocalEndpoint).Port;
+
+        _acceptLoop = Task.Run(AcceptLoopAsync);
+    }
+
+    public int Port { get; }
+
+    public FakeSocks5ServerOptions Options { get; }
+
+    public IReadOnlyList<RecordedConnect> RecordedConnects
+    {
+        get
+        {
+            lock (_recordedLock)
+            {
+                return _recordedConnects.ToArray();
+            }
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        try
+        {
+            _cts.Cancel();
+        }
+        catch (ObjectDisposedException)
+        {
+        }
+
+        try
+        {
+            _listener.Stop();
+        }
+        catch (Exception)
+        {
+            // ignore
+        }
+
+        try
+        {
+            await _acceptLoop.ConfigureAwait(false);
+        }
+        catch (Exception)
+        {
+            // ignore
+        }
+
+        Task[] pending;
+        lock (_connectionsLock)
+        {
+            pending = _connectionTasks.ToArray();
+        }
+
+        try
+        {
+            await Task.WhenAll(pending).ConfigureAwait(false);
+        }
+        catch (Exception)
+        {
+            // ignore
+        }
+
+        _cts.Dispose();
+    }
+
+    async Task AcceptLoopAsync()
+    {
+        try
+        {
+            while (!_cts.IsCancellationRequested)
+            {
+                TcpClient client;
+                try
+                {
+                    client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    return;
+                }
+                catch (SocketException)
+                {
+                    return;
+                }
+                catch (ObjectDisposedException)
+                {
+                    return;
+                }
+
+                var task = Task.Run(() => HandleClientAsync(client));
+                lock (_connectionsLock)
+                {
+                    _connectionTasks.Add(task);
+                }
+            }
+        }
+        catch (Exception)
+        {
+            // swallow — diagnostic-only
+        }
+    }
+
+    async Task HandleClientAsync(TcpClient client)
+    {
+        using (client)
+        using (var clientStream = client.GetStream())
+        {
+            try
+            {
+                if (Options.DropAfterAccept)
+                {
+                    return;
+                }
+
+                // Greeting: VER NMETHODS METHODS[]
+                var versionAndCount = new byte[2];
+                await ReadExactlyAsync(clientStream, versionAndCount, _cts.Token).ConfigureAwait(false);
+                if (versionAndCount[0] != 0x05)
+                {
+                    return;
+                }
+
+                var methods = new byte[versionAndCount[1]];
+                await ReadExactlyAsync(clientStream, methods, _cts.Token).ConfigureAwait(false);
+
+                if (Options.GreetingResponseOverride != null)
+                {
+                    await clientStream.WriteAsync(Options.GreetingResponseOverride, _cts.Token).ConfigureAwait(false);
+                    if (Options.DropAfterGreetingResponse)
+                    {
+                        return;
+                    }
+                }
+                else
+                {
+                    byte method = SelectMethod(methods);
+                    await clientStream.WriteAsync(new byte[] { 0x05, method }, _cts.Token).ConfigureAwait(false);
+
+                    if (method == 0xFF)
+                    {
+                        return;
+                    }
+
+                    if (method == 0x02)
+                    {
+                        if (!await HandleAuthAsync(clientStream).ConfigureAwait(false))
+                        {
+                            return;
+                        }
+                    }
+                }
+
+                // CONNECT request: VER CMD RSV ATYP DST.ADDR DST.PORT
+                var header = new byte[4];
+                await ReadExactlyAsync(clientStream, header, _cts.Token).ConfigureAwait(false);
+
+                if (header[0] != 0x05)
+                {
+                    // RFC 1928 §4: VER must be X'05' in every message.
+                    return;
+                }
+
+                if (header[1] != 0x01)
+                {
+                    // RFC 1928 §4: only CONNECT (0x01) is supported by this fake server. BIND (0x02)
+                    // and UDP ASSOCIATE (0x03) are signaled as unsupported per spec via REP=0x07.
+                    await WriteReplyAsync(clientStream, 0x07, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+                    return;
+                }
+
+                string targetHost;
+                IPAddress targetIp = null;
+                byte atyp = header[3];
+                switch (atyp)
+                {
+                    case 0x01:
+                    {
+                        var addr = new byte[4];
+                        await ReadExactlyAsync(clientStream, addr, _cts.Token).ConfigureAwait(false);
+                        targetIp = new IPAddress(addr);
+                        targetHost = targetIp.ToString();
+                        break;
+                    }
+
+                    case 0x03:
+                    {
+                        var lenBuf = new byte[1];
+                        await ReadExactlyAsync(clientStream, lenBuf, _cts.Token).ConfigureAwait(false);
+                        var hostBytes = new byte[lenBuf[0]];
+                        await ReadExactlyAsync(clientStream, hostBytes, _cts.Token).ConfigureAwait(false);
+                        targetHost = Encoding.UTF8.GetString(hostBytes);
+                        break;
+                    }
+
+                    case 0x04:
+                    {
+                        var addr = new byte[16];
+                        await ReadExactlyAsync(clientStream, addr, _cts.Token).ConfigureAwait(false);
+                        targetIp = new IPAddress(addr);
+                        targetHost = targetIp.ToString();
+                        break;
+                    }
+
+                    default:
+                        await WriteReplyAsync(clientStream, 0x08, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+                        return;
+                }
+
+                var portBuf = new byte[2];
+                await ReadExactlyAsync(clientStream, portBuf, _cts.Token).ConfigureAwait(false);
+                var targetPort = BinaryPrimitives.ReadUInt16BigEndian(portBuf);
+
+                lock (_recordedLock)
+                {
+                    _recordedConnects.Add(new RecordedConnect(atyp, targetHost, targetPort));
+                }
+
+                if (Options.ConnectReplyCode != 0x00)
+                {
+                    await WriteReplyAsync(clientStream, Options.ConnectReplyCode, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+                    return;
+                }
+
+                // Open the destination connection and start relaying.
+                TcpClient upstream;
+                try
+                {
+                    upstream = new TcpClient();
+                    if (targetIp != null)
+                    {
+                        await upstream.ConnectAsync(targetIp, targetPort, _cts.Token).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        await upstream.ConnectAsync(targetHost, targetPort, _cts.Token).ConfigureAwait(false);
+                    }
+                }
+                catch (Exception)
+                {
+                    await WriteReplyAsync(clientStream, 0x04, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+                    return;
+                }
+
+                using (upstream)
+                using (var upstreamStream = upstream.GetStream())
+                {
+                    await WriteReplyAsync(clientStream, 0x00, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+
+                    var c2u = RelayAsync(clientStream, upstreamStream, _cts.Token);
+                    var u2c = RelayAsync(upstreamStream, clientStream, _cts.Token);
+                    await Task.WhenAny(c2u, u2c).ConfigureAwait(false);
+                }
+            }
+            catch (Exception)
+            {
+                // swallow — diagnostic-only
+            }
+        }
+    }
+
+    byte SelectMethod(byte[] offered)
+    {
+        var allowedMethods = Options.AllowedMethods;
+        foreach (var m in offered)
+        {
+            foreach (var allowed in allowedMethods)
+            {
+                if (m == allowed)
+                {
+                    return m;
+                }
+            }
+        }
+
+        return 0xFF;
+    }
+
+    async Task<bool> HandleAuthAsync(NetworkStream stream)
+    {
+        var header = new byte[2];
+        await ReadExactlyAsync(stream, header, _cts.Token).ConfigureAwait(false);
+        if (header[0] != 0x01)
+        {
+            return false;
+        }
+
+        var ulen = header[1];
+        var userBytes = new byte[ulen];
+        await ReadExactlyAsync(stream, userBytes, _cts.Token).ConfigureAwait(false);
+
+        var plenBuf = new byte[1];
+        await ReadExactlyAsync(stream, plenBuf, _cts.Token).ConfigureAwait(false);
+        var passBytes = new byte[plenBuf[0]];
+        await ReadExactlyAsync(stream, passBytes, _cts.Token).ConfigureAwait(false);
+
+        var user = Encoding.UTF8.GetString(userBytes);
+        var pass = Encoding.UTF8.GetString(passBytes);
+
+        var ok = Options.ExpectedUsername == user && Options.ExpectedPassword == pass;
+        if (Options.ForceAuthFailure)
+        {
+            ok = false;
+        }
+
+        await stream.WriteAsync(new byte[] { 0x01, ok ? (byte)0x00 : (byte)0x01 }, _cts.Token).ConfigureAwait(false);
+        return ok;
+    }
+
+    static async Task WriteReplyAsync(NetworkStream stream, byte rep, byte replyReserved, byte replyAtyp, string replyHost, ushort replyPort)
+    {
+        // VER REP RSV ATYP BND.ADDR BND.PORT
+        //
+        // RFC 1928 §6 says BND.ADDR contains the "associated IP address" (i.e. the server-side
+        // endpoint it used to reach the destination). Most real SOCKS5 servers reply with
+        // ATYP=0x01 (IPv4) + a zeroed address when they have nothing meaningful to report,
+        // which is the safe default we use here. Tests can override the reply ATYP via
+        // FakeSocks5ServerOptions.ReplyAddressType to exercise the client's reply parser.
+
+        switch (replyAtyp)
+        {
+            case 0x04:
+            {
+                var reply = new byte[4 + 16 + 2];
+                reply[0] = 0x05;
+                reply[1] = rep;
+                reply[2] = replyReserved;
+                reply[3] = 0x04;
+                if (!string.IsNullOrEmpty(replyHost) && IPAddress.TryParse(replyHost, out var ip) && ip.AddressFamily == AddressFamily.InterNetworkV6)
+                {
+                    var bytes = ip.GetAddressBytes();
+                    Buffer.BlockCopy(bytes, 0, reply, 4, 16);
+                }
+                BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(20, 2), replyPort);
+                await stream.WriteAsync(reply).ConfigureAwait(false);
+                return;
+            }
+
+            case 0x03:
+            {
+                var host = string.IsNullOrEmpty(replyHost) ? "0.0.0.0" : replyHost;
+                var hostBytes = Encoding.UTF8.GetBytes(host);
+                if (hostBytes.Length > 255)
+                {
+                    Array.Resize(ref hostBytes, 255);
+                }
+
+                var reply = new byte[4 + 1 + hostBytes.Length + 2];
+                reply[0] = 0x05;
+                reply[1] = rep;
+                reply[2] = replyReserved;
+                reply[3] = 0x03;
+                reply[4] = (byte)hostBytes.Length;
+                Buffer.BlockCopy(hostBytes, 0, reply, 5, hostBytes.Length);
+                BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(5 + hostBytes.Length, 2), replyPort);
+                await stream.WriteAsync(reply).ConfigureAwait(false);
+                return;
+            }
+
+            default:
+            {
+                var reply = new byte[4 + 4 + 2];
+                reply[0] = 0x05;
+                reply[1] = rep;
+                reply[2] = replyReserved;
+                reply[3] = 0x01;
+                if (!string.IsNullOrEmpty(replyHost) && IPAddress.TryParse(replyHost, out var ip) && ip.AddressFamily == AddressFamily.InterNetwork)
+                {
+                    var bytes = ip.GetAddressBytes();
+                    Buffer.BlockCopy(bytes, 0, reply, 4, 4);
+                }
+                BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(8, 2), replyPort);
+                await stream.WriteAsync(reply).ConfigureAwait(false);
+                return;
+            }
+        }
+    }
+
+    static async Task RelayAsync(NetworkStream source, NetworkStream destination, CancellationToken cancellationToken)
+    {
+        var buffer = new byte[16 * 1024];
+        try
+        {
+            while (true)
+            {
+                var read = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+                if (read <= 0)
+                {
+                    return;
+                }
+
+                await destination.WriteAsync(buffer.AsMemory(0, read), cancellationToken).ConfigureAwait(false);
+            }
+        }
+        catch (Exception)
+        {
+            // ignore
+        }
+    }
+
+    static async Task ReadExactlyAsync(NetworkStream stream, byte[] buffer, CancellationToken cancellationToken)
+    {
+        var offset = 0;
+        while (offset < buffer.Length)
+        {
+            var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset), cancellationToken).ConfigureAwait(false);
+            if (read <= 0)
+            {
+                throw new EndOfStreamException();
+            }
+
+            offset += read;
+        }
+    }
+
+    public sealed record RecordedConnect(byte AddressType, string Host, int Port);
+}
+
+internal sealed class FakeSocks5ServerOptions
+{
+    /// <summary>
+    ///     The SOCKS5 auth methods the server will accept, in preference order matched against the
+    ///     client's offered methods. Defaults to no-auth only.
+    /// </summary>
+    public byte[] AllowedMethods { get; set; } = [0x00];
+
+    public string ExpectedUsername { get; set; }
+
+    public string ExpectedPassword { get; set; }
+
+    /// <summary>
+    ///     Forces the username/password sub-negotiation to return STATUS=0x01 regardless of the
+    ///     supplied credentials. Useful for tests that exercise the auth-failure path.
+    /// </summary>
+    public bool ForceAuthFailure { get; set; }
+
+    /// <summary>
+    ///     SOCKS5 reply code for the CONNECT response. <c>0x00</c> means "succeeded" and the
+    ///     server will additionally bridge bytes to the destination.
+    /// </summary>
+    public byte ConnectReplyCode { get; set; }
+
+    /// <summary>
+    ///     Reserved byte included in the CONNECT reply. Defaults to <c>0x00</c> per RFC 1928.
+    /// </summary>
+    public byte ReplyReservedByte { get; set; }
+
+    /// <summary>
+    ///     When set, the server sends this byte sequence instead of the regular greeting response.
+    ///     Used to simulate malformed responses (e.g. wrong version).
+    /// </summary>
+    public byte[] GreetingResponseOverride { get; set; }
+
+    /// <summary>
+    ///     If <c>true</c>, closes the connection right after writing the greeting response
+    ///     override. Used together with <see cref="GreetingResponseOverride"/> for protocol-error
+    ///     tests.
+    /// </summary>
+    public bool DropAfterGreetingResponse { get; set; }
+
+    /// <summary>
+    ///     If <c>true</c>, the server closes the connection immediately after accepting (before
+    ///     reading the greeting). Used to simulate proxy mid-handshake disconnects.
+    /// </summary>
+    public bool DropAfterAccept { get; set; }
+
+    /// <summary>
+    ///     Address type used when constructing CONNECT replies. Defaults to <c>0x01</c> (IPv4),
+    ///     which matches the behavior of real SOCKS5 servers — they always reply with a resolved
+    ///     address, regardless of how the client expressed the destination. Override to
+    ///     <c>0x03</c> or <c>0x04</c> to exercise the client's reply parser.
+    /// </summary>
+    public byte ReplyAddressType { get; set; } = 0x01;
+
+    /// <summary>
+    ///     Bound host included in the CONNECT reply (BND.ADDR). Format depends on
+    ///     <see cref="ReplyAddressType"/>: IPv4/IPv6 textual representation for ATYP 0x01/0x04,
+    ///     a domain name for ATYP 0x03. When <c>null</c> or empty the server writes a zeroed
+    ///     address (0.0.0.0, ::, or "0.0.0.0" respectively).
+    /// </summary>
+    public string ReplyBindHost { get; set; }
+
+    /// <summary>
+    ///     Bound port included in the CONNECT reply (BND.PORT). Defaults to <c>0</c>, which
+    ///     is what real SOCKS5 servers return when no meaningful port is available.
+    /// </summary>
+    public ushort ReplyBindPort { get; set; }
+}

Source/MQTTnet.Tests.Extensions.Socks5/GlobalUsings.cs

@@ -0,0 +1,5 @@
+// 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.
+
+global using Microsoft.VisualStudio.TestTools.UnitTesting;

Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <IsPackable>false</IsPackable>
+        <LangVersion>default</LangVersion>
+        <Nullable>disable</Nullable>
+        <NoWarn>$(NoWarn);CA1707;CA1859</NoWarn>
+        <EnableMSTestRunner>true</EnableMSTestRunner>
+        <RootNamespace>MQTTnet.Tests.Extensions.Socks5</RootNamespace>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="MSTest" Version="4.0.2" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
+        <ProjectReference Include="..\MQTTnet.Server\MQTTnet.Server.csproj" />
+        <ProjectReference Include="..\MQTTnet.Extensions.Socks5\MQTTnet.Extensions.Socks5.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <None Include=".\..\..\global.json"
+                Link="global.json"
+                CopyToOutputDirectory="PreserveNewest"/>
+    </ItemGroup>
+</Project>