Compare commits

..
Author SHA1 Message Date
christian 37631730e0 Enable nullable 2025-10-20 11:29:24 +02:00
321 changed files with 1887 additions and 1445 deletions
+6 -55
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> </Project>
<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>
+3 -2
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}"
@@ -86,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)
{ {
+12 -12
View File
@@ -7,6 +7,7 @@
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
// ReSharper disable UnusedMember.Local // ReSharper disable UnusedMember.Local
using MQTTnet.Extensions.TopicTemplate;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Samples.Helpers; using MQTTnet.Samples.Helpers;
@@ -15,6 +16,8 @@ namespace MQTTnet.Samples.Client;
public static class Client_Subscribe_Samples public static class Client_Subscribe_Samples
{ {
static readonly MqttTopicTemplate sampleTemplate = new("mqttnet/samples/topic/{id}");
public static async Task Handle_Received_Application_Message() public static async Task Handle_Received_Application_Message()
{ {
/* /*
@@ -39,7 +42,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("mqttnet/samples/topic/2").Build(); var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "2")).Build();
await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -52,7 +55,7 @@ public static class Client_Subscribe_Samples
public static async Task Send_Responses() public static async Task Send_Responses()
{ {
/* /*
* This sample subscribes to a topic and sends a detailed response to the client. This requires at least QoS level 1 to work! * This sample subscribes to a topic and sends a response to the broker. This requires at least QoS level 1 to work!
*/ */
var mqttFactory = new MqttClientFactory(); var mqttFactory = new MqttClientFactory();
@@ -77,7 +80,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("topic/test").Build(); var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -103,10 +106,9 @@ public static class Client_Subscribe_Samples
// Create the subscribe options including several topics with different options. // Create the subscribe options including several topics with different options.
// It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic. // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(t => t.WithTopic("topic/1").WithAtLeastOnceQoS()) .WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
.WithTopicFilter(t => t.WithTopic("topic/2").WithAtMostOnceQoS()) .WithTopicTemplate(sampleTemplate.WithParameter("id", "2"), noLocal: true)
.WithTopicFilter(t => t.WithTopic("topic/3").WithNoLocal()) .WithTopicTemplate(sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
.WithTopicFilter(t => t.WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
.Build(); .Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -130,9 +132,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var topicFilter = mqttFactory.CreateTopicFilterBuilder().WithTopic("mqttnet/samples/topic/2").WithAtLeastOnceQoS(); var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter(topicFilter).Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -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,75 +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 MQTTnet.Extensions.TopicTemplate;
using MQTTnet.Protocol;
using MQTTnet.Samples.Helpers;
// ReSharper disable UnusedType.Global
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
// ReSharper disable UnusedMember.Local
namespace MQTTnet.Samples.Client;
public static class Client_Subscribe_TopicTemplate_Samples
{
static readonly MqttTopicTemplate SampleTemplate = new("mqttnet/samples/topic/{id}");
public static async Task Subscribe_Multiple_Topics_With_Template()
{
/*
* This sample subscribes to several topics in a single request.
*
* This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
*/
var mqttFactory = new MqttClientFactory();
using var mqttClient = mqttFactory.CreateMqttClient();
var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
// Create the subscribe options including several topics with different options.
// It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicTemplate(SampleTemplate.WithParameter("id", "1"))
.WithTopicTemplate(SampleTemplate.WithParameter("id", "2"), noLocal: true)
.WithTopicTemplate(SampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
.Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
Console.WriteLine("MQTT client subscribed to topics.");
// The response contains additional data sent by the server after subscribing.
response.DumpToConsole();
}
public static async Task Subscribe_Topic_With_Template()
{
/*
* This sample subscribes to a topic.
*
* This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
*/
var mqttFactory = new MqttClientFactory();
using var mqttClient = mqttFactory.CreateMqttClient();
var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(SampleTemplate.WithParameter("id", "1")).Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
Console.WriteLine("MQTT client subscribed to topic.");
// The response contains additional data sent by the server after subscribing.
response.DumpToConsole();
}
}
+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());
+5 -7
View File
@@ -8,19 +8,17 @@ namespace MQTTnet.Samples.Helpers;
internal static class ObjectExtensions internal static class ObjectExtensions
{ {
static readonly JsonSerializerOptions SerializerOptions = new()
{
WriteIndented = true
};
public static TObject DumpToConsole<TObject>(this TObject @object) public static TObject DumpToConsole<TObject>(this TObject @object)
{ {
var output = "NULL"; var output = "NULL";
if (@object != null) if (@object != null)
{ {
output = JsonSerializer.Serialize(@object, 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}");
return @object; return @object;
} }
+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>
+3 -4
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();
@@ -59,7 +58,7 @@ try
{ {
if (selectedSampleMethod.Invoke(null, null) is Task task) if (selectedSampleMethod.Invoke(null, null) is Task task)
{ {
await task; await task;
} }
} }
catch (Exception exception) catch (Exception exception)
+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();
} }
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using MQTTnet.Server; using MQTTnet.Server;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Server; using MQTTnet.Server;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Routing; using Microsoft.AspNetCore.Routing;
@@ -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"/>
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Formatter; using MQTTnet.Formatter;
@@ -12,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)
{ {
@@ -2,10 +2,13 @@
// 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.Buffers; using System.Buffers;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Net; using System.Net;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Connections.Features; using Microsoft.AspNetCore.Http.Connections.Features;
using Microsoft.AspNetCore.Http.Features; using Microsoft.AspNetCore.Http.Features;
@@ -58,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
@@ -80,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)
@@ -250,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);
} }
} }
@@ -2,6 +2,8 @@
// 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.Threading.Tasks;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Connections.Features; using Microsoft.AspNetCore.Connections.Features;
using MQTTnet.Adapter; using MQTTnet.Adapter;
@@ -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 Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server; using MQTTnet.Server;
@@ -2,6 +2,9 @@
// 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.Linq;
using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Http;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
@@ -26,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"));
} }
} }
@@ -2,8 +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.Net; using System.Net;
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Http;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Buffers; using System.Buffers;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using MQTTnet.Adapter; using MQTTnet.Adapter;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
@@ -2,10 +2,13 @@
// 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.Diagnostics; using System.Diagnostics;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Net.Sockets; using System.Net.Sockets;
using System.Runtime.CompilerServices; using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.AspNetCore; namespace MQTTnet.AspNetCore;
@@ -2,9 +2,13 @@
// 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.IO;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Net; using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Features; using Microsoft.AspNetCore.Http.Features;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
@@ -77,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");
} }
@@ -214,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
+2 -6
View File
@@ -2,12 +2,13 @@
// 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.IO.Pipelines; using System.IO.Pipelines;
using System.Net.Sockets; 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();
@@ -32,9 +33,4 @@ public sealed class SocketReceiver : IDisposable
return _awaitable; return _awaitable;
} }
public void Dispose()
{
_eventArgs.Dispose();
}
} }
+3 -6
View File
@@ -2,7 +2,9 @@
// 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.Buffers; using System.Buffers;
using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Net.Sockets; using System.Net.Sockets;
@@ -10,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();
@@ -26,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)
@@ -1 +1,3 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
[assembly: DoNotParallelize] [assembly: DoNotParallelize]
@@ -2,6 +2,9 @@
// 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.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Internal; using MQTTnet.Internal;
@@ -40,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})!");
} }
} }
@@ -2,6 +2,9 @@
// 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.IO;
using System.Threading;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
@@ -12,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;
@@ -83,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>
@@ -1,5 +1,6 @@
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using System;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
@@ -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}");
} }
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Text; using System.Text;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Formatter; using MQTTnet.Formatter;
@@ -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();
}
} }
+4 -1
View File
@@ -2,6 +2,9 @@
// 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.Linq;
using BenchmarkDotNet.Running; using BenchmarkDotNet.Running;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
@@ -89,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)
{ {
@@ -4,14 +4,16 @@ using MQTTnet.AspNetCore;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using System.Buffers; using System.Buffers;
using System.IO;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Threading.Tasks;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
[SimpleJob(RuntimeMoniker.Net60)] [SimpleJob(RuntimeMoniker.Net60)]
[RPlotExporter, RankColumn] [RPlotExporter, RankColumn]
[MemoryDiagnoser] [MemoryDiagnoser]
public sealed class ReaderExtensionsBenchmark : IDisposable, IAsyncDisposable public class ReaderExtensionsBenchmark
{ {
MqttPacketFormatterAdapter _mqttPacketFormatter; MqttPacketFormatterAdapter _mqttPacketFormatter;
MemoryStream _stream; MemoryStream _stream;
@@ -81,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,6 +1,7 @@
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups; using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
@@ -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
@@ -4,7 +4,10 @@
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using MQTTnet.Packets; using MQTTnet.Packets;
using System;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Channel; using MQTTnet.Channel;
using MQTTnet.Formatter; using MQTTnet.Formatter;
@@ -23,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]
@@ -40,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++)
{ {
@@ -50,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());
@@ -63,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;
@@ -76,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; }
@@ -5,6 +5,7 @@
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups; using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
@@ -2,6 +2,8 @@
// 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.Collections.Generic;
using System.Linq;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using MQTTnet.Server; using MQTTnet.Server;
@@ -5,6 +5,8 @@
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Net; using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Jobs;
using MQTTnet.AspNetCore; using MQTTnet.AspNetCore;
@@ -1,3 +1,6 @@
using System;
using System.Collections.Generic;
namespace MQTTnet.Benchmarks; namespace MQTTnet.Benchmarks;
public class TopicGenerator public class TopicGenerator
@@ -2,6 +2,8 @@
// 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.Collections.Generic;
using System.Linq;
using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Attributes;
using MQTTnet.Server; using MQTTnet.Server;
@@ -2,6 +2,8 @@
// 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;
namespace MQTTnet.Extensions.Rpc; namespace MQTTnet.Extensions.Rpc;
public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
@@ -10,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 #.");
} }
@@ -2,13 +2,22 @@
// 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 MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc; 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>
@@ -2,6 +2,8 @@
// 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;
namespace MQTTnet.Extensions.Rpc; namespace MQTTnet.Extensions.Rpc;
public static class MqttFactoryExtensions public static class MqttFactoryExtensions
+22 -9
View File
@@ -2,8 +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;
using System.Buffers; using System.Buffers;
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
@@ -16,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)
{ {
@@ -38,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
@@ -56,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);
@@ -78,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);
} }
@@ -99,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);
} }
@@ -2,16 +2,19 @@
// 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.Text; using System.Text;
using System.Threading.Tasks;
using MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc; 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);
@@ -2,6 +2,8 @@
// 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;
namespace MQTTnet.Extensions.Rpc; namespace MQTTnet.Extensions.Rpc;
public sealed class MqttRpcClientOptionsBuilder public sealed class MqttRpcClientOptionsBuilder
@@ -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; }
} }
@@ -2,14 +2,20 @@
// 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 MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc; 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;
@@ -20,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>
@@ -2,7 +2,11 @@
// 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.Linq;
using System.Text.RegularExpressions; using System.Text.RegularExpressions;
using System.Threading;
using MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Extensions.TopicTemplate; namespace MQTTnet.Extensions.TopicTemplate;
@@ -80,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;
} }
@@ -157,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;
} }
@@ -2,6 +2,8 @@
// 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.Linq;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
@@ -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; } = [];
} }
+4 -1
View File
@@ -2,13 +2,16 @@
// 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.Threading;
using System.Threading.Tasks;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
namespace MQTTnet.TestApp; namespace MQTTnet.TestApp;
public sealed class AsyncLockTest public sealed class AsyncLockTest
{ {
public static async Task Run() public async Task Run()
{ {
try try
{ {
+2
View File
@@ -2,6 +2,8 @@
// 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.Threading.Tasks;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
namespace MQTTnet.TestApp; namespace MQTTnet.TestApp;
+2
View File
@@ -2,8 +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.Net; using System.Net;
using System.Text; using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
@@ -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>
+13 -12
View File
@@ -1,5 +1,10 @@
using MQTTnet.Server; using MQTTnet.Server;
using System.Collections.Generic;
using System;
using System.Diagnostics; using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using System.Linq;
using MQTTnet.Internal; using MQTTnet.Internal;
namespace MQTTnet.TestApp; namespace MQTTnet.TestApp;
@@ -9,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;
@@ -50,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");
@@ -69,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();
@@ -147,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");
} }
@@ -155,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");
} }
@@ -163,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");
} }
@@ -337,8 +342,4 @@ public sealed class MessageThroughputTest : IDisposable
Console.ResetColor(); Console.ResetColor();
} }
public void Dispose()
{
_cancellationTokenSource.Dispose();
}
} }
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Text; using System.Text;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
@@ -33,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());
+5 -1
View File
@@ -2,9 +2,13 @@
// 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.Diagnostics; using System.Diagnostics;
using System.Linq;
using System.Net; using System.Net;
using System.Text; using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
@@ -200,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));
+5 -1
View File
@@ -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.Threading;
using System.Threading.Tasks;
namespace MQTTnet.TestApp; namespace MQTTnet.TestApp;
public static class Program public static class Program
@@ -81,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);
+4 -1
View File
@@ -2,7 +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.Security.Authentication; using System.Security.Authentication;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
@@ -154,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,6 +2,8 @@
// 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.Threading;
using System.Threading.Tasks;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server; using MQTTnet.Server;
+6 -2
View File
@@ -2,7 +2,11 @@
// 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.IO;
using System.Text; using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.Logger;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
@@ -125,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;
+6 -4
View File
@@ -1,10 +1,12 @@
using System;
using System.Collections.Generic;
using MQTTnet.Extensions.TopicTemplate; using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.TestApp; 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,
@@ -82,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);
} }
} }
@@ -2,6 +2,8 @@
// 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.Threading.Tasks;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.AspNetCore; using MQTTnet.AspNetCore;
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Buffers; using System.Buffers;
namespace MQTTnet.Tests.ASP.Mockups; namespace MQTTnet.Tests.ASP.Mockups;
@@ -2,8 +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;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.AspNetCore; using MQTTnet.AspNetCore;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
@@ -115,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
{ {
} }
} }
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System.Buffers; using System.Buffers;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.AspNetCore; using MQTTnet.AspNetCore;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Packets; using MQTTnet.Packets;

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