Compare commits

...
Author SHA1 Message Date
christian 076039abc7 Add binary files to repo for testing purposes 2025-10-24 17:35:12 +02:00
christian 61cbaa06a8 Expose more properties 2025-10-20 21:41:20 +02:00
christian 1114fa041d Add new PowerShell project 2025-10-20 20:52:31 +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
111 changed files with 2513 additions and 736 deletions
+50 -1
View File
@@ -1,9 +1,58 @@
<Project> <Project>
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<Product>MQTTnet</Product>
<Authors>The contributors of MQTTnet</Authors>
<Company>The contributors of MQTTnet</Company>
<Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
<DelaySign>false</DelaySign>
<SignAssembly>false</SignAssembly>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<ImplicitUsings>enable</ImplicitUsings>
<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>
<TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure> <TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
<TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport> <TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
<EnableMSTestRunner>true</EnableMSTestRunner> <EnableMSTestRunner>true</EnableMSTestRunner>
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
</PropertyGroup>
<PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageIcon>nuget.png</PackageIcon>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
</PropertyGroup>
<ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
<None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup Condition="'$(IncludeSymbols)'=='true'">
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
</ItemGroup>
</Project> </Project>
+8 -3
View File
@@ -7,11 +7,12 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet", "Source\MQTTnet\M
EndProject EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}" Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
ProjectSection(SolutionItems) = preProject ProjectSection(SolutionItems) = preProject
.github\workflows\ci.yml = .github\workflows\ci.yml
CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
Directory.Build.props = Directory.Build.props
LICENSE = LICENSE LICENSE = LICENSE
README.md = README.md README.md = README.md
Source\ReleaseNotes.md = Source\ReleaseNotes.md Source\ReleaseNotes.md = Source\ReleaseNotes.md
.github\workflows\ci.yml = .github\workflows\ci.yml
EndProjectSection EndProjectSection
EndProject EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}" Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
@@ -32,6 +33,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.TopicTem
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}" Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}"
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.PowerShell", "Source\MQTTnet.PowerShell\MQTTnet.PowerShell.csproj", "{37B129C5-0231-4EB5-AF13-77959068911C}"
EndProject
Global Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU Debug|Any CPU = Debug|Any CPU
@@ -78,6 +81,10 @@ Global
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|Any CPU.Build.0 = 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.ActiveCfg = Release|Any CPU
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.Build.0 = Release|Any CPU {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.Build.0 = Release|Any CPU
{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.ActiveCfg = Release|Any CPU
{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection EndGlobalSection
GlobalSection(SolutionProperties) = preSolution GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE HideSolutionNode = FALSE
@@ -85,6 +92,4 @@ Global
GlobalSection(ExtensibilityGlobals) = postSolution GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894} SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
EndGlobalSection EndGlobalSection
GlobalSection(NestedProjects) = preSolution
EndGlobalSection
EndGlobal EndGlobal
+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) public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs)
{ {
+2 -2
View File
@@ -142,7 +142,7 @@ public static class Client_Subscribe_Samples
response.DumpToConsole(); 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 * This sample shows how to achieve concurrent processing and not have message AutoAcknowledged
@@ -170,7 +170,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: * This sample shows how to achieve concurrent processing, with:
+2 -1
View File
@@ -6,6 +6,7 @@
// ReSharper disable UnusedMember.Global // ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
using System.Globalization;
using System.Text; using System.Text;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
@@ -51,7 +52,7 @@ public static class Logger_Samples
mqttEventLogger.LogMessagePublished += (_, args) => mqttEventLogger.LogMessagePublished += (_, args) =>
{ {
var output = new StringBuilder(); 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) if (args.LogMessage.Exception != null)
{ {
output.AppendLine(args.LogMessage.Exception.ToString()); output.AppendLine(args.LogMessage.Exception.ToString());
+6 -4
View File
@@ -8,15 +8,17 @@ namespace MQTTnet.Samples.Helpers;
internal static class ObjectExtensions internal static class ObjectExtensions
{ {
static readonly JsonSerializerOptions SerializerOptions = new()
{
WriteIndented = true
};
public static TObject DumpToConsole<TObject>(this TObject @object) public static TObject DumpToConsole<TObject>(this TObject @object)
{ {
var output = "NULL"; var output = "NULL";
if (@object != null) if (@object != null)
{ {
output = JsonSerializer.Serialize(@object, new JsonSerializerOptions output = JsonSerializer.Serialize(@object, SerializerOptions);
{
WriteIndented = true
});
} }
Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}"); Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}");
+1 -10
View File
@@ -2,19 +2,10 @@
<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings> <ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>
<IsPackable>false</IsPackable> <IsPackable>false</IsPackable>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers> <NoWarn>$(NoWarn);CA1707</NoWarn>
<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>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
+3 -2
View File
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System.Globalization;
using System.Reflection; using System.Reflection;
Console.BackgroundColor = ConsoleColor.White; Console.BackgroundColor = ConsoleColor.White;
@@ -28,7 +29,7 @@ Console.Write("Please choose sample class (press Enter to continue): ");
Console.ResetColor(); Console.ResetColor();
var input = Console.ReadLine(); var input = Console.ReadLine();
var selectedIndex = int.Parse(input ?? "0"); var selectedIndex = int.Parse(input ?? "0", CultureInfo.InvariantCulture);
var selectedSampleClass = sampleClasses[selectedIndex]; var selectedSampleClass = sampleClasses[selectedIndex];
var sampleMethods = selectedSampleClass.GetMethods(BindingFlags.Static | BindingFlags.Public).OrderBy(m => m.Name).ToList(); 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(); Console.ResetColor();
input = Console.ReadLine(); input = Console.ReadLine();
selectedIndex = int.Parse(input ?? "0"); selectedIndex = int.Parse(input ?? "0", CultureInfo.InvariantCulture);
var selectedSampleMethod = sampleMethods[selectedIndex]; var selectedSampleMethod = sampleMethods[selectedIndex];
Console.WriteLine(); Console.WriteLine();
+7 -4
View File
@@ -8,6 +8,7 @@
// ReSharper disable EmptyConstructor // ReSharper disable EmptyConstructor
// ReSharper disable MemberCanBeMadeStatic.Local // ReSharper disable MemberCanBeMadeStatic.Local
using System.Diagnostics.CodeAnalysis;
using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
@@ -52,14 +53,14 @@ public static class Server_ASP_NET_Samples
// Inject other services via constructor. // Inject other services via constructor.
} }
public Task OnClientConnected(ClientConnectedEventArgs eventArgs) public static Task OnClientConnected(ClientConnectedEventArgs eventArgs)
{ {
Console.WriteLine($"Client '{eventArgs.ClientId}' connected."); Console.WriteLine($"Client '{eventArgs.ClientId}' connected.");
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task ValidateConnection(ValidatingConnectionEventArgs eventArgs) public static Task ValidateConnection(ValidatingConnectionEventArgs eventArgs)
{ {
Console.WriteLine($"Client '{eventArgs.ClientId}' wants to connect. Accepting!"); Console.WriteLine($"Client '{eventArgs.ClientId}' wants to connect. Accepting!");
return Task.CompletedTask; return Task.CompletedTask;
@@ -68,6 +69,7 @@ public static class Server_ASP_NET_Samples
sealed class Startup sealed class Startup
{ {
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController) public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController)
{ {
app.UseRouting(); app.UseRouting();
@@ -88,11 +90,12 @@ public static class Server_ASP_NET_Samples
* Attach event handlers etc. if required. * Attach event handlers etc. if required.
*/ */
server.ValidatingConnectionAsync += mqttController.ValidateConnection; server.ValidatingConnectionAsync += MqttController.ValidateConnection;
server.ClientConnectedAsync += mqttController.OnClientConnected; server.ClientConnectedAsync += MqttController.OnClientConnected;
}); });
} }
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public void ConfigureServices(IServiceCollection services) public void ConfigureServices(IServiceCollection services)
{ {
services.AddHostedMqttServer( services.AddHostedMqttServer(
@@ -9,6 +9,7 @@
// ReSharper disable MemberCanBeMadeStatic.Local // ReSharper disable MemberCanBeMadeStatic.Local
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
namespace MQTTnet.Samples.Server; namespace MQTTnet.Samples.Server;
@@ -42,4 +43,32 @@ public static class Server_Intercepting_Samples
Console.ReadLine(); Console.ReadLine();
await mqttServer.StopAsync(); 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 UnusedMember.Global
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
using System.Globalization;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
@@ -161,7 +162,7 @@ public static class Server_Simple_Samples
return server; return server;
} }
class ConsoleLogger : IMqttNetLogger sealed class ConsoleLogger : IMqttNetLogger
{ {
readonly object _consoleSyncRoot = new(); readonly object _consoleSyncRoot = new();
@@ -191,7 +192,7 @@ public static class Server_Simple_Samples
if (parameters?.Length > 0) if (parameters?.Length > 0)
{ {
message = string.Format(message, parameters); message = string.Format(CultureInfo.InvariantCulture, message, parameters);
} }
lock (_consoleSyncRoot) lock (_consoleSyncRoot)
@@ -1,19 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk.Web"> <Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings> <ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable> <IsPackable>false</IsPackable>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers> <NoWarn>$(NoWarn);CA1707</NoWarn>
<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>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -6,7 +6,7 @@ using Microsoft.AspNetCore.Mvc.RazorPages;
namespace MQTTnet.AspTestApp.Pages; namespace MQTTnet.AspTestApp.Pages;
public class IndexModel : PageModel public partial class IndexModel : PageModel
{ {
readonly ILogger<IndexModel> _logger; readonly ILogger<IndexModel> _logger;
@@ -17,6 +17,9 @@ public class IndexModel : PageModel
public void OnGet() public void OnGet()
{ {
_logger.LogDebug("OnGet"); LogOnGet();
} }
[LoggerMessage(Level = LogLevel.Information, Message = "OnGet")]
private partial void LogOnGet();
} }
@@ -1,59 +1,23 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.AspNetCore</AssemblyName> <AssemblyName>MQTTnet.AspNetCore</AssemblyName>
<RootNamespace>MQTTnet.AspNetCore</RootNamespace> <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> <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> </PropertyGroup>
<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"/>
<FrameworkReference Include="Microsoft.AspNetCore.App"/> <FrameworkReference Include="Microsoft.AspNetCore.App"/>
</ItemGroup> </ItemGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release'">
</PropertyGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\MQTTnet.Server\MQTTnet.Server.csproj"/> <ProjectReference Include="..\MQTTnet.Server\MQTTnet.Server.csproj"/>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/> <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
@@ -13,7 +13,7 @@ public sealed class MqttClientConnectionContextFactory : IMqttClientAdapterFacto
{ {
public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger) public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); ArgumentNullException.ThrowIfNull(options);
switch (options.ChannelOptions) switch (options.ChannelOptions)
{ {
@@ -61,27 +61,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 public bool IsSecureConnection
{ {
get get
@@ -104,8 +83,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 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) public async Task ConnectAsync(CancellationToken cancellationToken)
{ {
if (_connection is SocketConnection tcp && !tcp.IsConnected) if (_connection is SocketConnection tcp && !tcp.IsConnected)
@@ -232,8 +253,8 @@ public sealed class MqttConnectionContext : IMqttChannelAdapter
var span = output.GetSpan(buffer.Length); var span = output.GetSpan(buffer.Length);
buffer.Packet.AsSpan().CopyTo(span); buffer.Packet.AsSpan().CopyTo(span);
int offset = buffer.Packet.Count; var offset = buffer.Packet.Count;
buffer.Payload.CopyTo(destination: span.Slice(offset)); buffer.Payload.CopyTo(span.Slice(offset));
output.Advance(buffer.Length); output.Advance(buffer.Length);
} }
} }
@@ -29,6 +29,6 @@ public static class MqttSubProtocolSelector
ArgumentNullException.ThrowIfNull(requestedSubProtocolValues); ArgumentNullException.ThrowIfNull(requestedSubProtocolValues);
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc. // 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));
} }
} }
@@ -81,7 +81,7 @@ public sealed class SocketConnection : ConnectionContext
IsConnected = true; IsConnected = true;
} }
static Exception ConnectionAborted() static MqttCommunicationException ConnectionAborted()
{ {
return new MqttCommunicationException("Connection Aborted"); return new MqttCommunicationException("Connection Aborted");
} }
@@ -218,12 +218,16 @@ public sealed class SocketConnection : ConnectionContext
var flushTask = _application.Output.FlushAsync(); var flushTask = _application.Output.FlushAsync();
FlushResult result;
if (!flushTask.IsCompleted) if (!flushTask.IsCompleted)
{ {
await flushTask; result = await flushTask;
}
else
{
result = flushTask.GetAwaiter().GetResult();
} }
var result = flushTask.GetAwaiter().GetResult();
if (result.IsCompleted) if (result.IsCompleted)
{ {
// Pipe consumer is shut down, do we stop writing // Pipe consumer is shut down, do we stop writing
+6 -1
View File
@@ -8,7 +8,7 @@ using System.Net.Sockets;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
public sealed class SocketReceiver public sealed class SocketReceiver : IDisposable
{ {
readonly SocketAwaitable _awaitable; readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new(); readonly SocketAsyncEventArgs _eventArgs = new();
@@ -33,4 +33,9 @@ public sealed class SocketReceiver
return _awaitable; return _awaitable;
} }
public void Dispose()
{
_eventArgs.Dispose();
}
} }
+6 -1
View File
@@ -12,7 +12,7 @@ using System.Runtime.InteropServices;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
public sealed class SocketSender public sealed class SocketSender : IDisposable
{ {
readonly SocketAwaitable _awaitable; readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new(); readonly SocketAsyncEventArgs _eventArgs = new();
@@ -28,6 +28,11 @@ public sealed class SocketSender
_eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError); _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
} }
public void Dispose()
{
_eventArgs.Dispose();
}
public SocketAwaitable SendAsync(in ReadOnlySequence<byte> buffers) public SocketAwaitable SendAsync(in ReadOnlySequence<byte> buffers)
{ {
if (buffers.IsSingleSegment) if (buffers.IsSingleSegment)
@@ -43,7 +43,7 @@ public class AsyncLockBenchmark : BaseBenchmark
if (globalI != tasksCount) if (globalI != tasksCount)
{ {
throw new Exception($"Code is broken ({globalI})!"); throw new InvalidOperationException($"Code is broken ({globalI})!");
} }
} }
@@ -15,7 +15,7 @@ using MQTTnet.Tests.Mockups;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
[MemoryDiagnoser] [MemoryDiagnoser]
public sealed class ChannelAdapterBenchmark : BaseBenchmark public sealed class ChannelAdapterBenchmark : BaseBenchmark, IDisposable
{ {
MqttChannelAdapter _channelAdapter; MqttChannelAdapter _channelAdapter;
int _iterations; int _iterations;
@@ -86,4 +86,10 @@ public sealed class ChannelAdapterBenchmark : BaseBenchmark
return buffer.ToArray(); return buffer.ToArray();
} }
public void Dispose()
{
_channelAdapter?.Dispose();
_stream?.Dispose();
}
} }
@@ -1,20 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<DebugType>Full</DebugType> <DebugType>Full</DebugType>
<TargetFramework>net8.0</TargetFramework>
<IsPackable>false</IsPackable> <IsPackable>false</IsPackable>
<EnableNETAnalyzers>false</EnableNETAnalyzers> <NoWarn>$(NoWarn);CS8892;CA1707;CA1051;CA1822</NoWarn>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902;CS8892</NoWarn>
<TreatWarningsAsErrors>false</TreatWarningsAsErrors> <TreatWarningsAsErrors>false</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -2,10 +2,6 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // 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 BenchmarkDotNet.Attributes;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Server; using MQTTnet.Server;
@@ -13,7 +9,7 @@ using MQTTnet.Server;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
[MemoryDiagnoser] [MemoryDiagnoser]
public class MessageDeliveryBenchmark : BaseBenchmark public sealed class MessageDeliveryBenchmark : BaseBenchmark, IDisposable
{ {
List<string> _allSubscribedTopics; // Keep track of the subset of topics that are subscribed List<string> _allSubscribedTopics; // Keep track of the subset of topics that are subscribed
CancellationTokenSource _cancellationTokenSource; CancellationTokenSource _cancellationTokenSource;
@@ -99,10 +95,16 @@ public class MessageDeliveryBenchmark : BaseBenchmark
if (_messagesReceivedCount < _messagesExpectedCount) 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] [GlobalSetup]
public void Setup() public void Setup()
{ {
@@ -173,7 +175,7 @@ public class MessageDeliveryBenchmark : BaseBenchmark
var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers); var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers);
if (topicIndexStep * _numSubscribedTopicsPerSubscriber * _numSubscribers != totalNumTopics) 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}"); $"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}");
} }
@@ -4,8 +4,6 @@
using System.Buffers; using System.Buffers;
using System.Reflection; using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Channel; using MQTTnet.Channel;
@@ -18,9 +16,9 @@ namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)] [SimpleJob(RuntimeMoniker.Net60)]
[MemoryDiagnoser] [MemoryDiagnoser]
public class MqttTcpChannelBenchmark : BaseBenchmark public sealed class MqttTcpChannelBenchmark : BaseBenchmark, IDisposable
{ {
IMqttChannel _clientChannel; MqttTcpChannel _clientChannel;
MqttServer _mqttServer; MqttServer _mqttServer;
IMqttChannel _serverChannel; IMqttChannel _serverChannel;
@@ -84,4 +82,11 @@ public class MqttTcpChannelBenchmark : BaseBenchmark
await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false); await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
} }
} }
public void Dispose()
{
_clientChannel?.Dispose();
_mqttServer?.Dispose();
_serverChannel?.Dispose();
}
} }
+1 -1
View File
@@ -92,7 +92,7 @@ public static class Program
return; return;
} }
_selectedBenchmarkIndex = _benchmarks.FindIndex(b => b.Name.Equals(arguments[0])); _selectedBenchmarkIndex = _benchmarks.FindIndex(b => b.Name.Equals(arguments[0], StringComparison.Ordinal));
if (_selectedBenchmarkIndex < 0) if (_selectedBenchmarkIndex < 0)
{ {
@@ -13,7 +13,7 @@ namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)] [SimpleJob(RuntimeMoniker.Net60)]
[RPlotExporter, RankColumn] [RPlotExporter, RankColumn]
[MemoryDiagnoser] [MemoryDiagnoser]
public class ReaderExtensionsBenchmark public sealed class ReaderExtensionsBenchmark : IDisposable, IAsyncDisposable
{ {
MqttPacketFormatterAdapter _mqttPacketFormatter; MqttPacketFormatterAdapter _mqttPacketFormatter;
MemoryStream _stream; MemoryStream _stream;
@@ -83,4 +83,17 @@ public class ReaderExtensionsBenchmark
} }
} }
} }
public void Dispose()
{
_stream?.Dispose();
}
public async ValueTask DisposeAsync()
{
if (_stream != null)
{
await _stream.DisposeAsync();
}
}
} }
@@ -1,39 +1,19 @@
using System.Buffers;
using System.IO.Pipelines;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using System;
using System.Buffers;
using System.IO;
using System.IO.Pipelines;
using System.Threading.Tasks;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)] [SimpleJob(RuntimeMoniker.Net60)]
[RPlotExporter, RankColumn] [RPlotExporter]
[RankColumn]
[MemoryDiagnoser] [MemoryDiagnoser]
public class SendPacketAsyncBenchmark : BaseBenchmark public sealed class SendPacketAsyncBenchmark : BaseBenchmark, IDisposable, IAsyncDisposable
{ {
MemoryStream _stream;
MqttPacketBuffer _buffer; MqttPacketBuffer _buffer;
MemoryStream _stream;
[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();
}
[Benchmark] [Benchmark]
public async ValueTask After() 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) static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
{ {
// copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter // copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter
@@ -26,7 +26,7 @@ public class SerializerBenchmark : BaseBenchmark
{ {
MqttPacket _packet; MqttPacket _packet;
ArraySegment<byte> _serializedPacket; ArraySegment<byte> _serializedPacket;
IMqttPacketFormatter _serializer; MqttV3PacketFormatter _serializer;
MqttBufferWriter _bufferWriter; MqttBufferWriter _bufferWriter;
[GlobalSetup] [GlobalSetup]
@@ -43,7 +43,7 @@ public class SerializerBenchmark : BaseBenchmark
} }
[Benchmark] [Benchmark]
public void Serialize_10000_Messages() public void Serialize10000Messages()
{ {
for (var i = 0; i < 10000; i++) for (var i = 0; i < 10000; i++)
{ {
@@ -53,7 +53,7 @@ public class SerializerBenchmark : BaseBenchmark
} }
[Benchmark] [Benchmark]
public void Deserialize_10000_Messages() public void Deserialize10000Messages()
{ {
var channel = new BenchmarkMqttChannel(_serializedPacket); var channel = new BenchmarkMqttChannel(_serializedPacket);
var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger()); var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger());
@@ -66,7 +66,7 @@ public class SerializerBenchmark : BaseBenchmark
} }
} }
class BenchmarkMqttChannel : IMqttChannel sealed class BenchmarkMqttChannel : IMqttChannel
{ {
readonly ArraySegment<byte> _buffer; readonly ArraySegment<byte> _buffer;
int _position; int _position;
@@ -79,7 +79,9 @@ public class SerializerBenchmark : BaseBenchmark
public EndPoint RemoteEndPoint { get; set; } public EndPoint RemoteEndPoint { get; set; }
public bool IsSecureConnection { get; } = false; public EndPoint LocalEndPoint { get; set; }
public bool IsSecureConnection { get; }
public X509Certificate2 ClientCertificate { get; set; } public X509Certificate2 ClientCertificate { get; set; }
@@ -12,7 +12,7 @@ public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClient
{ {
ArgumentNullException.ThrowIfNull(context); 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 #."); throw new ArgumentException("The method name cannot contain /, + or #.");
} }
@@ -1,53 +1,19 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName> <AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName>
<RootNamespace>MQTTnet.Extensions.Rpc</RootNamespace> <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> <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> </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> <ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/> <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
</ItemGroup> </ItemGroup>
@@ -14,7 +14,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) 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); var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);
@@ -1,54 +1,24 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Extensions.TopicTemplate</AssemblyName> <AssemblyName>MQTTnet.Extensions.TopicTemplate</AssemblyName>
<RootNamespace>MQTTnet.Extensions.TopicTemplate</RootNamespace> <RootNamespace>MQTTnet.Extensions.TopicTemplate</RootNamespace>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<Description> <Description>
Provides mqtt topic templating logic to support dispatch, Provides mqtt topic templating logic to support dispatch,
routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible). routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible).
</Description> </Description>
<PackageReadmeFile>README.md</PackageReadmeFile> <PackageReadmeFile>README.md</PackageReadmeFile>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.TopicTemplate</PackageId> <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> <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> <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
<GenerateDocumentationFile>true</GenerateDocumentationFile> <GenerateDocumentationFile>true</GenerateDocumentationFile>
<NoWarn>1591;NETSDK1138</NoWarn>
<IsAotCompatible>true</IsAotCompatible>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <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" /> <PackageReference Include="System.ValueTuple" Version="4.6.1" />
</ItemGroup> </ItemGroup>
@@ -84,13 +84,13 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
{ {
var filter = TopicFilter; var filter = TopicFilter;
// append slash if neccessary // 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; filter += MqttTopicFilterComparer.LevelSeparator;
} }
// append hash if neccessary // append hash if neccessary
if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString())) if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard))
{ {
filter += MqttTopicFilterComparer.MultiLevelWildcard; filter += MqttTopicFilterComparer.MultiLevelWildcard;
} }
@@ -161,12 +161,12 @@ public sealed class MqttTopicTemplate : IEquatable<MqttTopicTemplate>
return new MqttTopicTemplate(MqttTopicFilterComparer.MultiLevelWildcard.ToString()); 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); 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; root += MqttTopicFilterComparer.SingleLevelWildcard;
} }
@@ -0,0 +1,47 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Connect, "MqttSession")]
[OutputType(typeof(MqttClientConnectResult))]
public class ConnectMqttSessionCmdlet : PSCmdlet
{
[Parameter]
public SwitchParameter CleanSession { get; set; } = true;
[Parameter]
public string? ClientId { get; set; } = Guid.NewGuid().ToString();
[Parameter(Mandatory = true)]
public required new string Host { get; set; }
[Parameter]
public string? Password { get; set; }
[Parameter]
public int Port { get; set; } = 1883;
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
public string? Username { get; set; }
[Parameter]
public SwitchParameter UseTls { get; set; }
protected override void ProcessRecord()
{
var clientOptionsBuilder = new MqttClientOptionsBuilder();
clientOptionsBuilder.WithTcpServer(Host, Port);
if (Username != null)
{
clientOptionsBuilder.WithCredentials(Username, Password);
}
var response = Session.GetClient().ConnectAsync(clientOptionsBuilder.Build()).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -0,0 +1,31 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Disconnect, "MqttSession")]
public class DisconnectMqttSessionCmdlet : PSCmdlet
{
[Parameter]
public MqttClientDisconnectOptionsReason Reason { get; set; } = MqttClientDisconnectOptionsReason.NormalDisconnection;
[Parameter]
public string? ReasonString { get; set; }
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
public uint SessionExpiryInterval { get; set; }
protected override void ProcessRecord()
{
if (Session.GetClient().IsConnected)
{
var options = new MqttClientDisconnectOptionsBuilder().WithSessionExpiryInterval(SessionExpiryInterval).WithReason(Reason).WithReasonString(ReasonString).Build();
Session.GetClient().DisconnectAsync(options).GetAwaiter().GetResult();
}
WriteObject("Disconnected.");
}
}
@@ -0,0 +1,13 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommon.New, "MqttSession")]
[OutputType(typeof(PsMqttSession))]
public class NewMqttSessionCmdlet : PSCmdlet
{
protected override void ProcessRecord()
{
WriteObject(new PsMqttSession());
}
}
@@ -0,0 +1,52 @@
using System.Management.Automation;
using System.Text;
using MQTTnet.Protocol;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsData.Publish, "MqttMessage")]
public class PublishMqttMessageCmdlet : PSCmdlet
{
[Parameter(Mandatory = true)]
public string? Payload { get; set; }
[Parameter]
public int QoS { get; set; } = 0;
[Parameter]
public SwitchParameter Retain { get; set; }
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter(Mandatory = true)]
public required string Topic { get; set; }
public string? ContentType { get; set; }
public string? ResponseTopic { get; set; }
public ushort TopicAlias { get; set; }
public uint MessageExpiryInterval { get; set; }
protected override void ProcessRecord()
{
// if (Session == null || !Session.IsConnected)
// throw new InvalidOperationException("Session not connected.");
var msg = new MqttApplicationMessageBuilder().WithTopic(Topic)
.WithPayload(Encoding.UTF8.GetBytes(Payload ?? string.Empty))
.WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
.WithRetainFlag(Retain)
.WithContentType(ContentType)
.WithResponseTopic(ResponseTopic)
.WithTopicAlias(TopicAlias)
.WithMessageExpiryInterval(MessageExpiryInterval)
.Build();
var response = Session.GetClient().PublishAsync(msg).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -0,0 +1,74 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Receive, "MqttMessage")]
[OutputType(typeof(PsMqttMessage))]
public class ReceiveMqttMessageCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
[ValidateRange(0, int.MaxValue)]
public int TimeoutSeconds { get; set; } = 0;
protected override void ProcessRecord()
{
var tcs = new TaskCompletionSource<PsMqttMessage>();
EventHandler<PsMqttMessage>? handler = null;
CancellationTokenSource? cts = null;
try
{
handler = (s, e) =>
{
if (!tcs.Task.IsCompleted)
{
tcs.TrySetResult(e);
}
};
Session.MessageReceived += handler;
// Optionaler Timeout
if (TimeoutSeconds > 0)
{
cts = new CancellationTokenSource();
cts.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
cts.Token.Register(() =>
{
if (!tcs.Task.IsCompleted)
{
tcs.TrySetCanceled();
}
});
}
WriteVerbose("Waiting for next MQTT message...");
try
{
var message = tcs.Task.GetAwaiter().GetResult();
WriteObject(message);
}
catch (TaskCanceledException)
{
WriteWarning($"Timeout after {TimeoutSeconds} seconds — no message received.");
}
}
finally
{
if (handler != null)
{
Session.MessageReceived -= handler;
}
cts?.Dispose();
}
}
}
@@ -0,0 +1,26 @@
using System.Management.Automation;
using System.Management.Automation.Runspaces;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsLifecycle.Register, "MqttMessageHandler")]
public class RegisterMqttMessageHandlerCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter(Mandatory = true)]
public ScriptBlock? Action { get; set; }
protected override void ProcessRecord()
{
EventHandler<PsMqttMessage> handler = (s, e) =>
{
//throw new NotImplementedException();
//InvokeCommand.InvokeScript(Action, false, PipelineResultTypes.Output, null, new object[] { e.Topic, e.Payload });
};
Session.MessageReceived += handler;
WriteObject($"Handler registered for MQTT messages on session.");
}
}
@@ -0,0 +1,16 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommon.Remove, "MqttSession")]
public class RemoveMqttSessionCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
protected override void ProcessRecord()
{
Session.Dispose();
WriteObject("Session removed.");
}
}
@@ -0,0 +1,55 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
using MQTTnet.Protocol;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet("Subscribe", "MqttTopic")]
[OutputType(typeof(MqttClientSubscribeResult))]
public class SubscribeMqttTopicCmdlet : PSCmdlet
{
[Parameter]
public bool NoLocal { get; set; }
[Parameter]
public int QoS { get; set; } = 0;
[Parameter]
public bool RetainAsPublished { get; set; }
[Parameter]
public MqttRetainHandling RetainHandling { get; set; } = MqttRetainHandling.SendAtSubscribe;
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
public uint SubscriptionIdentifier { get; set; }
[Parameter(Mandatory = true)]
public required string Topic { get; set; }
protected override void ProcessRecord()
{
var topicFilter = new MqttTopicFilterBuilder().WithTopic(Topic)
.WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
.WithNoLocal(NoLocal)
.WithRetainAsPublished(RetainAsPublished)
.WithRetainHandling(RetainHandling)
.Build();
var options = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter);
if (SubscriptionIdentifier > 0)
{
options.WithSubscriptionIdentifier(SubscriptionIdentifier);
}
var response = Session.GetClient().SubscribeAsync(options.Build()).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -0,0 +1,23 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet("Unsubscribe", "MqttTopic")]
[OutputType(typeof(MqttClientUnsubscribeResult))]
public class UnsubscribeMqttTopicCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter(Mandatory = true)]
public required string Topic { get; set; }
protected override void ProcessRecord()
{
var options = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(Topic).Build();
var response = Session.GetClient().UnsubscribeAsync(options).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -0,0 +1,28 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<AssemblyName>MQTTnet.PowerShell</AssemblyName>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="PowerShellStandard.Library" Version="5.1.1">
<PrivateAssets>All</PrivateAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
</ItemGroup>
<ItemGroup>
<Content Include="bin\Debug\net8.0\MQTTnet.dll" />
<Content Include="bin\Debug\net8.0\MQTTnet.pdb" />
<Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.deps.json" />
<Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.dll" />
<Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.pdb" />
<Content Include="bin\Debug\net8.0\MQTTnet.xml" />
</ItemGroup>
</Project>
@@ -0,0 +1,30 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
using MQTTnet.Packets;
namespace MQTTnet.PowerShell;
public class PsMqttPublishResult
{
}
public class PsMqttConnectResult
{
}
public class PsMqttMessage
{
public required string? ContentType { get; init; }
public required string Payload { get; init; }
public int QoS { get; init; }
public required byte[] RawPayload { get; init; }
public string? ResponseTopic { get; init; }
public bool Retain { get; init; }
public required string Topic { get; init; }
public required List<MqttUserProperty>? UserProperties { get; init; }
}
@@ -0,0 +1,50 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Buffers;
using System.Text;
namespace MQTTnet.PowerShell;
public sealed class PsMqttSession : IDisposable
{
readonly IMqttClient _client;
public PsMqttSession()
{
var factory = new MqttClientFactory();
_client = factory.CreateMqttClient();
_client.ApplicationMessageReceivedAsync += e =>
{
MessageReceived?.Invoke(
this,
new PsMqttMessage
{
Topic = e.ApplicationMessage.Topic,
Payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
RawPayload = e.ApplicationMessage.Payload.ToArray(),
UserProperties = e.ApplicationMessage.UserProperties,
QoS = (int)e.ApplicationMessage.QualityOfServiceLevel,
ContentType = e.ApplicationMessage.ContentType,
Retain = e.ApplicationMessage.Retain,
ResponseTopic = e.ApplicationMessage.ResponseTopic
});
return Task.CompletedTask;
};
}
public event EventHandler<PsMqttMessage>? MessageReceived;
public void Dispose()
{
_client.Dispose();
}
public IMqttClient GetClient()
{
return _client;
}
}
+15
View File
@@ -0,0 +1,15 @@
Import-Module ./bin/Debug/net8.0/MQTTnet.PowerShell.dll
$session = New-MqttSession
Connect-MqttSession -Session $session -Host "192.168.1.16"
Publish-MqttMessage -Session $session -Topic "Test" -Payload "Hello from PowerShell"
Subscribe-MqttTopic -Session $session -Topic "test/demo"
Receive-MqttMessage -Session $session -TimeoutSeconds 200
Disconnect-MqttSession -Session $session
Remove-MqttSession -Session $session
@@ -0,0 +1,55 @@
{
"runtimeTarget": {
"name": ".NETCoreApp,Version=v8.0",
"signature": ""
},
"compilationOptions": {},
"targets": {
".NETCoreApp,Version=v8.0": {
"MQTTnet.PowerShell/1.0.0": {
"dependencies": {
"MQTTnet": "1.0.0",
"PowerShellStandard.Library": "5.1.1"
},
"runtime": {
"MQTTnet.PowerShell.dll": {}
}
},
"PowerShellStandard.Library/5.1.1": {
"runtime": {
"lib/netstandard2.0/System.Management.Automation.dll": {
"assemblyVersion": "3.0.0.0",
"fileVersion": "5.1.1.0"
}
}
},
"MQTTnet/1.0.0": {
"runtime": {
"MQTTnet.dll": {
"assemblyVersion": "1.0.0.0",
"fileVersion": "1.0.0.0"
}
}
}
}
},
"libraries": {
"MQTTnet.PowerShell/1.0.0": {
"type": "project",
"serviceable": false,
"sha512": ""
},
"PowerShellStandard.Library/5.1.1": {
"type": "package",
"serviceable": true,
"sha512": "sha512-e31xJjG+Kjbv6YF3Yq6D4Dl3or8v7LrNF41k3CXrWozW6hR1zcOe5KYuZJaGSiAgLnwP8wcW+I3+IWEzMPZKXQ==",
"path": "powershellstandard.library/5.1.1",
"hashPath": "powershellstandard.library.5.1.1.nupkg.sha512"
},
"MQTTnet/1.0.0": {
"type": "project",
"serviceable": false,
"sha512": ""
}
}
}
File diff suppressed because it is too large Load Diff
@@ -53,10 +53,7 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value) public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value)
{ {
if (name == null) ArgumentNullException.ThrowIfNull(name);
{
throw new ArgumentNullException(nameof(name));
}
if (_options.UserProperties == null) if (_options.UserProperties == null)
{ {
@@ -10,7 +10,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingPacketEventArgs : EventArgs 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; CancellationToken = cancellationToken;
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
@@ -8,7 +8,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingPublishEventArgs : EventArgs 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)); ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
CancellationToken = cancellationToken; CancellationToken = cancellationToken;
@@ -30,11 +30,6 @@ public sealed class InterceptingPublishEventArgs : EventArgs
/// </summary> /// </summary>
public string ClientId { get; } public string ClientId { get; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
public bool CloseConnection { get; set; } public bool CloseConnection { get; set; }
/// <summary> /// <summary>
@@ -45,10 +40,15 @@ public sealed class InterceptingPublishEventArgs : EventArgs
/// <summary> /// <summary>
/// Gets the response which will be sent to the client via the PUBACK etc. packets. /// Gets the response which will be sent to the client via the PUBACK etc. packets.
/// </summary> /// </summary>
public PublishResponse Response { get; } = new PublishResponse(); public PublishResponse Response { get; } = new();
/// <summary> /// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary> /// </summary>
public IDictionary SessionItems { get; } 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 sealed class InterceptingSubscriptionEventArgs : EventArgs
{ {
public InterceptingSubscriptionEventArgs( public InterceptingSubscriptionEventArgs(
CancellationToken cancellationToken,
string clientId, string clientId,
string userName, string userName,
MqttSessionStatus session, MqttSessionStatus session,
MqttTopicFilter topicFilter, MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties) List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{ {
CancellationToken = cancellationToken; CancellationToken = cancellationToken;
ClientId = clientId; ClientId = clientId;
@@ -36,11 +36,6 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// </summary> /// </summary>
public string ClientId { get; } public string ClientId { get; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
/// <summary> /// <summary>
/// Gets or sets whether the broker should close the client connection. /// Gets or sets whether the broker should close the client connection.
/// </summary> /// </summary>
@@ -61,7 +56,7 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// <summary> /// <summary>
/// Gets the response which will be sent to the client via the SUBACK packet. /// Gets the response which will be sent to the client via the SUBACK packet.
/// </summary> /// </summary>
public SubscribeResponse Response { get; } = new SubscribeResponse(); public SubscribeResponse Response { get; } = new();
/// <summary> /// <summary>
/// Gets the current client session. /// Gets the current client session.
@@ -79,6 +74,11 @@ public sealed class InterceptingSubscriptionEventArgs : EventArgs
/// </summary> /// </summary>
public MqttTopicFilter TopicFilter { get; set; } public MqttTopicFilter TopicFilter { get; set; }
/// <summary>
/// Gets the user name of the client.
/// </summary>
public string UserName { get; }
/// <summary> /// <summary>
/// Gets or sets the user properties. /// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks> /// <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -9,7 +9,7 @@ namespace MQTTnet.Server;
public sealed class InterceptingUnsubscriptionEventArgs : EventArgs 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; CancellationToken = cancellationToken;
ClientId = clientId; ClientId = clientId;
@@ -211,7 +211,7 @@ public sealed class MqttTcpServerListener : IDisposable
var clientHandler = ClientHandler; var clientHandler = ClientHandler;
if (clientHandler != null) 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 bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax);
var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter); var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter);
@@ -131,7 +131,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
// Allow the user to intercept application message... // Allow the user to intercept application message...
if (_eventContainer.InterceptingPublishEvent.HasHandlers) 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)) if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
{ {
// This can happen if a topic alias us used but the topic is // 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(); _sessionsManagementLock.EnterWriteLock();
try try
@@ -448,7 +448,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
_subscriberSessions.Add(clientSession); _subscriberSessions.Add(clientSession);
} }
foreach (var topic in topics) foreach (var topic in subscriptionsTopics)
{ {
clientSession.AddSubscribedTopic(topic); clientSession.AddSubscribedTopic(topic);
} }
@@ -81,7 +81,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
return CheckSubscriptionsResult.NotSubscribed; return CheckSubscriptionsResult.NotSubscribed;
} }
var senderIsReceiver = string.Equals(senderId, _session.Id); var senderIsReceiver = string.Equals(senderId, _session.Id, StringComparison.Ordinal);
var maxQoSLevel = -1; // Not subscribed. var maxQoSLevel = -1; // Not subscribed.
HashSet<uint> subscriptionIdentifiers = null; HashSet<uint> subscriptionIdentifiers = null;
@@ -460,7 +460,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
List<MqttUserProperty> userProperties, List<MqttUserProperty> userProperties,
CancellationToken cancellationToken) 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) if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{ {
@@ -475,7 +475,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2; eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2;
} }
if (topicFilter.Topic.StartsWith("$share/")) if (topicFilter.Topic.StartsWith("$share/", StringComparison.InvariantCulture))
{ {
eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported; eventArgs.Response.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported;
} }
@@ -493,7 +493,7 @@ public sealed class MqttClientSubscriptionsManager : IDisposable
List<MqttUserProperty> userProperties, List<MqttUserProperty> userProperties,
CancellationToken cancellationToken) 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 = Response =
{ {
@@ -156,7 +156,7 @@ public sealed class MqttConnectedClient : IDisposable
// TODO: Maybe adding a configuration option is requested in the future. // TODO: Maybe adding a configuration option is requested in the future.
if (disconnectOptions != null) 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)) !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
{ {
// It is very important to send the DISCONNECT packet here BEFORE cancelling the // It is very important to send the DISCONNECT packet here BEFORE cancelling the
@@ -349,7 +349,7 @@ public sealed class MqttConnectedClient : IDisposable
return packet; 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); await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
if (!interceptingPacketEventArgs.ProcessPacket || packet == null) if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
@@ -395,7 +395,7 @@ public sealed class MqttConnectedClient : IDisposable
if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers) 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); await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
currentPacket = interceptingPacketEventArgs.Packet; currentPacket = interceptingPacketEventArgs.Packet;
processPacket = interceptingPacketEventArgs.ProcessPacket; processPacket = interceptingPacketEventArgs.ProcessPacket;
@@ -7,13 +7,12 @@ using MQTTnet.Internal;
namespace MQTTnet.Server.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 MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096);
readonly AsyncLock _storageAccessLock = new();
public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger) public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger)
{ {
@@ -24,6 +23,46 @@ public sealed class MqttRetainedMessagesManager
_logger = logger.WithSource(nameof(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() public async Task Start()
{ {
try try
@@ -79,8 +118,7 @@ public sealed class MqttRetainedMessagesManager
} }
else else
{ {
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
!MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
{ {
_messages[applicationMessage.Topic] = applicationMessage; _messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true; saveIsRequired = true;
@@ -113,39 +151,4 @@ public sealed class MqttRetainedMessagesManager
_logger.Error(exception, "Unhandled exception while handling retained messages."); _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"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<AssemblyName>MQTTnet.Server</AssemblyName> <AssemblyName>MQTTnet.Server</AssemblyName>
<RootNamespace>MQTTnet.Server</RootNamespace> <RootNamespace>MQTTnet.Server</RootNamespace>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<GenerateDocumentationFile>true</GenerateDocumentationFile> <GenerateDocumentationFile>true</GenerateDocumentationFile>
<Description>This is the server implementation of MQTTnet.</Description> <Description>This is the server implementation of MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Server</PackageId> <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> <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> </PropertyGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/> <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
</ItemGroup> </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> </Project>
@@ -2,12 +2,14 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System.Diagnostics.CodeAnalysis;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server.EnhancedAuthentication; using MQTTnet.Server.EnhancedAuthentication;
using MQTTnet.Server.Internal.Adapter; using MQTTnet.Server.Internal.Adapter;
namespace MQTTnet.Server; namespace MQTTnet.Server;
[SuppressMessage("Performance", "CA1822:Mark members as static")]
public sealed class MqttServerFactory public sealed class MqttServerFactory
{ {
public MqttServerFactory() : this(new MqttNetNullLogger()) public MqttServerFactory() : this(new MqttNetNullLogger())
@@ -5,8 +5,8 @@
using System.Net; using System.Net;
using System.Net.Security; using System.Net.Security;
using System.Security.Authentication; using System.Security.Authentication;
using MQTTnet.Certificates;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using MQTTnet.Certificates;
// ReSharper disable UnusedMember.Global // ReSharper disable UnusedMember.Global
namespace MQTTnet.Server; namespace MQTTnet.Server;
@@ -47,12 +47,22 @@ public class MqttServerOptionsBuilder
return this; 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) public MqttServerOptionsBuilder WithDefaultEndpointBoundIPAddress(IPAddress value)
{ {
_options.DefaultEndpointOptions.BoundInterNetworkAddress = value ?? IPAddress.Any; _options.DefaultEndpointOptions.BoundInterNetworkAddress = value ?? IPAddress.Any;
return this; 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) public MqttServerOptionsBuilder WithDefaultEndpointBoundIPV6Address(IPAddress value)
{ {
_options.DefaultEndpointOptions.BoundInterNetworkV6Address = value ?? IPAddress.Any; _options.DefaultEndpointOptions.BoundInterNetworkV6Address = value ?? IPAddress.Any;
@@ -77,12 +87,22 @@ public class MqttServerOptionsBuilder
return this; 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) public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPAddress(IPAddress value)
{ {
_options.TlsEndpointOptions.BoundInterNetworkAddress = value; _options.TlsEndpointOptions.BoundInterNetworkAddress = value;
return this; 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) public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPV6Address(IPAddress value)
{ {
_options.TlsEndpointOptions.BoundInterNetworkV6Address = value; _options.TlsEndpointOptions.BoundInterNetworkV6Address = value;
@@ -95,6 +115,35 @@ public class MqttServerOptionsBuilder
return this; 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) public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value)
{ {
_options.TlsEndpointOptions.SslProtocol = value; _options.TlsEndpointOptions.SslProtocol = value;
@@ -114,12 +163,6 @@ public class MqttServerOptionsBuilder
return this; return this;
} }
public MqttServerOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value)
{
_options.PendingMessagesOverflowStrategy = value;
return this;
}
public MqttServerOptionsBuilder WithoutDefaultEndpoint() public MqttServerOptionsBuilder WithoutDefaultEndpoint()
{ {
_options.DefaultEndpointOptions.IsEnabled = false; _options.DefaultEndpointOptions.IsEnabled = false;
@@ -144,6 +187,12 @@ public class MqttServerOptionsBuilder
return this; return this;
} }
public MqttServerOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value)
{
_options.PendingMessagesOverflowStrategy = value;
return this;
}
public MqttServerOptionsBuilder WithPersistentSessions(bool value = true) public MqttServerOptionsBuilder WithPersistentSessions(bool value = true)
{ {
_options.EnablePersistentSessions = value; _options.EnablePersistentSessions = value;
@@ -182,33 +231,4 @@ public class MqttServerOptionsBuilder
_options.TlsEndpointOptions.ReuseAddress = true; _options.TlsEndpointOptions.ReuseAddress = true;
return this; 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 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 int ConnectionBacklog { get; set; } = 100;
public bool IsEnabled { get; set; }
public bool NoDelay { get; set; } = true;
/// <summary> /// <summary>
/// Gets or sets whether the sockets keep alive feature should be used. /// Gets or sets whether the sockets keep alive feature should be used.
@@ -23,13 +40,16 @@ public abstract class MqttServerTcpEndpointBaseOptions
/// </summary> /// </summary>
public bool? KeepAlive { get; set; } 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> /// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets /// This requires admin permissions on Linux.
/// 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> /// </summary>
public bool AllowPacketFragmentation { get; set; } = true; public bool ReuseAddress { get; set; }
/// <summary> /// <summary>
/// Gets or sets the TCP keep alive interval. /// 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. /// The value _null_ indicates that the OS and framework defaults should be used.
/// </summary> /// </summary>
public int? TcpKeepAliveTime { get; set; } 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 -1
View File
@@ -11,7 +11,7 @@ namespace MQTTnet.TestApp;
public sealed class AsyncLockTest public sealed class AsyncLockTest
{ {
public async Task Run() public static async Task Run()
{ {
try try
{ {
@@ -3,17 +3,7 @@
<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<DebugType>Full</DebugType> <DebugType>Full</DebugType>
<TargetFramework>net8.0</TargetFramework>
<IsPackable>false</IsPackable> <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> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -14,7 +14,7 @@ namespace MQTTnet.TestApp;
/// the number of messages per second that can be exchanged between publishers and subscriber. /// 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. /// Measurements are performed for subscriptions containing no wildcard, a single wildcard or multiple wildcards.
/// </summary> /// </summary>
public class MessageThroughputTest public sealed class MessageThroughputTest : IDisposable
{ {
// Change these constants to suit // Change these constants to suit
const int NumPublishers = 5000; const int NumPublishers = 5000;
@@ -55,9 +55,9 @@ public class MessageThroughputTest
await Setup(); await Setup();
await Subscribe_to_No_Wildcard_Topics(); await SubscribeToNoWildcardTopics();
await Subscribe_to_Single_Wildcard_Topics(); await SubscribeToSingleWildcardTopics();
await Subscribe_to_Multi_Wildcard_Topics(); await SubscribeToMultiWildcardTopics();
Console.WriteLine(); Console.WriteLine();
Console.WriteLine("End message throughput test"); Console.WriteLine("End message throughput test");
@@ -74,7 +74,7 @@ public class MessageThroughputTest
public async Task Setup() 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 serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
var mqttClientFactory = new MqttClientFactory(); var mqttClientFactory = new MqttClientFactory();
@@ -152,7 +152,7 @@ public class MessageThroughputTest
/// <summary> /// <summary>
/// Measure no-wildcard topic subscription message exchange performance /// Measure no-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task Subscribe_to_No_Wildcard_Topics() public Task SubscribeToNoWildcardTopics()
{ {
return ProcessMessages(_topicsByPublisher, "no wildcards"); return ProcessMessages(_topicsByPublisher, "no wildcards");
} }
@@ -160,7 +160,7 @@ public class MessageThroughputTest
/// <summary> /// <summary>
/// Measure single-wildcard topic subscription message exchange performance /// Measure single-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task Subscribe_to_Single_Wildcard_Topics() public Task SubscribeToSingleWildcardTopics()
{ {
return ProcessMessages(_singleWildcardTopicsByPublisher, "single wildcard"); return ProcessMessages(_singleWildcardTopicsByPublisher, "single wildcard");
} }
@@ -168,7 +168,7 @@ public class MessageThroughputTest
/// <summary> /// <summary>
/// Measure multi-wildcard topic subscription message exchange performance /// Measure multi-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task Subscribe_to_Multi_Wildcard_Topics() public Task SubscribeToMultiWildcardTopics()
{ {
return ProcessMessages(_multiWildcardTopicsByPublisher, "multi wildcard"); return ProcessMessages(_multiWildcardTopicsByPublisher, "multi wildcard");
} }
@@ -342,4 +342,8 @@ public class MessageThroughputTest
Console.ResetColor(); Console.ResetColor();
} }
public void Dispose()
{
_cancellationTokenSource.Dispose();
}
} }
@@ -34,7 +34,9 @@ public static class MqttNetConsoleLogger
static void PrintToConsole(object sender, MqttNetLogMessagePublishedEventArgs e) static void PrintToConsole(object sender, MqttNetLogMessagePublishedEventArgs e)
{ {
var output = new StringBuilder(); 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}"); 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) if (e.LogMessage.Exception != null)
{ {
output.AppendLine(e.LogMessage.Exception.ToString()); output.AppendLine(e.LogMessage.Exception.ToString());
+1 -1
View File
@@ -204,7 +204,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); Interlocked.Increment(ref count);
return Task.Run(() => client.PublishAsync(applicationMessage)); return Task.Run(() => client.PublishAsync(applicationMessage));
+1 -1
View File
@@ -85,7 +85,7 @@ public static class Program
} }
else if (pressedKey.KeyChar == 'f') else if (pressedKey.KeyChar == 'f')
{ {
Task.Run(new AsyncLockTest().Run); Task.Run(AsyncLockTest.Run);
} }
Thread.Sleep(Timeout.Infinite); Thread.Sleep(Timeout.Infinite);
+1 -1
View File
@@ -157,7 +157,7 @@ public static class PublicBrokerTest
if (receivedMessage?.Topic != topic || receivedMessage?.ConvertPayloadToString() != "Hello_World") if (receivedMessage?.Topic != topic || receivedMessage?.ConvertPayloadToString() != "Hello_World")
{ {
throw new Exception("Message invalid."); throw new InvalidOperationException("Message invalid.");
} }
await client.UnsubscribeAsync(topic).ConfigureAwait(false); await client.UnsubscribeAsync(topic).ConfigureAwait(false);
+2 -2
View File
@@ -129,12 +129,12 @@ public static class ServerTest
mqttServer.InterceptingSubscriptionAsync += e => mqttServer.InterceptingSubscriptionAsync += e =>
{ {
if (e.TopicFilter.Topic.StartsWith("admin/foo/bar") && e.ClientId != "theAdmin") if (e.TopicFilter.Topic.StartsWith("admin/foo/bar", StringComparison.InvariantCulture) && e.ClientId != "theAdmin")
{ {
e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
} }
if (e.TopicFilter.Topic.StartsWith("the/secret/stuff") && e.ClientId != "Imperator") if (e.TopicFilter.Topic.StartsWith("the/secret/stuff", StringComparison.InvariantCulture) && e.ClientId != "Imperator")
{ {
e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
e.CloseConnection = true; e.CloseConnection = true;
+4 -4
View File
@@ -6,7 +6,7 @@ namespace MQTTnet.TestApp;
public class TopicGenerator public class TopicGenerator
{ {
public void Generate( public static void Generate(
int numPublishers, int numTopicsPerPublisher, int numPublishers, int numTopicsPerPublisher,
out Dictionary<string, List<string>> topicsByPublisher, out Dictionary<string, List<string>> topicsByPublisher,
out Dictionary<string, List<string>> singleWildcardTopicsByPublisher, out Dictionary<string, List<string>> singleWildcardTopicsByPublisher,
@@ -84,14 +84,14 @@ public class TopicGenerator
} }
} }
void AddPublisherTopic(string publisherName, string topic, Dictionary<string, List<string>> topicsByPublisher) static void AddPublisherTopic(string publisherName, string topic, Dictionary<string, List<string>> topicsByPublisher)
{ {
List<string> topicList; if (!topicsByPublisher.TryGetValue(publisherName, out var topicList))
if (!topicsByPublisher.TryGetValue(publisherName, out topicList))
{ {
topicList = new List<string>(); topicList = new List<string>();
topicsByPublisher.Add(publisherName, topicList); topicsByPublisher.Add(publisherName, topicList);
} }
topicList.Add(topic); topicList.Add(topic);
} }
} }
@@ -119,9 +119,11 @@ public class MqttConnectionContextTest
await Assert.ThrowsExactlyAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None)).ConfigureAwait(false); await Assert.ThrowsExactlyAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None)).ConfigureAwait(false);
} }
class Startup sealed class Startup
{ {
#pragma warning disable CA1822
public void Configure(IApplicationBuilder app) public void Configure(IApplicationBuilder app)
#pragma warning restore CA1822
{ {
} }
} }
+2 -2
View File
@@ -20,12 +20,12 @@ public abstract class BaseTestClass
return new TestEnvironment(TestContext, protocolVersion, trackUnobservedTaskException); return new TestEnvironment(TestContext, protocolVersion, trackUnobservedTaskException);
} }
protected Task LongTestDelay() public static Task LongTestDelay()
{ {
return Task.Delay(TimeSpan.FromSeconds(1)); return Task.Delay(TimeSpan.FromSeconds(1));
} }
protected Task MediumTestDelay() public static Task MediumTestDelay()
{ {
return Task.Delay(TimeSpan.FromSeconds(0.5)); return Task.Delay(TimeSpan.FromSeconds(0.5));
} }
@@ -188,7 +188,7 @@ public sealed class LowLevelMqttClient_Tests : BaseTestClass
return await client.ReceiveAsync(CancellationToken.None) as MqttConnAckPacket; return await client.ReceiveAsync(CancellationToken.None) as MqttConnAckPacket;
} }
async Task<MqttSubAckPacket> Subscribe(ILowLevelMqttClient client, string topic) static async Task<MqttSubAckPacket> Subscribe(ILowLevelMqttClient client, string topic)
{ {
await client.SendAsync( await client.SendAsync(
new MqttSubscribePacket new MqttSubscribePacket
@@ -73,7 +73,7 @@ public sealed class MqttClient_Connection_Tests : BaseTestClass
catch (Exception exception) catch (Exception exception)
{ {
Assert.IsNotNull(exception); Assert.IsNotNull(exception);
Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException)); Assert.IsInstanceOfType<MqttCommunicationException>(exception);
} }
await LongTestDelay(); // disconnected handler is called async await LongTestDelay(); // disconnected handler is called async
@@ -161,7 +161,7 @@ public sealed class MqttClient_Connection_Tests : BaseTestClass
Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value); Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value);
} }
class TestClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler sealed class TestClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler
{ {
public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs) public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs)
{ {
@@ -6,6 +6,7 @@ using System;
using System.Buffers; using System.Buffers;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.Globalization;
using System.Linq; using System.Linq;
using System.Net.Sockets; using System.Net.Sockets;
using System.Text; using System.Text;
@@ -122,8 +123,8 @@ public sealed class MqttClient_Tests : BaseTestClass
await Task.Delay(500); await Task.Delay(500);
Assert.IsNotNull(ex); Assert.IsNotNull(ex);
Assert.IsInstanceOfType(ex, typeof(MqttCommunicationException)); Assert.IsInstanceOfType<MqttCommunicationException>(ex);
Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException)); Assert.IsInstanceOfType<SocketException>(ex.InnerException);
} }
[TestMethod] [TestMethod]
@@ -245,8 +246,8 @@ public sealed class MqttClient_Tests : BaseTestClass
catch (Exception exception) catch (Exception exception)
{ {
Assert.IsNotNull(exception); Assert.IsNotNull(exception);
Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException)); Assert.IsInstanceOfType<MqttCommunicationException>(exception);
Assert.IsInstanceOfType(exception.InnerException, typeof(SocketException)); Assert.IsInstanceOfType<SocketException>(exception.InnerException);
} }
} }
@@ -355,7 +356,7 @@ public sealed class MqttClient_Tests : BaseTestClass
async Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs) async Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
{ {
var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString()); var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString(), CultureInfo.InvariantCulture);
await Task.Delay(value); await Task.Delay(value);
lock (receivedValues) lock (receivedValues)
@@ -369,7 +370,7 @@ public sealed class MqttClient_Tests : BaseTestClass
var client2 = await testEnvironment.ConnectClient(); var client2 = await testEnvironment.ConnectClient();
for (var i = MessagesCount; i > 0; i--) for (var i = MessagesCount; i > 0; i--)
{ {
await client2.PublishStringAsync("x", i.ToString()); await client2.PublishStringAsync("x", i.ToString(CultureInfo.InvariantCulture));
} }
await Task.Delay(5000); await Task.Delay(5000);
@@ -401,7 +402,7 @@ public sealed class MqttClient_Tests : BaseTestClass
var client2 = await testEnvironment.ConnectClient(); var client2 = await testEnvironment.ConnectClient();
for (var i = MessagesCount; i > 0; i--) for (var i = MessagesCount; i > 0; i--)
{ {
await client2.PublishStringAsync("x", i.ToString(), MqttQualityOfServiceLevel.ExactlyOnce); await client2.PublishStringAsync("x", i.ToString(CultureInfo.InvariantCulture), MqttQualityOfServiceLevel.ExactlyOnce);
} }
await Task.Delay(5000); await Task.Delay(5000);
@@ -415,7 +416,7 @@ public sealed class MqttClient_Tests : BaseTestClass
Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs) Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
{ {
var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString()); var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString(), CultureInfo.InvariantCulture);
eventArgs.AutoAcknowledge = false; eventArgs.AutoAcknowledge = false;
Task.Delay(value).ContinueWith(_ => eventArgs.AcknowledgeAsync(CancellationToken.None)); Task.Delay(value).ContinueWith(_ => eventArgs.AcknowledgeAsync(CancellationToken.None));
@@ -0,0 +1,24 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Exceptions;
namespace MQTTnet.Tests.Exceptions;
[TestClass]
public class MqttProtocolViolationException_Tests
{
[TestMethod]
public void No_For_Max_Large_Variable_Byte_Integer()
{
MqttProtocolViolationException.ThrowIfVariableByteIntegerExceedsLimit(MqttProtocolViolationException.VariableByteIntegerMaxValue);
}
[TestMethod]
public void Throw_Exception_For_Too_Large_Variable_Byte_Integer()
{
Assert.ThrowsExactly<MqttProtocolViolationException>(() => MqttProtocolViolationException.ThrowIfVariableByteIntegerExceedsLimit(1726790899));
}
}
+2 -2
View File
@@ -220,7 +220,7 @@ public sealed class Rpc_Tests : BaseTestClass
Assert.AreEqual("pong", Encoding.UTF8.GetString(response)); Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
} }
class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy sealed class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
{ {
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context) public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
{ {
@@ -232,7 +232,7 @@ public sealed class Rpc_Tests : BaseTestClass
} }
} }
class TestParametersTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy sealed class TestParametersTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
{ {
internal const string ExpectedParamName = "test_param_name"; internal const string ExpectedParamName = "test_param_name";
@@ -287,7 +287,7 @@ public sealed class MqttBufferReader_Tests
// Helper class to build up a reference to elements of various types in a buffer // Helper class to build up a reference to elements of various types in a buffer
class ElementReference sealed class ElementReference
{ {
public enum BufferElementType public enum BufferElementType
{ {
@@ -551,7 +551,7 @@ public sealed class MqttPacketSerialization_V3_Binary_Tests
} }
void DeserializeAndCompare(MqttPacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) static void DeserializeAndCompare(MqttPacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
{ {
var writer = WriterFactory(); var writer = WriterFactory();
@@ -35,15 +35,15 @@ public sealed class MqttPacketBus_Tests
Assert.AreEqual(9, bus.TotalItemsCount); Assert.AreEqual(9, bus.TotalItemsCount);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet);
Assert.AreEqual(0, bus.TotalItemsCount); Assert.AreEqual(0, bus.TotalItemsCount);
} }
+3 -11
View File
@@ -3,17 +3,10 @@
<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<IsPackable>false</IsPackable> <IsPackable>false</IsPackable>
<EnableNETAnalyzers>false</EnableNETAnalyzers> <LangVersion>default</LangVersion>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild> <Nullable>disable</Nullable>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers> <NoWarn>$(NoWarn);CA1707</NoWarn>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -21,7 +14,6 @@
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.0.0" /> <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.0.0" />
<PackageReference Include="MSTest" Version="4.0.1" /> <PackageReference Include="MSTest" Version="4.0.1" />
<FrameworkReference Include="Microsoft.AspNetCore.App" /> <FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup> </ItemGroup>
@@ -29,7 +29,9 @@ public sealed class MemoryMqttChannel : IMqttChannel
public EndPoint RemoteEndPoint { get; set; } public EndPoint RemoteEndPoint { get; set; }
public bool IsSecureConnection { get; } = false; public EndPoint LocalEndPoint { get; set; }
public bool IsSecureConnection { get; }
public X509Certificate2 ClientCertificate { get; set; } public X509Certificate2 ClientCertificate { get; set; }
@@ -58,5 +60,6 @@ public sealed class MemoryMqttChannel : IMqttChannel
public void Dispose() public void Dispose()
{ {
_stream.Dispose();
} }
} }
@@ -5,7 +5,6 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.Linq;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting; using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -182,7 +181,9 @@ public sealed class TestEnvironment : IDisposable
return new MqttRpcClient(await ConnectClient(), options); return new MqttRpcClient(await ConnectClient(), options);
} }
#pragma warning disable CA1822
public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IMqttClient mqttClient) public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IMqttClient mqttClient)
#pragma warning restore CA1822
{ {
return new TestApplicationMessageReceivedHandler(mqttClient); return new TestApplicationMessageReceivedHandler(mqttClient);
} }
@@ -199,7 +200,7 @@ public sealed class TestEnvironment : IDisposable
{ {
var clientOptions = e.ClientOptions; var clientOptions = e.ClientOptions;
var existingClientId = clientOptions.ClientId; var existingClientId = clientOptions.ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName, StringComparison.InvariantCulture))
{ {
clientOptions.ClientId = TestContext.TestName + "_" + existingClientId; clientOptions.ClientId = TestContext.TestName + "_" + existingClientId;
} }
@@ -257,7 +258,7 @@ public sealed class TestEnvironment : IDisposable
if (TestContext != null) if (TestContext != null)
{ {
// Null is used when the client id is assigned from the server! // Null is used when the client id is assigned from the server!
if (!string.IsNullOrEmpty(e.ClientId) && !e.ClientId.StartsWith(TestContext.TestName)) if (!string.IsNullOrEmpty(e.ClientId) && !e.ClientId.StartsWith(TestContext.TestName, StringComparison.InvariantCulture))
{ {
TrackException(new InvalidOperationException($"Invalid client ID used ({e.ClientId}). It must start with UnitTest name.")); TrackException(new InvalidOperationException($"Invalid client ID used ({e.ClientId}). It must start with UnitTest name."));
e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
@@ -326,10 +327,10 @@ public sealed class TestEnvironment : IDisposable
GC.WaitForFullGCComplete(); GC.WaitForFullGCComplete();
GC.WaitForPendingFinalizers(); GC.WaitForPendingFinalizers();
if (_exceptions.Any()) if (_exceptions.Count > 0)
{ {
// ReSharper disable once ThrowExceptionInUnexpectedLocation // ReSharper disable once ThrowExceptionInUnexpectedLocation
throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions)); throw new InvalidOperationException($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
} }
} }
finally finally
@@ -387,7 +388,7 @@ public sealed class TestEnvironment : IDisposable
{ {
var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)})."; var message = $"Server had {_serverErrors.Count} errors (${string.Join(Environment.NewLine, _serverErrors)}).";
Console.WriteLine(message); Console.WriteLine(message);
throw new Exception(message); throw new InvalidOperationException(message);
} }
} }
} }
@@ -400,7 +401,7 @@ public sealed class TestEnvironment : IDisposable
{ {
var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})"; var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})";
Console.WriteLine(message); Console.WriteLine(message);
throw new Exception(message); throw new InvalidOperationException(message);
} }
} }
} }
+4 -3
View File
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System; using System;
using System.Globalization;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
namespace MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Mockups;
@@ -13,12 +14,12 @@ public sealed class TestLogger : IMqttNetLogger
public bool IsEnabled { get; } = true; public bool IsEnabled { get; } = true;
public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) public void Publish(MqttNetLogLevel level, string source, string message, object[] parameters, Exception exception)
{ {
LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(new MqttNetLogMessage LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(new MqttNetLogMessage
{ {
Level = logLevel, Level = level,
Message = string.Format(message, parameters), Message = string.Format(CultureInfo.InvariantCulture, message, parameters),
Exception = exception Exception = exception
})); }));
} }
+2 -2
View File
@@ -45,7 +45,7 @@ public class MqttTcpChannel_Tests
using var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp); using var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
await clientSocket.ConnectAsync(remoteEndPoint, CancellationToken.None); await clientSocket.ConnectAsync(remoteEndPoint, CancellationToken.None);
var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), remoteEndPoint, null); var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), new DnsEndPoint("localhost", 50000), remoteEndPoint, null);
await Task.Delay(100, ct.Token); await Task.Delay(100, ct.Token);
@@ -71,7 +71,7 @@ public class MqttTcpChannel_Tests
} }
catch (Exception exception) catch (Exception exception)
{ {
Assert.IsInstanceOfType(exception, typeof(SocketException)); Assert.IsInstanceOfType<SocketException>(exception);
Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode); Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
} }
} }
+2 -1
View File
@@ -5,6 +5,7 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.Globalization;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting; using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Internal; using MQTTnet.Internal;
@@ -45,7 +46,7 @@ public class RoundtripTime_Tests
for (var i = 0; i < 100; i++) for (var i = 0; i < 100; i++)
{ {
response = new TaskCompletionSource<string>(); response = new TaskCompletionSource<string>();
await senderClient.PublishStringAsync("test", DateTime.UtcNow.Ticks.ToString()); await senderClient.PublishStringAsync("test", DateTime.UtcNow.Ticks.ToString(CultureInfo.InvariantCulture));
if (!response.Task.Wait(TimeSpan.FromSeconds(5))) if (!response.Task.Wait(TimeSpan.FromSeconds(5)))
{ {
throw new TimeoutException(); throw new TimeoutException();
+73 -75
View File
@@ -2,11 +2,6 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System; using System;
using System.Buffers; using System.Buffers;
using System.Collections.Generic; using System.Collections.Generic;
@@ -14,12 +9,18 @@ using System.Linq;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server; namespace MQTTnet.Tests.Server;
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
[TestClass] [TestClass]
public sealed class General_Tests : BaseTestClass public class General_Tests : BaseTestClass
{ {
Dictionary<string, bool> _connected; Dictionary<string, bool> _connected;
@@ -83,23 +84,53 @@ public sealed class General_Tests : BaseTestClass
} }
[TestMethod] [TestMethod]
public async Task Deny_Connection() public async Task Disconnect_Client_with_Reason()
{ {
using var testEnvironment = CreateTestEnvironment(); using var testEnvironment = CreateTestEnvironment();
testEnvironment.IgnoreClientLogErrors = true; var disconnectPacketReceived = false;
var server = await testEnvironment.StartServer(); string testClientId = null;
server.ValidatingConnectionAsync += e => await testEnvironment.StartServer();
testEnvironment.Server.ClientConnectedAsync += e =>
{ {
e.ReasonCode = MqttConnectReasonCode.NotAuthorized; testClientId = e.ClientId;
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
var client = testEnvironment.CreateClient(); var client = testEnvironment.CreateClient();
var response = await client.ConnectAsync(testEnvironment.CreateDefaultClientOptions());
Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, response.ResultCode); client.InspectPacketAsync += e =>
{
if (e.Buffer.Length > 0)
{
if (e.Buffer[0] == (byte)MqttControlPacketType.Disconnect << 4)
{
disconnectPacketReceived = true;
}
}
return CompletedTask.Instance;
};
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await LongTestDelay();
// Test client should be connected now
Assert.IsNotNull(testClientId);
// Have the server disconnect the client with SessionTakenOver reason; no DISCONNECT should be sent to MQTT 3.1.1 clients
await testEnvironment.Server.DisconnectClientAsync(testClientId, MqttDisconnectReasonCode.SessionTakenOver);
await LongTestDelay();
// MQTT 3.1.1 client should not receive a disconnect packet
Assert.IsFalse(disconnectPacketReceived, "Disconnect packet received when none is expected");
} }
[TestMethod] [TestMethod]
@@ -152,6 +183,27 @@ public sealed class General_Tests : BaseTestClass
Assert.AreEqual("y", buffer.ToString()); Assert.AreEqual("y", buffer.ToString());
} }
[TestMethod]
public async Task Expose_Local_And_Remote_Endpoint()
{
using var testEnvironment = CreateTestEnvironment();
var server = await testEnvironment.StartServer();
ValidatingConnectionEventArgs eventArgs = null;
server.ValidatingConnectionAsync += e =>
{
eventArgs = e;
return Task.CompletedTask;
};
await testEnvironment.ConnectClient();
await Task.Delay(500);
Assert.AreEqual("0.0.0.0:" + testEnvironment.ServerPort, eventArgs.ChannelAdapter.LocalEndPoint.ToString());
Assert.StartsWith("127.0.0.1:", eventArgs.ChannelAdapter.RemoteEndPoint.ToString());
}
[TestMethod] [TestMethod]
public async Task Handle_Clean_Disconnect() public async Task Handle_Clean_Disconnect()
{ {
@@ -205,8 +257,7 @@ public sealed class General_Tests : BaseTestClass
var testEnvironment2 = testEnvironment; var testEnvironment2 = testEnvironment;
tasks.Add( tasks.Add(
Task.Run( Task.Run(async () =>
async () =>
{ {
try try
{ {
@@ -299,7 +350,7 @@ public sealed class General_Tests : BaseTestClass
var isIntercepted = false; var isIntercepted = false;
c2.ApplicationMessageReceivedAsync += e => c2.ApplicationMessageReceivedAsync += e =>
{ {
isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; isIntercepted = string.Equals("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal);
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
@@ -432,21 +483,21 @@ public sealed class General_Tests : BaseTestClass
} }
[TestMethod] [TestMethod]
public async Task Publish_At_Least_Once_0x01() public Task Publish_At_Least_Once_0x01()
{ {
await TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.AtLeastOnce, "A/B/C", MqttQualityOfServiceLevel.AtLeastOnce, 1); return TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.AtLeastOnce, "A/B/C", MqttQualityOfServiceLevel.AtLeastOnce, 1);
} }
[TestMethod] [TestMethod]
public async Task Publish_At_Most_Once_0x00() public Task Publish_At_Most_Once_0x00()
{ {
await TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "A/B/C", MqttQualityOfServiceLevel.AtMostOnce, 1); return TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "A/B/C", MqttQualityOfServiceLevel.AtMostOnce, 1);
} }
[TestMethod] [TestMethod]
public async Task Publish_Exactly_Once_0x02() public Task Publish_Exactly_Once_0x02()
{ {
await TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, "A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, 1); return TestPublishAsync("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, "A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, 1);
} }
[TestMethod] [TestMethod]
@@ -886,59 +937,6 @@ public sealed class General_Tests : BaseTestClass
Assert.IsTrue(client.IsConnected); Assert.IsTrue(client.IsConnected);
} }
[TestMethod]
public async Task Disconnect_Client_with_Reason()
{
using var testEnvironment = CreateTestEnvironment();
var disconnectPacketReceived = false;
string testClientId = null;
await testEnvironment.StartServer();
testEnvironment.Server.ClientConnectedAsync += e =>
{
testClientId = e.ClientId;
return CompletedTask.Instance;
};
var client = testEnvironment.CreateClient();
client.InspectPacketAsync += e =>
{
if (e.Buffer.Length > 0)
{
if (e.Buffer[0] == (byte)MqttControlPacketType.Disconnect << 4)
{
disconnectPacketReceived = true;
}
}
return CompletedTask.Instance;
};
await client.ConnectAsync(
new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort)
.WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
.Build()
);
await LongTestDelay();
// Test client should be connected now
Assert.IsNotNull(testClientId);
// Have the server disconnect the client with SessionTakenOver reason; no DISCONNECT should be sent to MQTT 3.1.1 clients
await testEnvironment.Server.DisconnectClientAsync(testClientId, MqttDisconnectReasonCode.SessionTakenOver);
await LongTestDelay();
// MQTT 3.1.1 client should not receive a disconnect packet
Assert.IsFalse(disconnectPacketReceived, "Disconnect packet received when none is expected");
}
void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs) void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs)
{ {
if (!_connected.TryAdd(eventArgs.ClientId, true)) if (!_connected.TryAdd(eventArgs.ClientId, true))
@@ -295,10 +295,10 @@ public sealed class HotSwapCerts_Tests
} }
} }
class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable sealed class HotSwappableClientCertProvider : IMqttClientCertificatesProvider, IDisposable
{ {
X509Certificate2Collection _certificates; X509Certificate2Collection _certificates;
ConcurrentBag<X509Certificate2> _serverCerts = new ConcurrentBag<X509Certificate2>(); ConcurrentBag<X509Certificate2> _serverCerts = new();
public HotSwappableClientCertProvider() public HotSwappableClientCertProvider()
{ {
+3 -2
View File
@@ -1,6 +1,7 @@
#if DEBUG #if DEBUG
using System; using System;
using System.Globalization;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting; using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -56,7 +57,7 @@ public sealed class Load_Tests : BaseTestClass
for (var j = 0; j < 1000; j++) for (var j = 0; j < 1000; j++)
{ {
publishPacket.Topic = j.ToString(); publishPacket.Topic = j.ToString(CultureInfo.InvariantCulture);
await client.SendAsync(publishPacket, CancellationToken.None); await client.SendAsync(publishPacket, CancellationToken.None);
} }
@@ -143,7 +144,7 @@ public sealed class Load_Tests : BaseTestClass
for (var j = 0; j < 1000; j++) for (var j = 0; j < 1000; j++)
{ {
var message = applicationMessageBuilder.WithTopic(j.ToString()).Build(); var message = applicationMessageBuilder.WithTopic(j.ToString(CultureInfo.InvariantCulture)).Build();
await client.PublishAsync(message); await client.PublishAsync(message);
} }
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -16,10 +17,15 @@ namespace MQTTnet.Tests.Server;
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
[TestClass] [TestClass]
public sealed class MqttSubscriptionsManager_Tests : BaseTestClass public sealed class MqttSubscriptionsManager_Tests : BaseTestClass, IDisposable
{ {
MqttClientSubscriptionsManager _subscriptionsManager; MqttClientSubscriptionsManager _subscriptionsManager;
public void Dispose()
{
_subscriptionsManager?.Dispose();
}
[TestMethod] [TestMethod]
public async Task MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle() public async Task MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
{ {
@@ -46,7 +52,7 @@ public sealed class MqttSubscriptionsManager_Tests : BaseTestClass
{ {
TopicFilters = TopicFilters =
[ [
new() { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce } new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }
] ]
}; };
@@ -178,6 +184,48 @@ public sealed class MqttSubscriptionsManager_Tests : BaseTestClass
CheckIsNotSubscribed("house/2/room/bed"); CheckIsNotSubscribed("house/2/room/bed");
} }
[TestInitialize]
public void TestInitialize()
{
var logger = new TestLogger();
var options = new MqttServerOptions();
var retainedMessagesManager = new MqttRetainedMessagesManager(new MqttServerEventContainer(), logger);
var eventContainer = new MqttServerEventContainer();
var clientSessionManager = new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger);
var session = new MqttSession(
new MqttConnectPacket
{
ClientId = ""
},
new ConcurrentDictionary<object, object>(),
options,
eventContainer,
retainedMessagesManager,
clientSessionManager);
_subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerEventContainer(), retainedMessagesManager, clientSessionManager);
}
void CheckIsNotSubscribed(string topic)
{
var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, "");
Assert.IsFalse(result.IsSubscribed);
}
void CheckIsSubscribed(string topic)
{
var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, "");
Assert.IsTrue(result.IsSubscribed);
Assert.AreEqual(MqttQualityOfServiceLevel.AtMostOnce, result.QualityOfServiceLevel);
}
CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
{
MqttTopicHash.Calculate(topic, out var topicHash, out _, out _);
return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId);
}
async Task SubscribeToTopic(string topic) async Task SubscribeToTopic(string topic)
{ {
var sp = new MqttSubscribePacket var sp = new MqttSubscribePacket
@@ -190,40 +238,4 @@ public sealed class MqttSubscriptionsManager_Tests : BaseTestClass
await _subscriptionsManager.Subscribe(sp, CancellationToken.None); await _subscriptionsManager.Subscribe(sp, CancellationToken.None);
} }
void CheckIsSubscribed(string topic)
{
var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, "");
Assert.IsTrue(result.IsSubscribed);
Assert.AreEqual(MqttQualityOfServiceLevel.AtMostOnce, result.QualityOfServiceLevel);
}
void CheckIsNotSubscribed(string topic)
{
var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, "");
Assert.IsFalse(result.IsSubscribed);
}
[TestInitialize]
public void TestInitialize()
{
var logger = new TestLogger();
var options = new MqttServerOptions();
var retainedMessagesManager = new MqttRetainedMessagesManager(new MqttServerEventContainer(), logger);
var eventContainer = new MqttServerEventContainer();
var clientSessionManager = new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger);
var session = new MqttSession(new MqttConnectPacket
{
ClientId = ""
}, new ConcurrentDictionary<object, object>(), options, eventContainer, retainedMessagesManager, clientSessionManager);
_subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerEventContainer(), retainedMessagesManager, clientSessionManager);
}
CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
{
MqttTopicHash.Calculate(topic, out var topicHash, out _, out _);
return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId);
}
} }
@@ -24,7 +24,7 @@ public class Retained_Messages_Tests : BaseTestClass
var c1 = await testEnvironment.ConnectClient(); var c1 = await testEnvironment.ConnectClient();
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build()); await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload([]).WithRetainFlag().Build());
await c1.DisconnectAsync(); await c1.DisconnectAsync();
+22 -2
View File
@@ -14,8 +14,28 @@ namespace MQTTnet.Tests.Server;
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
[TestClass] [TestClass]
public sealed class Security_Tests : BaseTestClass public class Security_Tests : BaseTestClass
{ {
[TestMethod]
public async Task Deny_Connection()
{
using var testEnvironment = CreateTestEnvironment();
testEnvironment.IgnoreClientLogErrors = true;
var server = await testEnvironment.StartServer();
server.ValidatingConnectionAsync += e =>
{
e.ReasonCode = MqttConnectReasonCode.NotAuthorized;
return CompletedTask.Instance;
};
var client = testEnvironment.CreateClient();
var response = await client.ConnectAsync(testEnvironment.CreateDefaultClientOptions());
Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, response.ResultCode);
}
[TestMethod] [TestMethod]
public async Task Do_Not_Affect_Authorized_Clients() public async Task Do_Not_Affect_Authorized_Clients()
{ {
@@ -124,7 +144,7 @@ public sealed class Security_Tests : BaseTestClass
var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithCredentials(username, password).Build(); var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithCredentials(username, password).Build();
var ex = await Assert.ThrowsExactlyAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions)); var ex = await Assert.ThrowsExactlyAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions));
Assert.IsInstanceOfType(ex.InnerException, typeof(MqttProtocolViolationException)); Assert.IsInstanceOfType<MqttProtocolViolationException>(ex.InnerException);
Assert.AreEqual("Error while authenticating. If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].", ex.Message, false); Assert.AreEqual("Error while authenticating. If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].", ex.Message, false);
} }
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System; using System;
using System.Globalization;
using System.Linq; using System.Linq;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -222,7 +223,7 @@ public sealed class Subscribe_Tests : BaseTestClass
for (var i = 0; i < 500; i++) for (var i = 0; i < 500; i++)
{ {
var so = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(i.ToString()).Build(); var so = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(i.ToString(CultureInfo.InvariantCulture)).Build();
await c1.SubscribeAsync(so); await c1.SubscribeAsync(so);
@@ -234,7 +235,7 @@ public sealed class Subscribe_Tests : BaseTestClass
var messageBuilder = new MqttApplicationMessageBuilder(); var messageBuilder = new MqttApplicationMessageBuilder();
for (var i = 0; i < 500; i++) for (var i = 0; i < 500; i++)
{ {
messageBuilder.WithTopic(i.ToString()); messageBuilder.WithTopic(i.ToString(CultureInfo.InvariantCulture));
await c2.PublishAsync(messageBuilder.Build()); await c2.PublishAsync(messageBuilder.Build());
@@ -264,7 +265,7 @@ public sealed class Subscribe_Tests : BaseTestClass
var optionsBuilder = new MqttClientSubscribeOptionsBuilder(); var optionsBuilder = new MqttClientSubscribeOptionsBuilder();
for (var i = 0; i < 500; i++) for (var i = 0; i < 500; i++)
{ {
optionsBuilder.WithTopicFilter(i.ToString()); optionsBuilder.WithTopicFilter(i.ToString(CultureInfo.InvariantCulture));
} }
await c1.SubscribeAsync(optionsBuilder.Build()); await c1.SubscribeAsync(optionsBuilder.Build());
@@ -274,7 +275,7 @@ public sealed class Subscribe_Tests : BaseTestClass
var messageBuilder = new MqttApplicationMessageBuilder(); var messageBuilder = new MqttApplicationMessageBuilder();
for (var i = 0; i < 500; i++) for (var i = 0; i < 500; i++)
{ {
messageBuilder.WithTopic(i.ToString()); messageBuilder.WithTopic(i.ToString(CultureInfo.InvariantCulture));
await c2.PublishAsync(messageBuilder.Build()); await c2.PublishAsync(messageBuilder.Build());
} }
@@ -14,7 +14,7 @@ using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.Server; namespace MQTTnet.Tests.Server;
[TestClass] [TestClass]
public sealed class SubscriptionTopicHashTests public sealed class SubscriptionTopicHashTests : IDisposable
{ {
MqttSession _clientSession; MqttSession _clientSession;
@@ -84,6 +84,11 @@ public sealed class SubscriptionTopicHashTests
CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000); CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000);
} }
public void Dispose()
{
_clientSession?.Dispose();
}
/// <summary> /// <summary>
/// Test long topic name with last level being # /// Test long topic name with last level being #
/// </summary> /// </summary>
@@ -414,7 +419,7 @@ public sealed class SubscriptionTopicHashTests
foreach (var t in topics) foreach (var t in topics)
{ {
// replace last letter with x // replace last letter with x
topicsToFind.Add(t.Substring(0, t.Length - 1) + "x"); topicsToFind.Add(string.Concat(t.AsSpan(0, t.Length - 1), "x"));
} }
matchCount = CheckTopicSubscriptions(_clientSession, topicsToFind, "No Wildcard No Match (Replace X)"); matchCount = CheckTopicSubscriptions(_clientSession, topicsToFind, "No Wildcard No Match (Replace X)");
@@ -468,7 +473,7 @@ public sealed class SubscriptionTopicHashTests
Assert.AreEqual(0xff, hashMaskBytes[7], "mask 7 mismatch"); Assert.AreEqual(0xff, hashMaskBytes[7], "mask 7 mismatch");
} }
void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask) static void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask)
{ {
MqttTopicHash.Calculate(topic, out var topicHash, out var hashMask, out _); MqttTopicHash.Calculate(topic, out var topicHash, out var hashMask, out _);
@@ -481,7 +486,7 @@ public sealed class SubscriptionTopicHashTests
Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected"); Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected");
} }
int CheckTopicSubscriptions(MqttSession clientSession, List<string> topicsToFind, string subject) static int CheckTopicSubscriptions(MqttSession clientSession, List<string> topicsToFind, string subject)
{ {
var matchCount = 0; var matchCount = 0;
@@ -517,7 +522,7 @@ public sealed class SubscriptionTopicHashTests
return matchCount; return matchCount;
} }
byte[] GetBytes(ulong value) static byte[] GetBytes(ulong value)
{ {
var bytes = BitConverter.GetBytes(value); var bytes = BitConverter.GetBytes(value);
// Ensure that highest byte comes first for comparison left to right // Ensure that highest byte comes first for comparison left to right

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