Compare commits

...
Author SHA1 Message Date
christian b477ddbe60 Start using readonly memory 2025-10-22 11:35:56 +02:00
Christian d74d83453f Remove unused usings (#2217) 2025-10-21 13:10:09 +02:00
Christian 99e70fdbbb Update samples (#2216)
Separate the topic template samples from the other samples
2025-10-21 12:33:20 +02:00
Christian 535a4e010c Add validations of variable byte integers (#2214)
* Add validations and add docs

* Update release notes

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

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

* Adopt new changes in master

---------

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

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

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