Compare commits

...
Author SHA1 Message Date
Jeff 798a0e2250 Validate MQTT 5 SUBSCRIBE identifiers (#2262) 2026-09-30 20:55:47 +02:00
Jeff 6e5351ce75 Reject duplicate MQTT 5 PUBLISH singleton properties (#2261) 2026-09-30 20:25:24 +02:00
Jochen Kühner 483447e783 Fix unobserved task exceptions when pending messages overflow (#2272)
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
2026-09-30 20:23:57 +02:00
Jan Klass 18731d9f02 Drop unintended rendered space in README (#2258)
The line break lead to github rendering a space in `( broker`.
2026-08-09 15:20:52 +02:00
Suhas Holla Karkada Chandrashekar 7be55ce922 Fix default TLS certificate validation (BREAKING CHANGE) (#2254)
Fix default TLS certificate validation
2026-07-03 17:56:41 +02:00
christian 14463d1ef9 Merge remote-tracking branch 'origin/master' 2026-06-21 16:24:00 +02:00
christian eeb0f7a261 Update release notes 2026-06-21 16:23:56 +02:00
Suhas Holla Karkada Chandrashekar 52c8f00ef8 Fix keep alive ping send timeout disconnect handling (#2253)
Fix keep alive ping send timeout disconnect
2026-06-21 16:19:45 +02:00
christian bef2ec6fba Update version 2026-06-20 11:02:42 +02:00
Christian d8f06280d5 Migrate to SLNX (#2252) 2026-06-20 11:00:27 +02:00
christian cf68e5e8d5 Update release notes 2026-06-18 17:55:07 +02:00
Alexander KöpkeandCopilot 68da92fd4b Add support for pluggable stream provider to enable SOCKS5 proxies (#2251)
* Add SOCKS5 proxy support via pluggable stream provider

Introduces MQTTnet.Extensions.Socks5, enabling SOCKS5 proxy support for MQTTnet TCP transport. Adds IMqttClientStreamProvider interface, Socks5StreamProvider implementation, and related configuration options. Updates MqttTcpChannel to support stream providers and TLS layering. Includes unit tests with a fake SOCKS5 server and usage samples. Solution and project references updated accordingly.

* Add SOCKS5 options validation for invalid configurations

* Tighten SOCKS5 RFC compliance checks

---------

Co-authored-by: Copilot <Copilot@users.noreply.github.com>
2026-06-18 17:51:52 +02:00
c1b46b0ae7 Client: Tolerate late acknowledgement packets (#2078, #2079) (#2247)
When a publish, subscribe or unsubscribe request was cancelled or
timed out client-side, the corresponding awaitable was removed from
the packet dispatcher. A subsequent PUBACK / PUBCOMP / SUBACK /
UNSUBACK arriving from the broker then fell through to the default
case and raised an MqttProtocolViolationException, which forced a
disconnect. Under high publish load (issue #2079) and from multiple
threads (issue #2078) this manifested as sporadic connection drops.

Match these acknowledgement packets explicitly and log a warning
when no awaiter is registered, instead of treating it as a protocol
violation. The connection is now preserved.

Also documents the thread-safety guarantees of IMqttClient: publish,
subscribe, unsubscribe and ping operations are safe to invoke
concurrently. Connect / disconnect must still be coordinated.

Co-authored-by: meisa <meisa@MAFDEL>
Co-authored-by: christian <6939810+chkr1011@users.noreply.github.com>
2026-06-16 20:28:40 +02:00
christian 745d1d25fe Update release notes 2026-06-13 18:07:57 +02:00
christian 905f070c37 Reset release notes 2026-06-13 18:06:14 +02:00
Suhas Holla Karkada Chandrashekar bca2ded8eb Fix MQTT v5 topic alias forwarding (#2250) 2026-06-13 18:04:42 +02:00
christian 8321806acf Fix nuget package generation 2026-01-25 11:50:48 +01:00
christian 081ef3077a Trigger actions 2026-01-22 18:18:13 +01:00
christian a3fbb928d6 Improve CI 2025-12-21 16:46:53 +01:00
christian 42ba46b885 Update release notes 2025-12-20 10:26:19 +01:00
Alexander Köpke 07409e3a35 Allow user properties as pre-encoded UTF-8 binary buffer (#2228)
* Add support for binary user properties in MQTT 5.0

* Optimize byte array reversal in SubscriptionTopicHashTests

* Use InvariantCulture for topic string conversion in Load_Tests

* switched to `Microsoft.Testing.Platform`

* Refactor CI workflow to separate signed and unsigned package builds

* Add condition to sign job to prevent execution on external pull requests

* Remove fallback to value as string in MqttUserProperty

* Refactor MqttUserProperty handling to use byte arrays for improved performance and add deprecation warnings for legacy methods
2025-12-18 17:46:35 +01:00
Christian 1a3e97afec Add support for dotnet 10 (#2229)
* Add dotnet10 as target framework and fix breaking changes

* Fix tests

* Fix tests for CI

* Fix typo
2025-12-07 16:24:34 +01:00
Christian d74d83453f Remove unused usings (#2217) 2025-10-21 13:10:09 +02:00
Christian 99e70fdbbb Update samples (#2216)
Separate the topic template samples from the other samples
2025-10-21 12:33:20 +02:00
Christian 535a4e010c Add validations of variable byte integers (#2214)
* Add validations and add docs

* Update release notes

* Fix build
2025-10-20 14:33:50 +02:00
老九andchristian 37c1eecce5 Add Directory.Build.props. (#2120)
* Add Directory.Build.props.

* Adjust Analysis related properties and add conditional enablement for ContinuousIntegrationBuild.

* Adopt new changes in master

---------

Co-authored-by: christian <6939810+chkr1011@users.noreply.github.com>
2025-10-20 13:32:31 +02:00
Christian e791d9aba4 Expose local endpoint in channel adapter (#2213)
* Expose local endpoint

* Update release notes
2025-10-19 23:06:36 +02:00
333 changed files with 4542 additions and 1583 deletions
+15 -13
View File
@@ -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
run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}/certificate.snk
- name: Build package (signed)
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
+31 -6
View File
@@ -1,9 +1,34 @@
<Project>
<PropertyGroup>
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<Product>MQTTnet</Product>
<Authors>The contributors of MQTTnet</Authors>
<Company>The contributors of MQTTnet</Company>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<PropertyGroup>
<TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
<TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
<EnableMSTestRunner>true</EnableMSTestRunner>
</PropertyGroup>
<DelaySign>false</DelaySign>
<SignAssembly>false</SignAssembly>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>disable</Nullable>
</Project>
<NeutralLanguage>en-US</NeutralLanguage>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<EnableNETAnalyzers>true</EnableNETAnalyzers>
<AnalysisLevel>latest</AnalysisLevel>
<AnalysisMode>recommended</AnalysisMode>
<AnalysisModeStyle>default</AnalysisModeStyle>
</PropertyGroup>
<PropertyGroup Condition="'$(GITHUB_ACTIONS)'=='true'">
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
</PropertyGroup>
</Project>
+24
View File
@@ -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>
-90
View File
@@ -1,90 +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
CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
LICENSE = LICENSE
README.md = README.md
Source\ReleaseNotes.md = Source\ReleaseNotes.md
.github\workflows\ci.yml = .github\workflows\ci.yml
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
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
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
EndGlobalSection
EndGlobal
+28
View File
@@ -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>
+3 -3
View File
@@ -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 (
broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
version and CPU architecture.
MQTTnet is a high performance .NET library for MQTT based communication.
It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
It is compatible with mostly any supported .NET Framework version and CPU architecture.
## Features
+1 -1
View File
@@ -490,7 +490,7 @@ m/XriWr/Cq4h/JfB7NTsezVslgkBaoU=
}
}
class SampleClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler
sealed class SampleClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler
{
public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs)
{
+20
View File
@@ -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();
}
}
+86
View File
@@ -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);
}
}
+14 -13
View File
@@ -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"))
.WithTopicTemplate(sampleTemplate.WithParameter("id", "2"), noLocal: true)
.WithTopicTemplate(sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
.WithTopicFilter(t => t.WithTopic("topic/1").WithAtLeastOnceQoS())
.WithTopicFilter(t => t.WithTopic("topic/2").WithAtMostOnceQoS())
.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);
@@ -142,7 +143,7 @@ public static class Client_Subscribe_Samples
response.DumpToConsole();
}
static void ConcurrentProcessingDisableAutoAcknowledge(CancellationToken shutdownToken, IMqttClient mqttClient)
static void ConcurrentProcessingDisableAutoAcknowledge(IMqttClient mqttClient, CancellationToken shutdownToken)
{
/*
* This sample shows how to achieve concurrent processing and not have message AutoAcknowledged
@@ -170,7 +171,7 @@ public static class Client_Subscribe_Samples
};
}
static void ConcurrentProcessingWithLimit(CancellationToken shutdownToken, IMqttClient mqttClient)
static void ConcurrentProcessingWithLimit(IMqttClient mqttClient, CancellationToken shutdownToken)
{
/*
* This sample shows how to achieve concurrent processing, with:
@@ -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();
}
}
+2 -1
View File
@@ -6,6 +6,7 @@
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
using System.Globalization;
using System.Text;
using MQTTnet.Diagnostics.Logger;
@@ -51,7 +52,7 @@ public static class Logger_Samples
mqttEventLogger.LogMessagePublished += (_, args) =>
{
var output = new StringBuilder();
output.AppendLine($">> [{args.LogMessage.Timestamp:O}] [{args.LogMessage.ThreadId}] [{args.LogMessage.Source}] [{args.LogMessage.Level}]: {args.LogMessage.Message}");
output.AppendLine(CultureInfo.InvariantCulture, $">> [{args.LogMessage.Timestamp:O}] [{args.LogMessage.ThreadId}] [{args.LogMessage.Source}] [{args.LogMessage.Level}]: {args.LogMessage.Message}");
if (args.LogMessage.Exception != null)
{
output.AppendLine(args.LogMessage.Exception.ToString());
+7 -5
View File
@@ -8,17 +8,19 @@ namespace MQTTnet.Samples.Helpers;
internal static class ObjectExtensions
{
static readonly JsonSerializerOptions SerializerOptions = new()
{
WriteIndented = true
};
public static TObject DumpToConsole<TObject>(this TObject @object)
{
var output = "NULL";
if (@object != null)
{
output = JsonSerializer.Serialize(@object, new JsonSerializerOptions
{
WriteIndented = true
});
output = JsonSerializer.Serialize(@object, SerializerOptions);
}
Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}");
return @object;
}
+2 -10
View File
@@ -2,25 +2,17 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<IsPackable>false</IsPackable>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<NoWarn>$(NoWarn);CA1707</NoWarn>
</PropertyGroup>
<ItemGroup>
<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"/>
+4 -3
View File
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Globalization;
using System.Reflection;
Console.BackgroundColor = ConsoleColor.White;
@@ -28,7 +29,7 @@ Console.Write("Please choose sample class (press Enter to continue): ");
Console.ResetColor();
var input = Console.ReadLine();
var selectedIndex = int.Parse(input ?? "0");
var selectedIndex = int.Parse(input ?? "0", CultureInfo.InvariantCulture);
var selectedSampleClass = sampleClasses[selectedIndex];
var sampleMethods = selectedSampleClass.GetMethods(BindingFlags.Static | BindingFlags.Public).OrderBy(m => m.Name).ToList();
@@ -45,7 +46,7 @@ Console.Write("Please choose sample (press Enter to continue): ");
Console.ResetColor();
input = Console.ReadLine();
selectedIndex = int.Parse(input ?? "0");
selectedIndex = int.Parse(input ?? "0", CultureInfo.InvariantCulture);
var selectedSampleMethod = sampleMethods[selectedIndex];
Console.WriteLine();
@@ -58,7 +59,7 @@ try
{
if (selectedSampleMethod.Invoke(null, null) is Task task)
{
await task;
await task;
}
}
catch (Exception exception)
+7 -4
View File
@@ -8,6 +8,7 @@
// ReSharper disable EmptyConstructor
// ReSharper disable MemberCanBeMadeStatic.Local
using System.Diagnostics.CodeAnalysis;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
@@ -52,14 +53,14 @@ public static class Server_ASP_NET_Samples
// Inject other services via constructor.
}
public Task OnClientConnected(ClientConnectedEventArgs eventArgs)
public static Task OnClientConnected(ClientConnectedEventArgs eventArgs)
{
Console.WriteLine($"Client '{eventArgs.ClientId}' connected.");
return Task.CompletedTask;
}
public Task ValidateConnection(ValidatingConnectionEventArgs eventArgs)
public static Task ValidateConnection(ValidatingConnectionEventArgs eventArgs)
{
Console.WriteLine($"Client '{eventArgs.ClientId}' wants to connect. Accepting!");
return Task.CompletedTask;
@@ -68,6 +69,7 @@ public static class Server_ASP_NET_Samples
sealed class Startup
{
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController)
{
app.UseRouting();
@@ -88,11 +90,12 @@ public static class Server_ASP_NET_Samples
* Attach event handlers etc. if required.
*/
server.ValidatingConnectionAsync += mqttController.ValidateConnection;
server.ClientConnectedAsync += mqttController.OnClientConnected;
server.ValidatingConnectionAsync += MqttController.ValidateConnection;
server.ClientConnectedAsync += MqttController.OnClientConnected;
});
}
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public void ConfigureServices(IServiceCollection services)
{
services.AddHostedMqttServer(
@@ -9,6 +9,7 @@
// ReSharper disable MemberCanBeMadeStatic.Local
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Samples.Server;
@@ -42,4 +43,32 @@ public static class Server_Intercepting_Samples
Console.ReadLine();
await mqttServer.StopAsync();
}
public static async Task Intercept_Subscription()
{
/*
* This sample starts a simple MQTT server which does not allow to subscribe to '#'
* Start the server and try to subscribe to '#' with a client of choice.
*/
var mqttServerFactory = new MqttServerFactory();
var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
using var mqttServer = mqttServerFactory.CreateMqttServer(mqttServerOptions);
mqttServer.InterceptingSubscriptionAsync += args =>
{
if (args.TopicFilter.Topic.Equals("#", StringComparison.Ordinal))
{
args.Response.ReasonCode = MqttSubscribeReasonCode.NotAuthorized;
}
return CompletedTask.Instance;
};
await mqttServer.StartAsync();
Console.WriteLine("Press Enter to exit.");
Console.ReadLine();
await mqttServer.StopAsync();
}
}
+3 -2
View File
@@ -6,6 +6,7 @@
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
using System.Globalization;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -161,7 +162,7 @@ public static class Server_Simple_Samples
return server;
}
class ConsoleLogger : IMqttNetLogger
sealed class ConsoleLogger : IMqttNetLogger
{
readonly object _consoleSyncRoot = new();
@@ -191,7 +192,7 @@ public static class Server_Simple_Samples
if (parameters?.Length > 0)
{
message = string.Format(message, parameters);
message = string.Format(CultureInfo.InvariantCulture, message, parameters);
}
lock (_consoleSyncRoot)
+2
View File
@@ -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;
}
@@ -1,19 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<NoWarn>$(NoWarn);CA1707</NoWarn>
</PropertyGroup>
<ItemGroup>
@@ -6,7 +6,7 @@ using Microsoft.AspNetCore.Mvc.RazorPages;
namespace MQTTnet.AspTestApp.Pages;
public class IndexModel : PageModel
public partial class IndexModel : PageModel
{
readonly ILogger<IndexModel> _logger;
@@ -17,6 +17,9 @@ public class IndexModel : PageModel
public void OnGet()
{
_logger.LogDebug("OnGet");
LogOnGet();
}
[LoggerMessage(Level = LogLevel.Information, Message = "OnGet")]
private partial void LogOnGet();
}
@@ -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;
@@ -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;
@@ -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;
@@ -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;
@@ -1,59 +1,23 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.AspNetCore</AssemblyName>
<RootNamespace>MQTTnet.AspNetCore</RootNamespace>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.AspNetCore</PackageId>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<PackageTags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</PackageTags>
<NeutralLanguage>en-US</NeutralLanguage>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<PackageIcon>nuget.png</PackageIcon>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<PackageId>MQTTnet.AspNetCore</PackageId>
<PackageTags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</PackageTags>
<PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
<FrameworkReference Include="Microsoft.AspNetCore.App"/>
</ItemGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release'">
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\MQTTnet.Server\MQTTnet.Server.csproj"/>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
@@ -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;
@@ -13,7 +12,7 @@ public sealed class MqttClientConnectionContextFactory : IMqttClientAdapterFacto
{
public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
{
if (options == null) throw new ArgumentNullException(nameof(options));
ArgumentNullException.ThrowIfNull(options);
switch (options.ChannelOptions)
{
@@ -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;
@@ -61,27 +58,6 @@ public sealed class MqttConnectionContext : IMqttChannelAdapter
}
}
public EndPoint RemoteEndPoint
{
get
{
// mqtt over tcp
if (_connection.RemoteEndPoint != null)
{
return _connection.RemoteEndPoint;
}
// mqtt over websocket
var httpFeature = _connection.Features.Get<IHttpConnectionFeature>();
if (httpFeature?.RemoteIpAddress != null)
{
return new IPEndPoint(httpFeature.RemoteIpAddress, httpFeature.RemotePort);
}
return null;
}
}
public bool IsSecureConnection
{
get
@@ -104,8 +80,50 @@ public sealed class MqttConnectionContext : IMqttChannelAdapter
}
}
public EndPoint LocalEndPoint
{
get
{
// mqtt over tcp
if (_connection.LocalEndPoint != null)
{
return _connection.LocalEndPoint;
}
// mqtt over websocket
var httpFeature = _connection.Features.Get<IHttpConnectionFeature>();
if (httpFeature?.LocalIpAddress != null)
{
return new IPEndPoint(httpFeature.LocalIpAddress, httpFeature.LocalPort);
}
return null;
}
}
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public EndPoint RemoteEndPoint
{
get
{
// mqtt over tcp
if (_connection.RemoteEndPoint != null)
{
return _connection.RemoteEndPoint;
}
// mqtt over websocket
var httpFeature = _connection.Features.Get<IHttpConnectionFeature>();
if (httpFeature?.RemoteIpAddress != null)
{
return new IPEndPoint(httpFeature.RemoteIpAddress, httpFeature.RemotePort);
}
return null;
}
}
public async Task ConnectAsync(CancellationToken cancellationToken)
{
if (_connection is SocketConnection tcp && !tcp.IsConnected)
@@ -232,8 +250,8 @@ public sealed class MqttConnectionContext : IMqttChannelAdapter
var span = output.GetSpan(buffer.Length);
buffer.Packet.AsSpan().CopyTo(span);
int offset = buffer.Packet.Count;
buffer.Payload.CopyTo(destination: span.Slice(offset));
var offset = buffer.Packet.Count;
buffer.Payload.CopyTo(span.Slice(offset));
output.Advance(buffer.Length);
}
}
@@ -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;
@@ -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;
@@ -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;
@@ -29,6 +26,6 @@ public static class MqttSubProtocolSelector
ArgumentNullException.ThrowIfNull(requestedSubProtocolValues);
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
return requestedSubProtocolValues.OrderByDescending(p => p.Length).FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
return requestedSubProtocolValues.OrderByDescending(p => p.Length).FirstOrDefault(p => p.ToLowerInvariant().StartsWith("mqtt", StringComparison.InvariantCulture));
}
}
@@ -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;
@@ -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;
@@ -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;
@@ -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;
@@ -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;
@@ -81,7 +77,7 @@ public sealed class SocketConnection : ConnectionContext
IsConnected = true;
}
static Exception ConnectionAborted()
static MqttCommunicationException ConnectionAborted()
{
return new MqttCommunicationException("Connection Aborted");
}
@@ -218,12 +214,16 @@ public sealed class SocketConnection : ConnectionContext
var flushTask = _application.Output.FlushAsync();
FlushResult result;
if (!flushTask.IsCompleted)
{
await flushTask;
result = await flushTask;
}
else
{
result = flushTask.GetAwaiter().GetResult();
}
var result = flushTask.GetAwaiter().GetResult();
if (result.IsCompleted)
{
// Pipe consumer is shut down, do we stop writing
+6 -2
View File
@@ -2,13 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.IO.Pipelines;
using System.Net.Sockets;
namespace MQTTnet.AspNetCore;
public sealed class SocketReceiver
public sealed class SocketReceiver : IDisposable
{
readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new();
@@ -33,4 +32,9 @@ public sealed class SocketReceiver
return _awaitable;
}
public void Dispose()
{
_eventArgs.Dispose();
}
}
+6 -3
View File
@@ -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;
@@ -12,7 +10,7 @@ using System.Runtime.InteropServices;
namespace MQTTnet.AspNetCore;
public sealed class SocketSender
public sealed class SocketSender : IDisposable
{
readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new();
@@ -28,6 +26,11 @@ public sealed class SocketSender
_eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
}
public void Dispose()
{
_eventArgs.Dispose();
}
public SocketAwaitable SendAsync(in ReadOnlySequence<byte> buffers)
{
if (buffers.IsSingleSegment)
@@ -1,3 +1 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
[assembly: DoNotParallelize]
@@ -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;
@@ -43,7 +40,7 @@ public class AsyncLockBenchmark : BaseBenchmark
if (globalI != tasksCount)
{
throw new Exception($"Code is broken ({globalI})!");
throw new InvalidOperationException($"Code is broken ({globalI})!");
}
}
@@ -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;
@@ -15,7 +12,7 @@ using MQTTnet.Tests.Mockups;
namespace MQTTnet.Benchmarks;
[MemoryDiagnoser]
public sealed class ChannelAdapterBenchmark : BaseBenchmark
public sealed class ChannelAdapterBenchmark : BaseBenchmark, IDisposable
{
MqttChannelAdapter _channelAdapter;
int _iterations;
@@ -86,4 +83,10 @@ public sealed class ChannelAdapterBenchmark : BaseBenchmark
return buffer.ToArray();
}
public void Dispose()
{
_channelAdapter?.Dispose();
_stream?.Dispose();
}
}
@@ -1,20 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<OutputType>Exe</OutputType>
<DebugType>Full</DebugType>
<TargetFramework>net8.0</TargetFramework>
<IsPackable>false</IsPackable>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902;CS8892</NoWarn>
<NoWarn>$(NoWarn);CS8892;CA1707;CA1051;CA1822</NoWarn>
<TreatWarningsAsErrors>false</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
</PropertyGroup>
<ItemGroup>
@@ -1,6 +1,5 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using System;
namespace MQTTnet.Benchmarks;
@@ -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 BenchmarkDotNet.Attributes;
using MQTTnet.Packets;
using MQTTnet.Server;
@@ -13,7 +9,7 @@ using MQTTnet.Server;
namespace MQTTnet.Benchmarks;
[MemoryDiagnoser]
public class MessageDeliveryBenchmark : BaseBenchmark
public sealed class MessageDeliveryBenchmark : BaseBenchmark, IDisposable
{
List<string> _allSubscribedTopics; // Keep track of the subset of topics that are subscribed
CancellationTokenSource _cancellationTokenSource;
@@ -99,10 +95,16 @@ public class MessageDeliveryBenchmark : BaseBenchmark
if (_messagesReceivedCount < _messagesExpectedCount)
{
throw new Exception($"Messages Received Count mismatch, expected {_messagesExpectedCount}, received {_messagesReceivedCount}");
throw new InvalidOperationException($"Messages Received Count mismatch, expected {_messagesExpectedCount}, received {_messagesReceivedCount}");
}
}
public void Dispose()
{
_cancellationTokenSource?.Dispose();
_mqttServer?.Dispose();
}
[GlobalSetup]
public void Setup()
{
@@ -173,7 +175,7 @@ public class MessageDeliveryBenchmark : BaseBenchmark
var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers);
if (topicIndexStep * _numSubscribedTopicsPerSubscriber * _numSubscribers != totalNumTopics)
{
throw new Exception(
throw new InvalidOperationException(
$"The total number of topics must be divisible by the number of subscribed topics across all subscribers. Total number of topics: {totalNumTopics}, topic step: {topicIndexStep}");
}
@@ -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;
@@ -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;
@@ -4,8 +4,6 @@
using System.Buffers;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Channel;
@@ -18,9 +16,9 @@ namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)]
[MemoryDiagnoser]
public class MqttTcpChannelBenchmark : BaseBenchmark
public sealed class MqttTcpChannelBenchmark : BaseBenchmark, IDisposable
{
IMqttChannel _clientChannel;
MqttTcpChannel _clientChannel;
MqttServer _mqttServer;
IMqttChannel _serverChannel;
@@ -84,4 +82,11 @@ public class MqttTcpChannelBenchmark : BaseBenchmark
await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
}
}
public void Dispose()
{
_clientChannel?.Dispose();
_mqttServer?.Dispose();
_serverChannel?.Dispose();
}
}
+1 -4
View File
@@ -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;
@@ -92,7 +89,7 @@ public static class Program
return;
}
_selectedBenchmarkIndex = _benchmarks.FindIndex(b => b.Name.Equals(arguments[0]));
_selectedBenchmarkIndex = _benchmarks.FindIndex(b => b.Name.Equals(arguments[0], StringComparison.Ordinal));
if (_selectedBenchmarkIndex < 0)
{
@@ -4,16 +4,14 @@ 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;
[SimpleJob(RuntimeMoniker.Net60)]
[RPlotExporter, RankColumn]
[MemoryDiagnoser]
public class ReaderExtensionsBenchmark
public sealed class ReaderExtensionsBenchmark : IDisposable, IAsyncDisposable
{
MqttPacketFormatterAdapter _mqttPacketFormatter;
MemoryStream _stream;
@@ -83,4 +81,17 @@ public class ReaderExtensionsBenchmark
}
}
}
public void Dispose()
{
_stream?.Dispose();
}
public async ValueTask DisposeAsync()
{
if (_stream != null)
{
await _stream.DisposeAsync();
}
}
}
@@ -1,7 +1,6 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks;
@@ -1,39 +1,19 @@
using System.Buffers;
using System.IO.Pipelines;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Formatter;
using System;
using System.Buffers;
using System.IO;
using System.IO.Pipelines;
using System.Threading.Tasks;
namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)]
[RPlotExporter, RankColumn]
[RPlotExporter]
[RankColumn]
[MemoryDiagnoser]
public class SendPacketAsyncBenchmark : BaseBenchmark
public sealed class SendPacketAsyncBenchmark : BaseBenchmark, IDisposable, IAsyncDisposable
{
MemoryStream _stream;
MqttPacketBuffer _buffer;
[GlobalSetup]
public void GlobalSetup()
{
_stream = new MemoryStream(1024);
var packet = new ArraySegment<byte>(new byte[10]);
_buffer = new MqttPacketBuffer(packet);
}
[Benchmark(Baseline = true)]
public async ValueTask Before()
{
_stream.Position = 0;
var output = PipeWriter.Create(_stream);
WritePacketBuffer(output, _buffer);
await output.FlushAsync();
}
MemoryStream _stream;
[Benchmark]
public async ValueTask After()
@@ -52,6 +32,37 @@ public class SendPacketAsyncBenchmark : BaseBenchmark
}
}
[Benchmark(Baseline = true)]
public async ValueTask Before()
{
_stream.Position = 0;
var output = PipeWriter.Create(_stream);
WritePacketBuffer(output, _buffer);
await output.FlushAsync();
}
public void Dispose()
{
_stream?.Dispose();
}
public async ValueTask DisposeAsync()
{
if (_stream != null)
{
await _stream.DisposeAsync();
}
}
[GlobalSetup]
public void GlobalSetup()
{
_stream = new MemoryStream(1024);
var packet = new ArraySegment<byte>(new byte[10]);
_buffer = new MqttPacketBuffer(packet);
}
static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
{
// copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter
@@ -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;
@@ -26,7 +23,7 @@ public class SerializerBenchmark : BaseBenchmark
{
MqttPacket _packet;
ArraySegment<byte> _serializedPacket;
IMqttPacketFormatter _serializer;
MqttV3PacketFormatter _serializer;
MqttBufferWriter _bufferWriter;
[GlobalSetup]
@@ -43,7 +40,7 @@ public class SerializerBenchmark : BaseBenchmark
}
[Benchmark]
public void Serialize_10000_Messages()
public void Serialize10000Messages()
{
for (var i = 0; i < 10000; i++)
{
@@ -53,7 +50,7 @@ public class SerializerBenchmark : BaseBenchmark
}
[Benchmark]
public void Deserialize_10000_Messages()
public void Deserialize10000Messages()
{
var channel = new BenchmarkMqttChannel(_serializedPacket);
var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger());
@@ -66,7 +63,7 @@ public class SerializerBenchmark : BaseBenchmark
}
}
class BenchmarkMqttChannel : IMqttChannel
sealed class BenchmarkMqttChannel : IMqttChannel
{
readonly ArraySegment<byte> _buffer;
int _position;
@@ -79,7 +76,9 @@ public class SerializerBenchmark : BaseBenchmark
public EndPoint RemoteEndPoint { get; set; }
public bool IsSecureConnection { get; } = false;
public EndPoint LocalEndPoint { get; set; }
public bool IsSecureConnection { get; }
public X509Certificate2 ClientCertificate { get; set; }
@@ -5,7 +5,6 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks;
@@ -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;
@@ -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;
@@ -1,6 +1,3 @@
using System;
using System.Collections.Generic;
namespace MQTTnet.Benchmarks;
public class TopicGenerator
@@ -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;
@@ -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
@@ -12,7 +10,7 @@ public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClient
{
ArgumentNullException.ThrowIfNull(context);
if (context.MethodName.Contains("/") || context.MethodName.Contains("+") || context.MethodName.Contains("#"))
if (context.MethodName.Contains('/') || context.MethodName.Contains('+') || context.MethodName.Contains('#'))
{
throw new ArgumentException("The method name cannot contain /, + or #.");
}
@@ -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;
@@ -1,53 +1,19 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName>
<RootNamespace>MQTTnet.Extensions.Rpc</RootNamespace>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<Description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.Rpc</PackageId>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<PackageTags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</PackageTags>
<NeutralLanguage>en-US</NeutralLanguage>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<PackageIcon>nuget.png</PackageIcon>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</Description>
<PackageId>MQTTnet.Extensions.Rpc</PackageId>
<PackageTags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</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>
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
</ItemGroup>
@@ -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
@@ -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;
@@ -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;
@@ -14,7 +11,7 @@ public static class MqttRpcClientExtensions
{
public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null)
{
if (client == null) throw new ArgumentNullException(nameof(client));
ArgumentNullException.ThrowIfNull(client);
var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);
@@ -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
@@ -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;
@@ -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>
@@ -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());
}
}
@@ -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);
}
}
@@ -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;
}
@@ -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}";
}
}
@@ -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;
}
}
@@ -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);
}
}
@@ -1,54 +1,24 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Extensions.TopicTemplate</AssemblyName>
<RootNamespace>MQTTnet.Extensions.TopicTemplate</RootNamespace>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<Description>
Provides mqtt topic templating logic to support dispatch,
routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible).
</Description>
<PackageReadmeFile>README.md</PackageReadmeFile>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.TopicTemplate</PackageId>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<PackageTags>MQTT Message Queue Telemetry MQTTClient Messaging Routing AsyncAPI Template</PackageTags>
<NeutralLanguage>en-US</NeutralLanguage>
<PackageIcon>nuget.png</PackageIcon>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>1591;NETSDK1138</NoWarn>
<IsAotCompatible>true</IsAotCompatible>
</PropertyGroup>
<ItemGroup>
<None Include="README.md">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
</ItemGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All" />
<PackageReference Include="System.ValueTuple" Version="4.6.1" />
</ItemGroup>
@@ -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;
@@ -84,13 +80,13 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
{
var filter = TopicFilter;
// append slash if neccessary
if (filter.Length > 0 && !filter.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) && !filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
if (filter.Length > 0 && !filter.EndsWith(MqttTopicFilterComparer.LevelSeparator) && !filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard))
{
filter += MqttTopicFilterComparer.LevelSeparator;
}
// append hash if neccessary
if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard))
{
filter += MqttTopicFilterComparer.MultiLevelWildcard;
}
@@ -161,12 +157,12 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
return new MqttTopicTemplate(MqttTopicFilterComparer.MultiLevelWildcard.ToString());
}
if (root.Contains(MqttTopicFilterComparer.LevelSeparator) && !root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) && !root.EndsWith("}"))
if (root.Contains(MqttTopicFilterComparer.LevelSeparator) && !root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString(), StringComparison.InvariantCulture) && !root.EndsWith('}'))
{
root = root.Substring(0, root.LastIndexOf(MqttTopicFilterComparer.LevelSeparator) + 1);
}
if (root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()))
if (root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString(), StringComparison.InvariantCulture))
{
root += MqttTopicFilterComparer.SingleLevelWildcard;
}
@@ -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;
@@ -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;
}
}
@@ -51,13 +51,39 @@ 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)
{
if (name == null)
ArgumentNullException.ThrowIfNull(name);
if (_options.UserProperties == null)
{
throw new ArgumentNullException(nameof(name));
_options.UserProperties = new List<MqttUserProperty>();
}
_options.UserProperties.Add(new MqttUserProperty(name, value));
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>();
@@ -10,7 +10,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingPacketEventArgs : EventArgs
{
public InterceptingPacketEventArgs(CancellationToken cancellationToken, string clientId, string userName, EndPoint remoteEndPoint, MqttPacket packet, IDictionary sessionItems)
public InterceptingPacketEventArgs(string clientId, string userName, EndPoint remoteEndPoint, MqttPacket packet, IDictionary sessionItems, CancellationToken cancellationToken)
{
CancellationToken = cancellationToken;
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
@@ -8,7 +8,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingPublishEventArgs : EventArgs
{
public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems)
public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, string clientId, string userName, IDictionary sessionItems, CancellationToken cancellationToken)
{
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
CancellationToken = cancellationToken;
@@ -30,11 +30,6 @@ public sealed class InterceptingPublishEventArgs : EventArgs
/// </summary>
public string ClientId { get; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
public bool CloseConnection { get; set; }
/// <summary>
@@ -45,10 +40,15 @@ public sealed class InterceptingPublishEventArgs : EventArgs
/// <summary>
/// Gets the response which will be sent to the client via the PUBACK etc. packets.
/// </summary>
public PublishResponse Response { get; } = new PublishResponse();
public PublishResponse Response { get; } = new();
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary SessionItems { get; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
}
@@ -10,12 +10,12 @@ namespace MQTTnet.Server;
public sealed class InterceptingSubscriptionEventArgs : EventArgs
{
public InterceptingSubscriptionEventArgs(
CancellationToken cancellationToken,
string clientId,
string userName,
MqttSessionStatus session,
MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties)
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
CancellationToken = cancellationToken;
ClientId = clientId;
@@ -36,11 +36,6 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// </summary>
public string ClientId { get; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
/// <summary>
/// Gets or sets whether the broker should close the client connection.
/// </summary>
@@ -61,7 +56,7 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// <summary>
/// Gets the response which will be sent to the client via the SUBACK packet.
/// </summary>
public SubscribeResponse Response { get; } = new SubscribeResponse();
public SubscribeResponse Response { get; } = new();
/// <summary>
/// Gets the current client session.
@@ -79,6 +74,11 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// </summary>
public MqttTopicFilter TopicFilter { get; set; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
/// <summary>
/// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -9,7 +9,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
{
public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties)
public InterceptingUnsubscriptionEventArgs(string clientId, string userName, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties, CancellationToken cancellationToken)
{
CancellationToken = cancellationToken;
ClientId = clientId;
@@ -202,16 +202,21 @@ 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
}
}
var clientHandler = ClientHandler;
if (clientHandler != null)
{
var tcpChannel = new MqttTcpChannel(stream, remoteEndPoint, clientCertificate);
var tcpChannel = new MqttTcpChannel(stream, _localEndPoint, remoteEndPoint, clientCertificate);
var bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax);
var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter);
@@ -131,7 +131,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
// Allow the user to intercept application message...
if (_eventContainer.InterceptingPublishEvent.HasHandlers)
{
var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, senderId, senderUserName, senderSessionItems);
var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, senderId, senderUserName, senderSessionItems, cancellationToken);
if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
{
// This can happen if a topic alias us used but the topic is
@@ -437,7 +437,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
}
}
public void OnSubscriptionsAdded(MqttSession clientSession, List<string> topics)
public void OnSubscriptionsAdded(MqttSession clientSession, List<string> subscriptionsTopics)
{
_sessionsManagementLock.EnterWriteLock();
try
@@ -448,7 +448,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
_subscriberSessions.Add(clientSession);
}
foreach (var topic in topics)
foreach (var topic in subscriptionsTopics)
{
clientSession.AddSubscribedTopic(topic);
}
@@ -81,7 +81,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
return CheckSubscriptionsResult.NotSubscribed;
}
var senderIsReceiver = string.Equals(senderId, _session.Id);
var senderIsReceiver = string.Equals(senderId, _session.Id, StringComparison.Ordinal);
var maxQoSLevel = -1; // Not subscribed.
HashSet<uint> subscriptionIdentifiers = null;
@@ -460,7 +460,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, _session.UserName, new MqttSessionStatus(_session), topicFilter, userProperties);
var eventArgs = new InterceptingSubscriptionEventArgs(_session.Id, _session.UserName, new MqttSessionStatus(_session), topicFilter, userProperties, cancellationToken);
if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
@@ -475,7 +475,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2;
}
if (topicFilter.Topic.StartsWith("$share/"))
if (topicFilter.Topic.StartsWith("$share/", StringComparison.InvariantCulture))
{
eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported;
}
@@ -493,7 +493,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.UserName, _session.Items, topicFilter, userProperties)
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(_session.Id, _session.UserName, _session.Items, topicFilter, userProperties, cancellationToken)
{
Response =
{
@@ -156,7 +156,7 @@ public sealed class MqttConnectedClient : IDisposable
// TODO: Maybe adding a configuration option is requested in the future.
if (disconnectOptions != null)
{
if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Count > 0 ||
!string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
{
// It is very important to send the DISCONNECT packet here BEFORE cancelling the
@@ -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);
@@ -349,7 +351,7 @@ public sealed class MqttConnectedClient : IDisposable
return packet;
}
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, UserName, RemoteEndPoint, packet, Session.Items);
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(Id, UserName, RemoteEndPoint, packet, Session.Items, cancellationToken);
await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
@@ -395,7 +397,7 @@ public sealed class MqttConnectedClient : IDisposable
if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
{
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, UserName, RemoteEndPoint, currentPacket, Session.Items);
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(Id, UserName, RemoteEndPoint, currentPacket, Session.Items, cancellationToken);
await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
currentPacket = interceptingPacketEventArgs.Packet;
processPacket = interceptingPacketEventArgs.ProcessPacket;
@@ -579,4 +581,4 @@ public sealed class MqttConnectedClient : IDisposable
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
}
}
}
}
@@ -7,13 +7,12 @@ using MQTTnet.Internal;
namespace MQTTnet.Server.Internal;
public sealed class MqttRetainedMessagesManager
public sealed class MqttRetainedMessagesManager : IDisposable
{
readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096);
readonly AsyncLock _storageAccessLock = new();
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096);
readonly AsyncLock _storageAccessLock = new();
public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger)
{
@@ -24,6 +23,46 @@ public sealed class MqttRetainedMessagesManager
_logger = logger.WithSource(nameof(MqttRetainedMessagesManager));
}
public async Task ClearMessages()
{
lock (_messages)
{
_messages.Clear();
}
using (await _storageAccessLock.EnterAsync().ConfigureAwait(false))
{
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
public void Dispose()
{
_storageAccessLock.Dispose();
}
public Task<MqttApplicationMessage> GetMessage(string topic)
{
lock (_messages)
{
if (_messages.TryGetValue(topic, out var message))
{
return Task.FromResult(message);
}
}
return Task.FromResult<MqttApplicationMessage>(null);
}
public Task<IList<MqttApplicationMessage>> GetMessages()
{
lock (_messages)
{
var result = new List<MqttApplicationMessage>(_messages.Values);
return Task.FromResult((IList<MqttApplicationMessage>)result);
}
}
public async Task Start()
{
try
@@ -79,8 +118,7 @@ public sealed class MqttRetainedMessagesManager
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
!MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
@@ -113,39 +151,4 @@ public sealed class MqttRetainedMessagesManager
_logger.Error(exception, "Unhandled exception while handling retained messages.");
}
}
public Task<IList<MqttApplicationMessage>> GetMessages()
{
lock (_messages)
{
var result = new List<MqttApplicationMessage>(_messages.Values);
return Task.FromResult((IList<MqttApplicationMessage>)result);
}
}
public Task<MqttApplicationMessage> GetMessage(string topic)
{
lock (_messages)
{
if (_messages.TryGetValue(topic, out var message))
{
return Task.FromResult(message);
}
}
return Task.FromResult<MqttApplicationMessage>(null);
}
public async Task ClearMessages()
{
lock (_messages)
{
_messages.Clear();
}
using (await _storageAccessLock.EnterAsync().ConfigureAwait(false))
{
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
}
+2 -39
View File
@@ -1,56 +1,19 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Server</AssemblyName>
<RootNamespace>MQTTnet.Server</RootNamespace>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<Description>This is the server implementation of MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Server</PackageId>
<DelaySign>false</DelaySign>
<SignAssembly>false</SignAssembly>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<PackageTags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</PackageTags>
<NeutralLanguage>en-US</NeutralLanguage>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<PackageIcon>nuget.png</PackageIcon>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>disable</Nullable>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<IsAotCompatible>true</IsAotCompatible>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
</ItemGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
</ItemGroup>
</Project>
@@ -2,12 +2,14 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics.CodeAnalysis;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server.EnhancedAuthentication;
using MQTTnet.Server.Internal.Adapter;
namespace MQTTnet.Server;
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public sealed class MqttServerFactory
{
public MqttServerFactory() : this(new MqttNetNullLogger())
@@ -5,8 +5,8 @@
using System.Net;
using System.Net.Security;
using System.Security.Authentication;
using MQTTnet.Certificates;
using System.Security.Cryptography.X509Certificates;
using MQTTnet.Certificates;
// ReSharper disable UnusedMember.Global
namespace MQTTnet.Server;
@@ -47,12 +47,22 @@ public class MqttServerOptionsBuilder
return this;
}
/// <summary>
/// Sets the IPv4 network address to bind.
/// Defaults to any available IPv4 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv4 support entirely.
/// </summary>
public MqttServerOptionsBuilder WithDefaultEndpointBoundIPAddress(IPAddress value)
{
_options.DefaultEndpointOptions.BoundInterNetworkAddress = value ?? IPAddress.Any;
return this;
}
/// <summary>
/// Sets the IPv6 network address to bind.
/// Defaults to any available IPv6 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv6 support entirely.
/// </summary>
public MqttServerOptionsBuilder WithDefaultEndpointBoundIPV6Address(IPAddress value)
{
_options.DefaultEndpointOptions.BoundInterNetworkV6Address = value ?? IPAddress.Any;
@@ -77,12 +87,22 @@ public class MqttServerOptionsBuilder
return this;
}
/// <summary>
/// Sets the IPv4 network address to bind.
/// Defaults to any available IPv4 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv4 support entirely.
/// </summary>
public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPAddress(IPAddress value)
{
_options.TlsEndpointOptions.BoundInterNetworkAddress = value;
return this;
}
/// <summary>
/// Sets the IPv6 network address to bind.
/// Defaults to any available IPv6 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv6 support entirely.
/// </summary>
public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPV6Address(IPAddress value)
{
_options.TlsEndpointOptions.BoundInterNetworkV6Address = value;
@@ -95,6 +115,35 @@ public class MqttServerOptionsBuilder
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value, IMqttServerCertificateCredentials credentials = null)
{
ArgumentNullException.ThrowIfNull(value);
_options.TlsEndpointOptions.CertificateProvider = new BlobCertificateProvider(value)
{
Password = credentials?.Password
};
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(X509Certificate2 certificate)
{
ArgumentNullException.ThrowIfNull(certificate);
_options.TlsEndpointOptions.CertificateProvider = new X509CertificateProvider(certificate);
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(ICertificateProvider certificateProvider)
{
ArgumentNullException.ThrowIfNull(certificateProvider);
_options.TlsEndpointOptions.CertificateProvider = certificateProvider;
return this;
}
public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value)
{
_options.TlsEndpointOptions.SslProtocol = value;
@@ -114,12 +163,6 @@ public class MqttServerOptionsBuilder
return this;
}
public MqttServerOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value)
{
_options.PendingMessagesOverflowStrategy = value;
return this;
}
public MqttServerOptionsBuilder WithoutDefaultEndpoint()
{
_options.DefaultEndpointOptions.IsEnabled = false;
@@ -144,6 +187,12 @@ public class MqttServerOptionsBuilder
return this;
}
public MqttServerOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value)
{
_options.PendingMessagesOverflowStrategy = value;
return this;
}
public MqttServerOptionsBuilder WithPersistentSessions(bool value = true)
{
_options.EnablePersistentSessions = value;
@@ -182,33 +231,4 @@ public class MqttServerOptionsBuilder
_options.TlsEndpointOptions.ReuseAddress = true;
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value, IMqttServerCertificateCredentials credentials = null)
{
ArgumentNullException.ThrowIfNull(value);
_options.TlsEndpointOptions.CertificateProvider = new BlobCertificateProvider(value)
{
Password = credentials?.Password
};
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(X509Certificate2 certificate)
{
ArgumentNullException.ThrowIfNull(certificate);
_options.TlsEndpointOptions.CertificateProvider = new X509CertificateProvider(certificate);
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(ICertificateProvider certificateProvider)
{
ArgumentNullException.ThrowIfNull(certificateProvider);
_options.TlsEndpointOptions.CertificateProvider = certificateProvider;
return this;
}
}
@@ -9,13 +9,30 @@ namespace MQTTnet.Server;
public abstract class MqttServerTcpEndpointBaseOptions
{
public bool IsEnabled { get; set; }
/// <summary>
/// Usually the MQTT packets can be sent partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all clients do support this feature and
/// will close the connection when receiving such packets. If such clients are connecting to this
/// server the flag must be set to _false_.
/// </summary>
public bool AllowPacketFragmentation { get; set; } = true;
public int Port { get; set; }
/// <summary>
/// Gets or sets the IPv4 network address to bind.
/// Defaults to any available IPv4 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv4 support entirely.
/// </summary>
public IPAddress BoundInterNetworkAddress { get; set; } = IPAddress.Any;
/// <summary>
/// Gets or sets the IPv6 network address to bind.
/// Defaults to any available IPv6 address of the machine.
/// Set this to `IPAddress.None` to disable the IPv6 support entirely.
/// </summary>
public IPAddress BoundInterNetworkV6Address { get; set; } = IPAddress.IPv6Any;
public int ConnectionBacklog { get; set; } = 100;
public bool NoDelay { get; set; } = true;
public bool IsEnabled { get; set; }
/// <summary>
/// Gets or sets whether the sockets keep alive feature should be used.
@@ -23,13 +40,16 @@ public abstract class MqttServerTcpEndpointBaseOptions
/// </summary>
public bool? KeepAlive { get; set; }
public LingerOption LingerState { get; set; } = new(true, 0);
public bool NoDelay { get; set; } = true;
public int Port { get; set; }
/// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all clients do support this feature and
/// will close the connection when receiving such packets. If such clients are connecting to this
/// server the flag must be set to _false_.
/// This requires admin permissions on Linux.
/// </summary>
public bool AllowPacketFragmentation { get; set; } = true;
public bool ReuseAddress { get; set; }
/// <summary>
/// Gets or sets the TCP keep alive interval.
@@ -48,15 +68,4 @@ public abstract class MqttServerTcpEndpointBaseOptions
/// The value _null_ indicates that the OS and framework defaults should be used.
/// </summary>
public int? TcpKeepAliveTime { get; set; }
public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
public IPAddress BoundInterNetworkAddress { get; set; } = IPAddress.Any;
public IPAddress BoundInterNetworkV6Address { get; set; } = IPAddress.IPv6Any;
/// <summary>
/// This requires admin permissions on Linux.
/// </summary>
public bool ReuseAddress { get; set; }
}
+1 -4
View File
@@ -2,16 +2,13 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Exceptions;
namespace MQTTnet.TestApp;
public sealed class AsyncLockTest
{
public async Task Run()
public static async Task Run()
{
try
{
-2
View File
@@ -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;
-2
View File
@@ -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;
@@ -2,18 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<DebugType>Full</DebugType>
<TargetFramework>net8.0</TargetFramework>
<IsPackable>false</IsPackable>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
</PropertyGroup>
<ItemGroup>
+12 -13
View File
@@ -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;
@@ -14,7 +9,7 @@ namespace MQTTnet.TestApp;
/// the number of messages per second that can be exchanged between publishers and subscriber.
/// Measurements are performed for subscriptions containing no wildcard, a single wildcard or multiple wildcards.
/// </summary>
public class MessageThroughputTest
public sealed class MessageThroughputTest : IDisposable
{
// Change these constants to suit
const int NumPublishers = 5000;
@@ -55,9 +50,9 @@ public class MessageThroughputTest
await Setup();
await Subscribe_to_No_Wildcard_Topics();
await Subscribe_to_Single_Wildcard_Topics();
await Subscribe_to_Multi_Wildcard_Topics();
await SubscribeToNoWildcardTopics();
await SubscribeToSingleWildcardTopics();
await SubscribeToMultiWildcardTopics();
Console.WriteLine();
Console.WriteLine("End message throughput test");
@@ -74,7 +69,7 @@ public class MessageThroughputTest
public async Task Setup()
{
new TopicGenerator().Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out _singleWildcardTopicsByPublisher, out _multiWildcardTopicsByPublisher);
TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out _singleWildcardTopicsByPublisher, out _multiWildcardTopicsByPublisher);
var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
var mqttClientFactory = new MqttClientFactory();
@@ -152,7 +147,7 @@ public class MessageThroughputTest
/// <summary>
/// Measure no-wildcard topic subscription message exchange performance
/// </summary>
public Task Subscribe_to_No_Wildcard_Topics()
public Task SubscribeToNoWildcardTopics()
{
return ProcessMessages(_topicsByPublisher, "no wildcards");
}
@@ -160,7 +155,7 @@ public class MessageThroughputTest
/// <summary>
/// Measure single-wildcard topic subscription message exchange performance
/// </summary>
public Task Subscribe_to_Single_Wildcard_Topics()
public Task SubscribeToSingleWildcardTopics()
{
return ProcessMessages(_singleWildcardTopicsByPublisher, "single wildcard");
}
@@ -168,7 +163,7 @@ public class MessageThroughputTest
/// <summary>
/// Measure multi-wildcard topic subscription message exchange performance
/// </summary>
public Task Subscribe_to_Multi_Wildcard_Topics()
public Task SubscribeToMultiWildcardTopics()
{
return ProcessMessages(_multiWildcardTopicsByPublisher, "multi wildcard");
}
@@ -342,4 +337,8 @@ public class MessageThroughputTest
Console.ResetColor();
}
public void Dispose()
{
_cancellationTokenSource.Dispose();
}
}
@@ -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;
@@ -34,7 +33,9 @@ public static class MqttNetConsoleLogger
static void PrintToConsole(object sender, MqttNetLogMessagePublishedEventArgs e)
{
var output = new StringBuilder();
#pragma warning disable CA1305
output.AppendLine($">> [{e.LogMessage.Timestamp:O}] [{e.LogMessage.ThreadId}] [{e.LogMessage.Source}] [{e.LogMessage.Level}]: {e.LogMessage.Message}");
#pragma warning restore CA1305
if (e.LogMessage.Exception != null)
{
output.AppendLine(e.LogMessage.Exception.ToString());
+1 -5
View File
@@ -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;
@@ -204,7 +200,7 @@ public static class PerformanceTest
};
}
static Task PublishSingleMessage(IMqttClient client, MqttApplicationMessage applicationMessage, ref int count)
static Task<MqttClientPublishResult> PublishSingleMessage(IMqttClient client, MqttApplicationMessage applicationMessage, ref int count)
{
Interlocked.Increment(ref count);
return Task.Run(() => client.PublishAsync(applicationMessage));
+1 -5
View File
@@ -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
@@ -85,7 +81,7 @@ public static class Program
}
else if (pressedKey.KeyChar == 'f')
{
Task.Run(new AsyncLockTest().Run);
Task.Run(AsyncLockTest.Run);
}
Thread.Sleep(Timeout.Infinite);
+1 -4
View File
@@ -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;
@@ -157,7 +154,7 @@ public static class PublicBrokerTest
if (receivedMessage?.Topic != topic || receivedMessage?.ConvertPayloadToString() != "Hello_World")
{
throw new Exception("Message invalid.");
throw new InvalidOperationException("Message invalid.");
}
await client.UnsubscribeAsync(topic).ConfigureAwait(false);

Some files were not shown because too many files have changed in this diff Show More