Compare commits

..
Author SHA1 Message Date
christian 37631730e0 Enable nullable 2025-10-20 11:29:24 +02:00
208 changed files with 1216 additions and 2959 deletions
+5 -54
View File
@@ -1,58 +1,9 @@
<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>
<DelaySign>false</DelaySign> <PropertyGroup>
<SignAssembly>false</SignAssembly> <TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks> <TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild> <EnableMSTestRunner>true</EnableMSTestRunner>
<ImplicitUsings>enable</ImplicitUsings> </PropertyGroup>
<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> </Project>
+3 -8
View File
@@ -7,12 +7,11 @@ 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}"
@@ -33,8 +32,6 @@ 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
@@ -81,10 +78,6 @@ 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
@@ -92,4 +85,6 @@ 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=
} }
} }
sealed class SampleClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler 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(IMqttClient mqttClient, CancellationToken shutdownToken) static void ConcurrentProcessingDisableAutoAcknowledge(CancellationToken shutdownToken, IMqttClient mqttClient)
{ {
/* /*
* 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(IMqttClient mqttClient, CancellationToken shutdownToken) static void ConcurrentProcessingWithLimit(CancellationToken shutdownToken, IMqttClient mqttClient)
{ {
/* /*
* This sample shows how to achieve concurrent processing, with: * This sample shows how to achieve concurrent processing, with:
+1 -2
View File
@@ -6,7 +6,6 @@
// 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;
@@ -52,7 +51,7 @@ public static class Logger_Samples
mqttEventLogger.LogMessagePublished += (_, args) => mqttEventLogger.LogMessagePublished += (_, args) =>
{ {
var output = new StringBuilder(); var output = new StringBuilder();
output.AppendLine(CultureInfo.InvariantCulture, $">> [{args.LogMessage.Timestamp:O}] [{args.LogMessage.ThreadId}] [{args.LogMessage.Source}] [{args.LogMessage.Level}]: {args.LogMessage.Message}"); output.AppendLine($">> [{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());
+4 -6
View File
@@ -8,17 +8,15 @@ 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, SerializerOptions); output = JsonSerializer.Serialize(@object, new JsonSerializerOptions
{
WriteIndented = true
});
} }
Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}"); Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}");
+11 -1
View File
@@ -2,10 +2,20 @@
<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>
<NoWarn>$(NoWarn);CA1707</NoWarn> <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
<LangVersion>default</LangVersion>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
+2 -3
View File
@@ -2,7 +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.Globalization;
using System.Reflection; using System.Reflection;
Console.BackgroundColor = ConsoleColor.White; Console.BackgroundColor = ConsoleColor.White;
@@ -29,7 +28,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", CultureInfo.InvariantCulture); var selectedIndex = int.Parse(input ?? "0");
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();
@@ -46,7 +45,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", CultureInfo.InvariantCulture); selectedIndex = int.Parse(input ?? "0");
var selectedSampleMethod = sampleMethods[selectedIndex]; var selectedSampleMethod = sampleMethods[selectedIndex];
Console.WriteLine(); Console.WriteLine();
+4 -7
View File
@@ -8,7 +8,6 @@
// 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;
@@ -53,14 +52,14 @@ public static class Server_ASP_NET_Samples
// Inject other services via constructor. // Inject other services via constructor.
} }
public static Task OnClientConnected(ClientConnectedEventArgs eventArgs) public Task OnClientConnected(ClientConnectedEventArgs eventArgs)
{ {
Console.WriteLine($"Client '{eventArgs.ClientId}' connected."); Console.WriteLine($"Client '{eventArgs.ClientId}' connected.");
return Task.CompletedTask; return Task.CompletedTask;
} }
public static Task ValidateConnection(ValidatingConnectionEventArgs eventArgs) public 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;
@@ -69,7 +68,6 @@ 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();
@@ -90,12 +88,11 @@ 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,7 +9,6 @@
// 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;
@@ -43,32 +42,4 @@ 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();
}
} }
+2 -3
View File
@@ -6,7 +6,6 @@
// 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;
@@ -162,7 +161,7 @@ public static class Server_Simple_Samples
return server; return server;
} }
sealed class ConsoleLogger : IMqttNetLogger class ConsoleLogger : IMqttNetLogger
{ {
readonly object _consoleSyncRoot = new(); readonly object _consoleSyncRoot = new();
@@ -192,7 +191,7 @@ public static class Server_Simple_Samples
if (parameters?.Length > 0) if (parameters?.Length > 0)
{ {
message = string.Format(CultureInfo.InvariantCulture, message, parameters); message = string.Format(message, parameters);
} }
lock (_consoleSyncRoot) lock (_consoleSyncRoot)
@@ -1,10 +1,19 @@
<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>
<NoWarn>$(NoWarn);CA1707</NoWarn> <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<EnableNETAnalyzers>false</EnableNETAnalyzers>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<NoWarn>1591;NETSDK1138;NU1803;NU1901;NU1902</NoWarn>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<NuGetAuditMode>all</NuGetAuditMode>
<NuGetAudit>true</NuGetAudit>
<NuGetAuditLevel>low</NuGetAuditLevel>
<AnalysisLevel>latest-Recommended</AnalysisLevel>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -6,7 +6,7 @@ using Microsoft.AspNetCore.Mvc.RazorPages;
namespace MQTTnet.AspTestApp.Pages; namespace MQTTnet.AspTestApp.Pages;
public partial class IndexModel : PageModel public class IndexModel : PageModel
{ {
readonly ILogger<IndexModel> _logger; readonly ILogger<IndexModel> _logger;
@@ -17,9 +17,6 @@ public partial class IndexModel : PageModel
public void OnGet() public void OnGet()
{ {
LogOnGet(); _logger.LogDebug("OnGet");
} }
[LoggerMessage(Level = LogLevel.Information, Message = "OnGet")]
private partial void LogOnGet();
} }
@@ -1,23 +1,61 @@
<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>
<IsAotCompatible>true</IsAotCompatible> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Company>The contributors of MQTTnet</Company>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild> <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> <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> <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> <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> <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> <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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</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)
{ {
ArgumentNullException.ThrowIfNull(options); if (options == null) throw new ArgumentNullException(nameof(options));
switch (options.ChannelOptions) switch (options.ChannelOptions)
{ {
@@ -61,6 +61,27 @@ 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
@@ -83,50 +104,8 @@ 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)
@@ -253,8 +232,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);
var offset = buffer.Packet.Count; int offset = buffer.Packet.Count;
buffer.Payload.CopyTo(span.Slice(offset)); buffer.Payload.CopyTo(destination: 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.ToLowerInvariant().StartsWith("mqtt", StringComparison.InvariantCulture)); return requestedSubProtocolValues.OrderByDescending(p => p.Length).FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
} }
} }
@@ -81,7 +81,7 @@ public sealed class SocketConnection : ConnectionContext
IsConnected = true; IsConnected = true;
} }
static MqttCommunicationException ConnectionAborted() static Exception ConnectionAborted()
{ {
return new MqttCommunicationException("Connection Aborted"); return new MqttCommunicationException("Connection Aborted");
} }
@@ -218,16 +218,12 @@ public sealed class SocketConnection : ConnectionContext
var flushTask = _application.Output.FlushAsync(); var flushTask = _application.Output.FlushAsync();
FlushResult result;
if (!flushTask.IsCompleted) if (!flushTask.IsCompleted)
{ {
result = await flushTask; 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
+1 -6
View File
@@ -8,7 +8,7 @@ using System.Net.Sockets;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
public sealed class SocketReceiver : IDisposable public sealed class SocketReceiver
{ {
readonly SocketAwaitable _awaitable; readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new(); readonly SocketAsyncEventArgs _eventArgs = new();
@@ -33,9 +33,4 @@ public sealed class SocketReceiver : IDisposable
return _awaitable; return _awaitable;
} }
public void Dispose()
{
_eventArgs.Dispose();
}
} }
+1 -6
View File
@@ -12,7 +12,7 @@ using System.Runtime.InteropServices;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
public sealed class SocketSender : IDisposable public sealed class SocketSender
{ {
readonly SocketAwaitable _awaitable; readonly SocketAwaitable _awaitable;
readonly SocketAsyncEventArgs _eventArgs = new(); readonly SocketAsyncEventArgs _eventArgs = new();
@@ -28,11 +28,6 @@ public sealed class SocketSender : IDisposable
_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 InvalidOperationException($"Code is broken ({globalI})!"); throw new Exception($"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, IDisposable public sealed class ChannelAdapterBenchmark : BaseBenchmark
{ {
MqttChannelAdapter _channelAdapter; MqttChannelAdapter _channelAdapter;
int _iterations; int _iterations;
@@ -86,10 +86,4 @@ public sealed class ChannelAdapterBenchmark : BaseBenchmark, IDisposable
return buffer.ToArray(); return buffer.ToArray();
} }
public void Dispose()
{
_channelAdapter?.Dispose();
_stream?.Dispose();
}
} }
@@ -1,11 +1,20 @@
<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>
<NoWarn>$(NoWarn);CS8892;CA1707;CA1051;CA1822</NoWarn> <EnableNETAnalyzers>false</EnableNETAnalyzers>
<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,6 +2,10 @@
// 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;
@@ -9,7 +13,7 @@ using MQTTnet.Server;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
[MemoryDiagnoser] [MemoryDiagnoser]
public sealed class MessageDeliveryBenchmark : BaseBenchmark, IDisposable public class MessageDeliveryBenchmark : BaseBenchmark
{ {
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;
@@ -95,16 +99,10 @@ public sealed class MessageDeliveryBenchmark : BaseBenchmark, IDisposable
if (_messagesReceivedCount < _messagesExpectedCount) if (_messagesReceivedCount < _messagesExpectedCount)
{ {
throw new InvalidOperationException($"Messages Received Count mismatch, expected {_messagesExpectedCount}, received {_messagesReceivedCount}"); throw new Exception($"Messages Received Count mismatch, expected {_messagesExpectedCount}, received {_messagesReceivedCount}");
} }
} }
public void Dispose()
{
_cancellationTokenSource?.Dispose();
_mqttServer?.Dispose();
}
[GlobalSetup] [GlobalSetup]
public void Setup() public void Setup()
{ {
@@ -175,7 +173,7 @@ public sealed class MessageDeliveryBenchmark : BaseBenchmark, IDisposable
var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers); var topicIndexStep = totalNumTopics / (_numSubscribedTopicsPerSubscriber * _numSubscribers);
if (topicIndexStep * _numSubscribedTopicsPerSubscriber * _numSubscribers != totalNumTopics) if (topicIndexStep * _numSubscribedTopicsPerSubscriber * _numSubscribers != totalNumTopics)
{ {
throw new InvalidOperationException( throw new Exception(
$"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,6 +4,8 @@
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;
@@ -16,9 +18,9 @@ namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)] [SimpleJob(RuntimeMoniker.Net60)]
[MemoryDiagnoser] [MemoryDiagnoser]
public sealed class MqttTcpChannelBenchmark : BaseBenchmark, IDisposable public class MqttTcpChannelBenchmark : BaseBenchmark
{ {
MqttTcpChannel _clientChannel; IMqttChannel _clientChannel;
MqttServer _mqttServer; MqttServer _mqttServer;
IMqttChannel _serverChannel; IMqttChannel _serverChannel;
@@ -82,11 +84,4 @@ public sealed class MqttTcpChannelBenchmark : BaseBenchmark, IDisposable
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], StringComparison.Ordinal)); _selectedBenchmarkIndex = _benchmarks.FindIndex(b => b.Name.Equals(arguments[0]));
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 sealed class ReaderExtensionsBenchmark : IDisposable, IAsyncDisposable public class ReaderExtensionsBenchmark
{ {
MqttPacketFormatterAdapter _mqttPacketFormatter; MqttPacketFormatterAdapter _mqttPacketFormatter;
MemoryStream _stream; MemoryStream _stream;
@@ -83,17 +83,4 @@ public sealed class ReaderExtensionsBenchmark : IDisposable, IAsyncDisposable
} }
} }
} }
public void Dispose()
{
_stream?.Dispose();
}
public async ValueTask DisposeAsync()
{
if (_stream != null)
{
await _stream.DisposeAsync();
}
}
} }
@@ -1,19 +1,39 @@
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] [RPlotExporter, RankColumn]
[RankColumn]
[MemoryDiagnoser] [MemoryDiagnoser]
public sealed class SendPacketAsyncBenchmark : BaseBenchmark, IDisposable, IAsyncDisposable public class SendPacketAsyncBenchmark : BaseBenchmark
{ {
MqttPacketBuffer _buffer;
MemoryStream _stream; MemoryStream _stream;
MqttPacketBuffer _buffer;
[GlobalSetup]
public void GlobalSetup()
{
_stream = new MemoryStream(1024);
var packet = new ArraySegment<byte>(new byte[10]);
_buffer = new MqttPacketBuffer(packet);
}
[Benchmark(Baseline = true)]
public async ValueTask Before()
{
_stream.Position = 0;
var output = PipeWriter.Create(_stream);
WritePacketBuffer(output, _buffer);
await output.FlushAsync();
}
[Benchmark] [Benchmark]
public async ValueTask After() public async ValueTask After()
@@ -32,37 +52,6 @@ public sealed class SendPacketAsyncBenchmark : BaseBenchmark, IDisposable, IAsyn
} }
} }
[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;
MqttV3PacketFormatter _serializer; IMqttPacketFormatter _serializer;
MqttBufferWriter _bufferWriter; MqttBufferWriter _bufferWriter;
[GlobalSetup] [GlobalSetup]
@@ -43,7 +43,7 @@ public class SerializerBenchmark : BaseBenchmark
} }
[Benchmark] [Benchmark]
public void Serialize10000Messages() public void Serialize_10000_Messages()
{ {
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 Deserialize10000Messages() public void Deserialize_10000_Messages()
{ {
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
} }
} }
sealed class BenchmarkMqttChannel : IMqttChannel class BenchmarkMqttChannel : IMqttChannel
{ {
readonly ArraySegment<byte> _buffer; readonly ArraySegment<byte> _buffer;
int _position; int _position;
@@ -79,9 +79,7 @@ public class SerializerBenchmark : BaseBenchmark
public EndPoint RemoteEndPoint { get; set; } public EndPoint RemoteEndPoint { get; set; }
public EndPoint LocalEndPoint { get; set; } public bool IsSecureConnection { get; } = false;
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,7 +12,12 @@ 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(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default); Task<byte[]> ExecuteAsync(
string methodName,
byte[] payload,
MqttQualityOfServiceLevel qualityOfServiceLevel,
IDictionary<string, object>? parameters = null,
CancellationToken cancellationToken = default);
} }
@@ -1,19 +1,55 @@
<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>
<IsAotCompatible>true</IsAotCompatible>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <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> <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> <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> <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> <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>
<DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
<GenerateDocumentationFile>true</GenerateDocumentationFile> <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>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</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>
+18 -9
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 ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>>(); readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new();
public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options) public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options)
{ {
@@ -42,7 +42,12 @@ public sealed class MqttRpcClient : IMqttRpcClient
_waitingCalls.Clear(); _waitingCalls.Clear();
} }
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null) public async Task<byte[]> ExecuteAsync(
TimeSpan timeout,
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
@@ -60,7 +65,12 @@ public sealed class MqttRpcClient : IMqttRpcClient
} }
} }
public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default) public async Task<byte[]> ExecuteAsync(
string methodName,
byte[] payload,
MqttQualityOfServiceLevel qualityOfServiceLevel,
IDictionary<string, object>? parameters = null,
CancellationToken cancellationToken = default)
{ {
ArgumentNullException.ThrowIfNull(methodName); ArgumentNullException.ThrowIfNull(methodName);
@@ -82,7 +92,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);
} }
@@ -103,11 +113,10 @@ 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)
{ {
ArgumentNullException.ThrowIfNull(client); if (client == null) throw new ArgumentNullException(nameof(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 string RequestTopic { get; set; } public required string RequestTopic { get; init; }
public string ResponseTopic { get; set; } public required string ResponseTopic { get; init; }
} }
@@ -10,8 +10,12 @@ namespace MQTTnet.Extensions.Rpc;
public sealed class TopicGenerationContext public sealed class TopicGenerationContext
{ {
public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName, public TopicGenerationContext(
IDictionary<string, object> parameters, MqttQualityOfServiceLevel qualityOfServiceLevel) IMqttClient mqttClient,
MqttRpcClientOptions options,
string methodName,
IDictionary<string, object>? parameters,
MqttQualityOfServiceLevel qualityOfServiceLevel)
{ {
MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName)); MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
Parameters = parameters; Parameters = parameters;
@@ -22,12 +26,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,24 +1,56 @@
<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) && !filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard)) if (filter.Length > 0 && !filter.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) && !filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
{ {
filter += MqttTopicFilterComparer.LevelSeparator; filter += MqttTopicFilterComparer.LevelSeparator;
} }
// append hash if neccessary // append hash if neccessary
if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard)) if (!filter.EndsWith(MqttTopicFilterComparer.MultiLevelWildcard.ToString()))
{ {
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(), StringComparison.InvariantCulture) && !root.EndsWith('}')) if (root.Contains(MqttTopicFilterComparer.LevelSeparator) && !root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()) && !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(), StringComparison.InvariantCulture)) if (root.EndsWith(MqttTopicFilterComparer.LevelSeparator.ToString()))
{ {
root += MqttTopicFilterComparer.SingleLevelWildcard; root += MqttTopicFilterComparer.SingleLevelWildcard;
} }
@@ -1,47 +0,0 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Connect, "MqttSession")]
[OutputType(typeof(MqttClientConnectResult))]
public class ConnectMqttSessionCmdlet : PSCmdlet
{
[Parameter]
public SwitchParameter CleanSession { get; set; } = true;
[Parameter]
public string? ClientId { get; set; } = Guid.NewGuid().ToString();
[Parameter(Mandatory = true)]
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);
}
}
@@ -1,31 +0,0 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Disconnect, "MqttSession")]
public class DisconnectMqttSessionCmdlet : PSCmdlet
{
[Parameter]
public MqttClientDisconnectOptionsReason Reason { get; set; } = MqttClientDisconnectOptionsReason.NormalDisconnection;
[Parameter]
public string? ReasonString { get; set; }
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
public uint SessionExpiryInterval { get; set; }
protected override void ProcessRecord()
{
if (Session.GetClient().IsConnected)
{
var options = new MqttClientDisconnectOptionsBuilder().WithSessionExpiryInterval(SessionExpiryInterval).WithReason(Reason).WithReasonString(ReasonString).Build();
Session.GetClient().DisconnectAsync(options).GetAwaiter().GetResult();
}
WriteObject("Disconnected.");
}
}
@@ -1,13 +0,0 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommon.New, "MqttSession")]
[OutputType(typeof(PsMqttSession))]
public class NewMqttSessionCmdlet : PSCmdlet
{
protected override void ProcessRecord()
{
WriteObject(new PsMqttSession());
}
}
@@ -1,52 +0,0 @@
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);
}
}
@@ -1,74 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommunications.Receive, "MqttMessage")]
[OutputType(typeof(PsMqttMessage))]
public class ReceiveMqttMessageCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
[ValidateRange(0, int.MaxValue)]
public int TimeoutSeconds { get; set; } = 0;
protected override void ProcessRecord()
{
var tcs = new TaskCompletionSource<PsMqttMessage>();
EventHandler<PsMqttMessage>? handler = null;
CancellationTokenSource? cts = null;
try
{
handler = (s, e) =>
{
if (!tcs.Task.IsCompleted)
{
tcs.TrySetResult(e);
}
};
Session.MessageReceived += handler;
// Optionaler Timeout
if (TimeoutSeconds > 0)
{
cts = new CancellationTokenSource();
cts.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
cts.Token.Register(() =>
{
if (!tcs.Task.IsCompleted)
{
tcs.TrySetCanceled();
}
});
}
WriteVerbose("Waiting for next MQTT message...");
try
{
var message = tcs.Task.GetAwaiter().GetResult();
WriteObject(message);
}
catch (TaskCanceledException)
{
WriteWarning($"Timeout after {TimeoutSeconds} seconds — no message received.");
}
}
finally
{
if (handler != null)
{
Session.MessageReceived -= handler;
}
cts?.Dispose();
}
}
}
@@ -1,26 +0,0 @@
using System.Management.Automation;
using System.Management.Automation.Runspaces;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsLifecycle.Register, "MqttMessageHandler")]
public class RegisterMqttMessageHandlerCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter(Mandatory = true)]
public ScriptBlock? Action { get; set; }
protected override void ProcessRecord()
{
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.");
}
}
@@ -1,16 +0,0 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet(VerbsCommon.Remove, "MqttSession")]
public class RemoveMqttSessionCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
protected override void ProcessRecord()
{
Session.Dispose();
WriteObject("Session removed.");
}
}
@@ -1,55 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
using MQTTnet.Protocol;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet("Subscribe", "MqttTopic")]
[OutputType(typeof(MqttClientSubscribeResult))]
public class SubscribeMqttTopicCmdlet : PSCmdlet
{
[Parameter]
public bool NoLocal { get; set; }
[Parameter]
public int QoS { get; set; } = 0;
[Parameter]
public bool RetainAsPublished { get; set; }
[Parameter]
public MqttRetainHandling RetainHandling { get; set; } = MqttRetainHandling.SendAtSubscribe;
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter]
public uint SubscriptionIdentifier { get; set; }
[Parameter(Mandatory = true)]
public required string Topic { get; set; }
protected override void ProcessRecord()
{
var topicFilter = new MqttTopicFilterBuilder().WithTopic(Topic)
.WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
.WithNoLocal(NoLocal)
.WithRetainAsPublished(RetainAsPublished)
.WithRetainHandling(RetainHandling)
.Build();
var options = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter);
if (SubscriptionIdentifier > 0)
{
options.WithSubscriptionIdentifier(SubscriptionIdentifier);
}
var response = Session.GetClient().SubscribeAsync(options.Build()).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -1,23 +0,0 @@
using System.Management.Automation;
namespace MQTTnet.PowerShell.Cmdlets;
[Cmdlet("Unsubscribe", "MqttTopic")]
[OutputType(typeof(MqttClientUnsubscribeResult))]
public class UnsubscribeMqttTopicCmdlet : PSCmdlet
{
[Parameter(Mandatory = true, ValueFromPipeline = true)]
public required PsMqttSession Session { get; set; }
[Parameter(Mandatory = true)]
public required string Topic { get; set; }
protected override void ProcessRecord()
{
var options = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(Topic).Build();
var response = Session.GetClient().UnsubscribeAsync(options).GetAwaiter().GetResult();
WriteObject(response);
}
}
@@ -1,28 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<AssemblyName>MQTTnet.PowerShell</AssemblyName>
<LangVersion>default</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="PowerShellStandard.Library" Version="5.1.1">
<PrivateAssets>All</PrivateAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
</ItemGroup>
<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>
@@ -1,30 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Management.Automation;
using MQTTnet.Packets;
namespace MQTTnet.PowerShell;
public class PsMqttPublishResult
{
}
public class PsMqttConnectResult
{
}
public class PsMqttMessage
{
public required string? ContentType { get; init; }
public required string Payload { get; init; }
public int QoS { get; init; }
public required byte[] RawPayload { get; init; }
public string? ResponseTopic { get; init; }
public bool Retain { get; init; }
public required string Topic { get; init; }
public required List<MqttUserProperty>? UserProperties { get; init; }
}
@@ -1,50 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Buffers;
using System.Text;
namespace MQTTnet.PowerShell;
public sealed class PsMqttSession : IDisposable
{
readonly IMqttClient _client;
public PsMqttSession()
{
var factory = new MqttClientFactory();
_client = factory.CreateMqttClient();
_client.ApplicationMessageReceivedAsync += e =>
{
MessageReceived?.Invoke(
this,
new PsMqttMessage
{
Topic = e.ApplicationMessage.Topic,
Payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
RawPayload = e.ApplicationMessage.Payload.ToArray(),
UserProperties = e.ApplicationMessage.UserProperties,
QoS = (int)e.ApplicationMessage.QualityOfServiceLevel,
ContentType = e.ApplicationMessage.ContentType,
Retain = e.ApplicationMessage.Retain,
ResponseTopic = e.ApplicationMessage.ResponseTopic
});
return Task.CompletedTask;
};
}
public event EventHandler<PsMqttMessage>? MessageReceived;
public void Dispose()
{
_client.Dispose();
}
public IMqttClient GetClient()
{
return _client;
}
}
-15
View File
@@ -1,15 +0,0 @@
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
@@ -1,55 +0,0 @@
{
"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,7 +53,10 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value) public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value)
{ {
ArgumentNullException.ThrowIfNull(name); if (name == null)
{
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; set; } public string? ReasonString { get; init; }
public List<MqttUserProperty> UserProperties { get; set; } public List<MqttUserProperty>? UserProperties { get; init; }
public byte[] AuthenticationData { get; set; } public byte[]? AuthenticationData { get; init; }
} }
@@ -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(string clientId, string userName, EndPoint remoteEndPoint, MqttPacket packet, IDictionary sessionItems, CancellationToken cancellationToken) public InterceptingPacketEventArgs(CancellationToken cancellationToken, string clientId, string userName, EndPoint remoteEndPoint, MqttPacket packet, IDictionary sessionItems)
{ {
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, string clientId, string userName, IDictionary sessionItems, CancellationToken cancellationToken) public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems)
{ {
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
CancellationToken = cancellationToken; CancellationToken = cancellationToken;
@@ -30,6 +30,11 @@ 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>
@@ -40,15 +45,10 @@ 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(); public PublishResponse Response { get; } = new PublishResponse();
/// <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,6 +36,11 @@ 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>
@@ -56,7 +61,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(); public SubscribeResponse Response { get; } = new SubscribeResponse();
/// <summary> /// <summary>
/// Gets the current client session. /// Gets the current client session.
@@ -74,11 +79,6 @@ 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(string clientId, string userName, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties, CancellationToken cancellationToken) public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, string userName, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties)
{ {
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 List<MqttTcpServerListener>(); readonly List<MqttTcpServerListener> _listeners = new();
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, _localEndPoint, remoteEndPoint, clientCertificate); var tcpChannel = new MqttTcpChannel(stream, 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, senderId, senderUserName, senderSessionItems, cancellationToken); var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, senderId, senderUserName, senderSessionItems);
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> subscriptionsTopics) public void OnSubscriptionsAdded(MqttSession clientSession, List<string> topics)
{ {
_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 subscriptionsTopics) foreach (var topic in topics)
{ {
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, StringComparison.Ordinal); var senderIsReceiver = string.Equals(senderId, _session.Id);
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(_session.Id, _session.UserName, new MqttSessionStatus(_session), topicFilter, userProperties, cancellationToken); var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, _session.UserName, new MqttSessionStatus(_session), topicFilter, userProperties);
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/", StringComparison.InvariantCulture)) if (topicFilter.Topic.StartsWith("$share/"))
{ {
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(_session.Id, _session.UserName, _session.Items, topicFilter, userProperties, cancellationToken) var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.UserName, _session.Items, topicFilter, userProperties)
{ {
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?.Count > 0 || if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
!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(Id, UserName, RemoteEndPoint, packet, Session.Items, cancellationToken); var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, UserName, RemoteEndPoint, packet, Session.Items);
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(Id, UserName, RemoteEndPoint, currentPacket, Session.Items, cancellationToken); var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, UserName, RemoteEndPoint, currentPacket, Session.Items);
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,14 @@ using MQTTnet.Internal;
namespace MQTTnet.Server.Internal; namespace MQTTnet.Server.Internal;
public sealed class MqttRetainedMessagesManager : IDisposable public sealed class MqttRetainedMessagesManager
{ {
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096); readonly Dictionary<string, MqttApplicationMessage> _messages = new(4096);
readonly AsyncLock _storageAccessLock = new(); readonly AsyncLock _storageAccessLock = new();
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger) public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger)
{ {
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
@@ -23,46 +24,6 @@ public sealed class MqttRetainedMessagesManager : IDisposable
_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
@@ -118,7 +79,8 @@ public sealed class MqttRetainedMessagesManager : IDisposable
} }
else else
{ {
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload)) if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
!MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
{ {
_messages[applicationMessage.Topic] = applicationMessage; _messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true; saveIsRequired = true;
@@ -151,4 +113,39 @@ public sealed class MqttRetainedMessagesManager : IDisposable
_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);
}
}
} }
+40 -2
View File
@@ -1,19 +1,57 @@
<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,14 +2,12 @@
// 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 System.Security.Cryptography.X509Certificates;
using MQTTnet.Certificates; using MQTTnet.Certificates;
using System.Security.Cryptography.X509Certificates;
// 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,22 +47,12 @@ 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;
@@ -87,22 +77,12 @@ 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;
@@ -115,35 +95,6 @@ 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;
@@ -163,6 +114,12 @@ 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;
@@ -187,12 +144,6 @@ 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;
@@ -231,4 +182,33 @@ 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,30 +9,13 @@ namespace MQTTnet.Server;
public abstract class MqttServerTcpEndpointBaseOptions public abstract class MqttServerTcpEndpointBaseOptions
{ {
/// <summary> public bool IsEnabled { get; set; }
/// 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;
/// <summary> public int Port { get; set; }
/// 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.
@@ -40,16 +23,13 @@ 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>
/// This requires admin permissions on Linux. /// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all clients do support this feature and
/// will close the connection when receiving such packets. If such clients are connecting to this
/// server the flag must be set to _false_.
/// </summary> /// </summary>
public bool ReuseAddress { get; set; } public bool AllowPacketFragmentation { get; set; } = true;
/// <summary> /// <summary>
/// Gets or sets the TCP keep alive interval. /// Gets or sets the TCP keep alive interval.
@@ -68,4 +48,15 @@ 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; } = []; public List<MqttUserProperty>? UserProperties { get; set; } = [];
} }
+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 static async Task Run() public async Task Run()
{ {
try try
{ {
@@ -3,7 +3,19 @@
<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 sealed class MessageThroughputTest : IDisposable public class MessageThroughputTest
{ {
// Change these constants to suit // Change these constants to suit
const int NumPublishers = 5000; const int NumPublishers = 5000;
@@ -55,9 +55,9 @@ public sealed class MessageThroughputTest : IDisposable
await Setup(); await Setup();
await SubscribeToNoWildcardTopics(); await Subscribe_to_No_Wildcard_Topics();
await SubscribeToSingleWildcardTopics(); await Subscribe_to_Single_Wildcard_Topics();
await SubscribeToMultiWildcardTopics(); await Subscribe_to_Multi_Wildcard_Topics();
Console.WriteLine(); Console.WriteLine();
Console.WriteLine("End message throughput test"); Console.WriteLine("End message throughput test");
@@ -74,7 +74,7 @@ public sealed class MessageThroughputTest : IDisposable
public async Task Setup() public async Task Setup()
{ {
TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out _singleWildcardTopicsByPublisher, out _multiWildcardTopicsByPublisher); new 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 sealed class MessageThroughputTest : IDisposable
/// <summary> /// <summary>
/// Measure no-wildcard topic subscription message exchange performance /// Measure no-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task SubscribeToNoWildcardTopics() public Task Subscribe_to_No_Wildcard_Topics()
{ {
return ProcessMessages(_topicsByPublisher, "no wildcards"); return ProcessMessages(_topicsByPublisher, "no wildcards");
} }
@@ -160,7 +160,7 @@ public sealed class MessageThroughputTest : IDisposable
/// <summary> /// <summary>
/// Measure single-wildcard topic subscription message exchange performance /// Measure single-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task SubscribeToSingleWildcardTopics() public Task Subscribe_to_Single_Wildcard_Topics()
{ {
return ProcessMessages(_singleWildcardTopicsByPublisher, "single wildcard"); return ProcessMessages(_singleWildcardTopicsByPublisher, "single wildcard");
} }
@@ -168,7 +168,7 @@ public sealed class MessageThroughputTest : IDisposable
/// <summary> /// <summary>
/// Measure multi-wildcard topic subscription message exchange performance /// Measure multi-wildcard topic subscription message exchange performance
/// </summary> /// </summary>
public Task SubscribeToMultiWildcardTopics() public Task Subscribe_to_Multi_Wildcard_Topics()
{ {
return ProcessMessages(_multiWildcardTopicsByPublisher, "multi wildcard"); return ProcessMessages(_multiWildcardTopicsByPublisher, "multi wildcard");
} }
@@ -342,8 +342,4 @@ public sealed class MessageThroughputTest : IDisposable
Console.ResetColor(); Console.ResetColor();
} }
public void Dispose()
{
_cancellationTokenSource.Dispose();
}
} }
@@ -34,9 +34,7 @@ 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<MqttClientPublishResult> PublishSingleMessage(IMqttClient client, MqttApplicationMessage applicationMessage, ref int count) static Task 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(AsyncLockTest.Run); Task.Run(new 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 InvalidOperationException("Message invalid."); throw new Exception("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", StringComparison.InvariantCulture) && e.ClientId != "theAdmin") if (e.TopicFilter.Topic.StartsWith("admin/foo/bar") && e.ClientId != "theAdmin")
{ {
e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; e.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
} }
if (e.TopicFilter.Topic.StartsWith("the/secret/stuff", StringComparison.InvariantCulture) && e.ClientId != "Imperator") if (e.TopicFilter.Topic.StartsWith("the/secret/stuff") && 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 static void Generate( public 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
} }
} }
static void AddPublisherTopic(string publisherName, string topic, Dictionary<string, List<string>> topicsByPublisher) void AddPublisherTopic(string publisherName, string topic, Dictionary<string, List<string>> topicsByPublisher)
{ {
if (!topicsByPublisher.TryGetValue(publisherName, out var topicList)) List<string> 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,11 +119,9 @@ public class MqttConnectionContextTest
await Assert.ThrowsExactlyAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None)).ConfigureAwait(false); await Assert.ThrowsExactlyAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None)).ConfigureAwait(false);
} }
sealed class Startup 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);
} }
public static Task LongTestDelay() protected Task LongTestDelay()
{ {
return Task.Delay(TimeSpan.FromSeconds(1)); return Task.Delay(TimeSpan.FromSeconds(1));
} }
public static Task MediumTestDelay() protected 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;
} }
static async Task<MqttSubAckPacket> Subscribe(ILowLevelMqttClient client, string topic) 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<MqttCommunicationException>(exception); Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException));
} }
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);
} }
sealed class TestClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler class TestClientKerberosAuthenticationHandler : IMqttEnhancedAuthenticationHandler
{ {
public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs) public async Task HandleEnhancedAuthenticationAsync(MqttEnhancedAuthenticationEventArgs eventArgs)
{ {
@@ -6,7 +6,6 @@ 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;
@@ -123,8 +122,8 @@ public sealed class MqttClient_Tests : BaseTestClass
await Task.Delay(500); await Task.Delay(500);
Assert.IsNotNull(ex); Assert.IsNotNull(ex);
Assert.IsInstanceOfType<MqttCommunicationException>(ex); Assert.IsInstanceOfType(ex, typeof(MqttCommunicationException));
Assert.IsInstanceOfType<SocketException>(ex.InnerException); Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException));
} }
[TestMethod] [TestMethod]
@@ -246,8 +245,8 @@ public sealed class MqttClient_Tests : BaseTestClass
catch (Exception exception) catch (Exception exception)
{ {
Assert.IsNotNull(exception); Assert.IsNotNull(exception);
Assert.IsInstanceOfType<MqttCommunicationException>(exception); Assert.IsInstanceOfType(exception, typeof(MqttCommunicationException));
Assert.IsInstanceOfType<SocketException>(exception.InnerException); Assert.IsInstanceOfType(exception.InnerException, typeof(SocketException));
} }
} }
@@ -356,7 +355,7 @@ public sealed class MqttClient_Tests : BaseTestClass
async Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs) async Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
{ {
var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString(), CultureInfo.InvariantCulture); var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString());
await Task.Delay(value); await Task.Delay(value);
lock (receivedValues) lock (receivedValues)
@@ -370,7 +369,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(CultureInfo.InvariantCulture)); await client2.PublishStringAsync("x", i.ToString());
} }
await Task.Delay(5000); await Task.Delay(5000);
@@ -402,7 +401,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(CultureInfo.InvariantCulture), MqttQualityOfServiceLevel.ExactlyOnce); await client2.PublishStringAsync("x", i.ToString(), MqttQualityOfServiceLevel.ExactlyOnce);
} }
await Task.Delay(5000); await Task.Delay(5000);
@@ -416,7 +415,7 @@ public sealed class MqttClient_Tests : BaseTestClass
Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs) Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
{ {
var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString(), CultureInfo.InvariantCulture); var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString());
eventArgs.AutoAcknowledge = false; eventArgs.AutoAcknowledge = false;
Task.Delay(value).ContinueWith(_ => eventArgs.AcknowledgeAsync(CancellationToken.None)); Task.Delay(value).ContinueWith(_ => eventArgs.AcknowledgeAsync(CancellationToken.None));
@@ -1,24 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using 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));
} }
sealed class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
{ {
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context) public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
{ {
@@ -232,7 +232,7 @@ public sealed class Rpc_Tests : BaseTestClass
} }
} }
sealed class TestParametersTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy 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
sealed class ElementReference class ElementReference
{ {
public enum BufferElementType public enum BufferElementType
{ {
@@ -551,7 +551,7 @@ public sealed class MqttPacketSerialization_V3_Binary_Tests
} }
static void DeserializeAndCompare(MqttPacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) 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<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.IsInstanceOfType<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.IsInstanceOfType<MqttPublishPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType<MqttSubAckPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType<MqttPingRespPacket>(bus.DequeueItemAsync(CancellationToken.None).Result.Packet); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.AreEqual(0, bus.TotalItemsCount); Assert.AreEqual(0, bus.TotalItemsCount);
} }
+12 -2
View File
@@ -3,10 +3,19 @@
<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>disable</Nullable> <Nullable>enable</Nullable>
<NoWarn>$(NoWarn);CA1707</NoWarn>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@@ -14,6 +23,7 @@
<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,9 +29,7 @@ public sealed class MemoryMqttChannel : IMqttChannel
public EndPoint RemoteEndPoint { get; set; } public EndPoint RemoteEndPoint { get; set; }
public EndPoint LocalEndPoint { get; set; } public bool IsSecureConnection { get; } = false;
public bool IsSecureConnection { get; }
public X509Certificate2 ClientCertificate { get; set; } public X509Certificate2 ClientCertificate { get; set; }
@@ -60,6 +58,5 @@ 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