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
208 changed files with 2969 additions and 1226 deletions
+55 -6
View File
@@ -1,9 +1,58 @@
<Project> <Project>
<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>
<PropertyGroup> <DelaySign>false</DelaySign>
<TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure> <SignAssembly>false</SignAssembly>
<TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport> <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<EnableMSTestRunner>true</EnableMSTestRunner> <EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
</PropertyGroup> <ImplicitUsings>enable</ImplicitUsings>
</Project> <NeutralLanguage>en-US</NeutralLanguage>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<EnableNETAnalyzers>true</EnableNETAnalyzers>
<AnalysisLevel>latest</AnalysisLevel>
<AnalysisMode>recommended</AnalysisMode>
<AnalysisModeStyle>default</AnalysisModeStyle>
<TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
<TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
<EnableMSTestRunner>true</EnableMSTestRunner>
</PropertyGroup>
<PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
</PropertyGroup>
<PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageIcon>nuget.png</PackageIcon>
<PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
<RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
</PropertyGroup>
<ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
<None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
<None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
<None Include="..\..\LICENSE" Pack="true" PackagePath="" />
</ItemGroup>
<ItemGroup Condition="'$(IncludeSymbols)'=='true'">
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
</ItemGroup>
</Project>
+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());
+7 -5
View File
@@ -8,17 +8,19 @@ 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}");
return @object; return @object;
} }
+1 -11
View File
@@ -2,20 +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>
<LangVersion>default</LangVersion>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
+4 -3
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();
@@ -58,7 +59,7 @@ try
{ {
if (selectedSampleMethod.Invoke(null, null) is Task task) if (selectedSampleMethod.Invoke(null, null) is Task task)
{ {
await task; await task;
} }
} }
catch (Exception exception) catch (Exception exception)
+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,61 +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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable> <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 #.");
} }
@@ -12,12 +12,7 @@ namespace MQTTnet.Extensions.Rpc;
public interface IMqttRpcClient : IDisposable public interface IMqttRpcClient : IDisposable
{ {
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object>? parameters = null); Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null);
Task<byte[]> ExecuteAsync( Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default);
string methodName,
byte[] payload,
MqttQualityOfServiceLevel qualityOfServiceLevel,
IDictionary<string, object>? parameters = null,
CancellationToken cancellationToken = default);
} }
@@ -1,55 +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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable> <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>
+9 -18
View File
@@ -20,7 +20,7 @@ public sealed class MqttRpcClient : IMqttRpcClient
readonly IMqttClient _mqttClient; readonly IMqttClient _mqttClient;
readonly MqttRpcClientOptions _options; readonly MqttRpcClientOptions _options;
readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new(); readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>>();
public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options) public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options)
{ {
@@ -42,12 +42,7 @@ public sealed class MqttRpcClient : IMqttRpcClient
_waitingCalls.Clear(); _waitingCalls.Clear();
} }
public async Task<byte[]> ExecuteAsync( public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null)
TimeSpan timeout,
string methodName,
byte[] payload,
MqttQualityOfServiceLevel qualityOfServiceLevel,
IDictionary<string, object>? parameters = null)
{ {
using var timeoutToken = new CancellationTokenSource(timeout); using var timeoutToken = new CancellationTokenSource(timeout);
try try
@@ -65,12 +60,7 @@ public sealed class MqttRpcClient : IMqttRpcClient
} }
} }
public async Task<byte[]> ExecuteAsync( public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
string methodName,
byte[] payload,
MqttQualityOfServiceLevel qualityOfServiceLevel,
IDictionary<string, object>? parameters = null,
CancellationToken cancellationToken = default)
{ {
ArgumentNullException.ThrowIfNull(methodName); ArgumentNullException.ThrowIfNull(methodName);
@@ -92,7 +82,7 @@ public sealed class MqttRpcClient : IMqttRpcClient
var requestMessageBuilder = new MqttApplicationMessageBuilder().WithTopic(requestTopic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel); var requestMessageBuilder = new MqttApplicationMessageBuilder().WithTopic(requestTopic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel);
if (_mqttClient.Options!.ProtocolVersion == MqttProtocolVersion.V500) if (_mqttClient.Options.ProtocolVersion == MqttProtocolVersion.V500)
{ {
requestMessageBuilder.WithResponseTopic(responseTopic); requestMessageBuilder.WithResponseTopic(responseTopic);
} }
@@ -113,10 +103,11 @@ public sealed class MqttRpcClient : IMqttRpcClient
await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false); await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false); await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
using (cancellationToken.Register(() => using (cancellationToken.Register(
{ () =>
awaitable.TrySetCanceled(); {
})) awaitable.TrySetCanceled();
}))
{ {
return await awaitable.Task.ConfigureAwait(false); return await awaitable.Task.ConfigureAwait(false);
} }
@@ -12,9 +12,9 @@ namespace MQTTnet.Extensions.Rpc;
public static class MqttRpcClientExtensions 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);
@@ -6,7 +6,7 @@ namespace MQTTnet.Extensions.Rpc;
public sealed class MqttRpcTopicPair public sealed class MqttRpcTopicPair
{ {
public required string RequestTopic { get; init; } public string RequestTopic { get; set; }
public required string ResponseTopic { get; init; } public string ResponseTopic { get; set; }
} }
@@ -10,12 +10,8 @@ namespace MQTTnet.Extensions.Rpc;
public sealed class TopicGenerationContext public sealed class TopicGenerationContext
{ {
public TopicGenerationContext( public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName,
IMqttClient mqttClient, IDictionary<string, object> parameters, MqttQualityOfServiceLevel qualityOfServiceLevel)
MqttRpcClientOptions options,
string methodName,
IDictionary<string, object>? parameters,
MqttQualityOfServiceLevel qualityOfServiceLevel)
{ {
MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName)); MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
Parameters = parameters; Parameters = parameters;
@@ -26,12 +22,12 @@ public sealed class TopicGenerationContext
public string MethodName { get; } public string MethodName { get; }
public IDictionary<string, object> Parameters { get; }
public IMqttClient MqttClient { get; } public IMqttClient MqttClient { get; }
public MqttRpcClientOptions Options { get; } public MqttRpcClientOptions Options { get; }
public IDictionary<string, object>? Parameters { get; }
/// <summary> /// <summary>
/// Gets or sets the quality of service level. /// Gets or sets the quality of service level.
/// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message /// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message
@@ -1,56 +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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</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
@@ -8,9 +8,9 @@ namespace MQTTnet.Server.EnhancedAuthentication;
public sealed class ExchangeEnhancedAuthenticationOptions public sealed class ExchangeEnhancedAuthenticationOptions
{ {
public byte[]? AuthenticationData { get; set; } public byte[] AuthenticationData { get; set; }
public string? ReasonString { get; set; } public string ReasonString { get; set; }
public List<MqttUserProperty>? UserProperties { get; set; } public List<MqttUserProperty> UserProperties { get; set; }
} }
@@ -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)
{ {
@@ -4,9 +4,9 @@ namespace MQTTnet.Server.EnhancedAuthentication;
public sealed class ExchangeEnhancedAuthenticationResult public sealed class ExchangeEnhancedAuthenticationResult
{ {
public string? ReasonString { get; init; } public string ReasonString { get; set; }
public List<MqttUserProperty>? UserProperties { get; init; } public List<MqttUserProperty> UserProperties { get; set; }
public byte[]? AuthenticationData { get; init; } public byte[] AuthenticationData { get; set; }
} }
@@ -22,15 +22,15 @@ public sealed class ClientConnectedEventArgs : EventArgs
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
} }
public byte[]? AuthenticationData => _connectPacket.AuthenticationData; public byte[] AuthenticationData => _connectPacket.AuthenticationData;
public string? AuthenticationMethod => _connectPacket.AuthenticationMethod; public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
/// <summary> /// <summary>
/// Gets the client identifier of the connected client. /// Gets the client identifier of the connected client.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary> /// </summary>
public string ClientId => _connectPacket.ClientId!; public string ClientId => _connectPacket.ClientId;
/// <summary> /// <summary>
/// Gets the endpoint of the connected client. /// Gets the endpoint of the connected client.
@@ -38,7 +38,7 @@ public sealed class ClientConnectedEventArgs : EventArgs
public EndPoint RemoteEndPoint { get; } public EndPoint RemoteEndPoint { get; }
[Obsolete("Use RemoteEndPoint instead.")] [Obsolete("Use RemoteEndPoint instead.")]
public string? Endpoint => RemoteEndPoint?.ToString(); public string Endpoint => RemoteEndPoint?.ToString();
/// <summary> /// <summary>
/// Gets the protocol version which is used by the connected client. /// Gets the protocol version which is used by the connected client.
@@ -53,16 +53,16 @@ public sealed class ClientConnectedEventArgs : EventArgs
/// <summary> /// <summary>
/// Gets the user name of the connected client. /// Gets the user name of the connected client.
/// </summary> /// </summary>
public string? UserName => _connectPacket.Username; public string UserName => _connectPacket.Username;
/// <summary> /// <summary>
/// Gets the password of the connected client. /// Gets the password of the connected client.
/// </summary> /// </summary>
public string? Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan()); public string Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
/// <summary> /// <summary>
/// Gets the user properties sent by the client. /// Gets the user properties sent by the client.
/// <remarks>MQTT 5.0.0+ feature.</remarks> /// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary> /// </summary>
public List<MqttUserProperty>? UserProperties => _connectPacket?.UserProperties; public List<MqttUserProperty> UserProperties => _connectPacket?.UserProperties;
} }
@@ -35,24 +35,24 @@ public sealed class ClientDisconnectedEventArgs : EventArgs
/// Gets the client identifier. /// Gets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary> /// </summary>
public string ClientId => _connectPacket.ClientId!; public string ClientId => _connectPacket.ClientId;
/// <summary> /// <summary>
/// Gets the user name of the client. /// Gets the user name of the client.
/// </summary> /// </summary>
public string? UserName => _connectPacket.Username; public string UserName => _connectPacket.Username;
/// <summary> /// <summary>
/// Gets the password of the client. /// Gets the password of the client.
/// </summary> /// </summary>
public string? Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan()); public string Password => Encoding.UTF8.GetString(_connectPacket.Password.AsSpan());
public MqttClientDisconnectType DisconnectType { get; } public MqttClientDisconnectType DisconnectType { get; }
public EndPoint RemoteEndPoint { get; } public EndPoint RemoteEndPoint { get; }
[Obsolete("Use RemoteEndPoint instead.")] [Obsolete("Use RemoteEndPoint instead.")]
public string? Endpoint => RemoteEndPoint?.ToString(); public string Endpoint => RemoteEndPoint?.ToString();
/// <summary> /// <summary>
/// Gets the reason code sent by the client. /// Gets the reason code sent by the client.
@@ -66,7 +66,7 @@ public sealed class ClientDisconnectedEventArgs : EventArgs
/// Only available for clean disconnects. /// Only available for clean disconnects.
/// <remarks>MQTT 5.0.0+ feature.</remarks> /// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary> /// </summary>
public string? ReasonString => _disconnectPacket?.ReasonString; public string ReasonString => _disconnectPacket?.ReasonString;
/// <summary> /// <summary>
/// Gets the session expiry interval sent by the client. /// Gets the session expiry interval sent by the client.
@@ -85,5 +85,5 @@ public sealed class ClientDisconnectedEventArgs : EventArgs
/// Only available for clean disconnects. /// Only available for clean disconnects.
/// <remarks>MQTT 5.0.0+ feature.</remarks> /// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary> /// </summary>
public List<MqttUserProperty>? UserProperties => _disconnectPacket?.UserProperties; public List<MqttUserProperty> UserProperties => _disconnectPacket?.UserProperties;
} }
@@ -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;
@@ -12,7 +12,7 @@ namespace MQTTnet.Server.Internal.Adapter;
public sealed class MqttTcpServerAdapter : IMqttServerAdapter public sealed class MqttTcpServerAdapter : IMqttServerAdapter
{ {
readonly List<MqttTcpServerListener> _listeners = new(); readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
CancellationTokenSource _cancellationTokenSource; CancellationTokenSource _cancellationTokenSource;
@@ -165,8 +165,8 @@ public sealed class MqttTcpServerListener : IDisposable
async Task TryHandleClientConnectionAsync(CrossPlatformSocket clientSocket) async Task TryHandleClientConnectionAsync(CrossPlatformSocket clientSocket)
{ {
Stream? stream = null; Stream stream = null;
EndPoint? remoteEndPoint = null; EndPoint remoteEndPoint = null;
try try
{ {
@@ -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);
} }
@@ -667,7 +667,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
} }
} }
async Task<MqttConnectPacket?> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) async Task<MqttConnectPacket> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{ {
try try
{ {
@@ -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 -40
View File
@@ -1,57 +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>enable</Nullable>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<IsAotCompatible>true</IsAotCompatible>
<LangVersion>default</LangVersion>
</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>
@@ -21,7 +21,7 @@ public static class MqttServerExtensions
public static Task InjectApplicationMessage( public static Task InjectApplicationMessage(
this MqttServer server, this MqttServer server,
string topic, string topic,
string? payload = null, string payload = null,
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
bool retain = false) bool retain = false)
{ {
@@ -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;
@@ -20,7 +20,7 @@ public class MqttServerOptionsBuilder
return _options; return _options;
} }
public MqttServerOptionsBuilder WithClientCertificate(RemoteCertificateValidationCallback? validationCallback = null, bool checkCertificateRevocation = false) public MqttServerOptionsBuilder WithClientCertificate(RemoteCertificateValidationCallback validationCallback = null, bool checkCertificateRevocation = false)
{ {
_options.TlsEndpointOptions.ClientCertificateRequired = true; _options.TlsEndpointOptions.ClientCertificateRequired = true;
_options.TlsEndpointOptions.CheckCertificateRevocation = checkCertificateRevocation; _options.TlsEndpointOptions.CheckCertificateRevocation = checkCertificateRevocation;
@@ -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; }
} }
+2 -2
View File
@@ -16,7 +16,7 @@ public sealed class SubscribeResponse
/// </summary> /// </summary>
public MqttSubscribeReasonCode ReasonCode { get; set; } public MqttSubscribeReasonCode ReasonCode { get; set; }
public string? ReasonString { get; set; } public string ReasonString { get; set; }
public List<MqttUserProperty>? UserProperties { get; set; } = []; public List<MqttUserProperty> UserProperties { get; } = [];
} }
+2 -2
View File
@@ -15,7 +15,7 @@ public sealed class UnsubscribeResponse
/// </summary> /// </summary>
public MqttUnsubscribeReasonCode ReasonCode { get; set; } public MqttUnsubscribeReasonCode ReasonCode { get; set; }
public string? ReasonString { get; set; } public string ReasonString { get; set; }
public List<MqttUserProperty>? UserProperties { get; } = []; public List<MqttUserProperty> UserProperties { get; } = [];
} }
+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,19 +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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</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);
} }
+2 -12
View File
@@ -3,19 +3,10 @@
<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<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>
<LangVersion>default</LangVersion> <LangVersion>default</LangVersion>
<Nullable>enable</Nullable> <Nullable>disable</Nullable>
<NoWarn>$(NoWarn);CA1707</NoWarn>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -23,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();
} }
} }

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