diff --git a/.editorconfig b/.editorconfig
index da3f0a4de..46144365e 100644
--- a/.editorconfig
+++ b/.editorconfig
@@ -21,14 +21,14 @@ dotnet_naming_rule.static_readonly_rule.symbols = static_readonly_symbols
dotnet_naming_style.lower_camel_case_style.capitalization = camel_case
dotnet_naming_style.lower_camel_case_style.required_prefix = _
dotnet_naming_style.upper_camel_case_style.capitalization = pascal_case
-dotnet_naming_symbols.constants_symbols.applicable_accessibilities = public,internal,protected,protected_internal,private_protected
+dotnet_naming_symbols.constants_symbols.applicable_accessibilities = public, internal, protected, protected_internal, private_protected
dotnet_naming_symbols.constants_symbols.applicable_kinds = field
dotnet_naming_symbols.constants_symbols.required_modifiers = const
-dotnet_naming_symbols.public_fields_symbols.applicable_accessibilities = public,internal,protected,protected_internal,private_protected
+dotnet_naming_symbols.public_fields_symbols.applicable_accessibilities = public, internal, protected, protected_internal, private_protected
dotnet_naming_symbols.public_fields_symbols.applicable_kinds = field
-dotnet_naming_symbols.static_readonly_symbols.applicable_accessibilities = public,internal,protected,protected_internal,private_protected
+dotnet_naming_symbols.static_readonly_symbols.applicable_accessibilities = public, internal, protected, protected_internal, private_protected
dotnet_naming_symbols.static_readonly_symbols.applicable_kinds = field
-dotnet_naming_symbols.static_readonly_symbols.required_modifiers = static,readonly
+dotnet_naming_symbols.static_readonly_symbols.required_modifiers = static, readonly
dotnet_style_parentheses_in_arithmetic_binary_operators = never_if_unnecessary:none
dotnet_style_parentheses_in_other_binary_operators = never_if_unnecessary:none
dotnet_style_parentheses_in_relational_binary_operators = never_if_unnecessary:none
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0a826e9a4..23bbcf213 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -32,13 +32,13 @@ jobs:
- name: Build solution
run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}/certificate.snk
- - name: Collect nuget Packages
+ - name: Collect nuget packages
uses: actions/upload-artifact@v2
with:
name: nuget Packages
path: |
- **\*.nupkg
- **\*.snupkg
+ **/*.nupkg
+ **/*.snupkg
- name: Execute tests
run: dotnet test --no-restore --framework net8.0 Source/MQTTnet.Tests/MQTTnet.Tests.csproj
diff --git a/README.md b/README.md
index 90c25b1ea..70bfe37a7 100644
--- a/README.md
+++ b/README.md
@@ -12,7 +12,9 @@
# MQTTnet
-MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework version and CPU architecture.
+MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (
+broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
+version and CPU architecture.
## Features
@@ -28,13 +30,15 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* Unit tested (~636 tests)
* No external dependencies
-\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
+\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP
+channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
### Client
* Communication via TCP (+TLS) or WS (WebSocket) supported
* Included core _LowLevelMqttClient_ with low level functionality
-* Also included _ManagedMqttClient_ which maintains the connection and subscriptions automatically. Also application messages are queued and re-scheduled for higher QoS levels automatically.
+* Also included _ManagedMqttClient_ which maintains the connection and subscriptions automatically. Also application
+ messages are queued and re-scheduled for higher QoS levels automatically.
* Rx support (via another project)
* Compatible with Microsoft Azure IoT Hub
@@ -52,9 +56,11 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
## Getting Started
-MQTTnet is delivered via NuGet package manager. You can find the packages here: https://www.nuget.org/packages/MQTTnet/
+MQTTnet is delivered via NuGet package manager. You can find the packages
+here: https://www.nuget.org/packages/MQTTnet/
Use these command in the Package Manager console to install MQTTnet manually:
+
```
Install-Package MQTTnet
```
@@ -68,7 +74,8 @@ Samples for using MQTTnet are part of this repository. For starters these sample
## Code of Conduct
-This project has adopted the code of conduct defined by the Contributor Covenant to clarify expected behavior in our community.
+This project has adopted the code of conduct defined by the Contributor Covenant to clarify expected behavior in our
+community.
For more information see the [.NET Foundation Code of Conduct](https://dotnetfoundation.org/code-of-conduct).
## .NET Foundation
diff --git a/Samples/MQTTnet.Samples.csproj b/Samples/MQTTnet.Samples.csproj
index bfc3a9408..90640f8b6 100644
--- a/Samples/MQTTnet.Samples.csproj
+++ b/Samples/MQTTnet.Samples.csproj
@@ -14,11 +14,11 @@
-
-
-
-
-
+
+
+
+
+
diff --git a/Samples/Server/Server_Diagnostics_Samples.cs b/Samples/Server/Server_Diagnostics_Samples.cs
index b72fec3d0..07bc8e0b6 100644
--- a/Samples/Server/Server_Diagnostics_Samples.cs
+++ b/Samples/Server/Server_Diagnostics_Samples.cs
@@ -13,7 +13,6 @@ namespace MQTTnet.Samples.Server;
// ReSharper disable InconsistentNaming
// ReSharper disable EmptyConstructor
// ReSharper disable MemberCanBeMadeStatic.Local
-
public static class Server_Diagnostics_Samples
{
public static async Task Get_Notified_When_Client_Received_Message()
diff --git a/Samples/Server/Server_Simple_Samples.cs b/Samples/Server/Server_Simple_Samples.cs
index 4abededb0..bfb86f3b0 100644
--- a/Samples/Server/Server_Simple_Samples.cs
+++ b/Samples/Server/Server_Simple_Samples.cs
@@ -216,4 +216,4 @@ public void Publish(MqttNetLogLevel logLevel, string source, string message, obj
}
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj b/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj
index a624c945a..f977d5cd0 100644
--- a/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj
+++ b/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj
@@ -12,13 +12,13 @@
-
-
-
+
+
+
-
+
diff --git a/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
index d7bc3e3d7..f98983da9 100644
--- a/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
+++ b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
@@ -7,15 +7,17 @@
using Microsoft.Extensions.DependencyInjection;
using MQTTnet.Server;
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
+
+public static class ApplicationBuilderExtensions
{
- public static class ApplicationBuilderExtensions
+ [Obsolete(
+ "This class is obsolete and will be removed in a future version. The recommended alternative is to use MapMqtt inside Microsoft.AspNetCore.Builder.UseEndpoints(...).")]
+ public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app, string path = "/mqtt")
{
- [Obsolete("This class is obsolete and will be removed in a future version. The recommended alternative is to use MapMqtt inside Microsoft.AspNetCore.Builder.UseEndpoints(...).")]
- public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app, string path = "/mqtt")
- {
- app.UseWebSockets();
- app.Use(async (context, next) =>
+ app.UseWebSockets();
+ app.Use(
+ async (context, next) =>
{
if (!context.WebSockets.IsWebSocketRequest || context.Request.Path != path)
{
@@ -37,16 +39,15 @@ public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app,
}
});
- return app;
- }
+ return app;
+ }
- public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action configure)
- {
- var server = app.ApplicationServices.GetRequiredService();
+ public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action configure)
+ {
+ var server = app.ApplicationServices.GetRequiredService();
- configure(server);
+ configure(server);
- return app;
- }
+ return app;
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs
index a95e0017a..394483959 100644
--- a/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs
+++ b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs
@@ -2,18 +2,17 @@
// 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.Server;
using System;
+using MQTTnet.Server;
+
+namespace MQTTnet.AspNetCore;
-namespace MQTTnet.AspNetCore
+public sealed class AspNetMqttServerOptionsBuilder : MqttServerOptionsBuilder
{
- public sealed class AspNetMqttServerOptionsBuilder : MqttServerOptionsBuilder
+ public AspNetMqttServerOptionsBuilder(IServiceProvider serviceProvider)
{
- public AspNetMqttServerOptionsBuilder(IServiceProvider serviceProvider)
- {
- ServiceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
- }
-
- public IServiceProvider ServiceProvider { get; }
+ ServiceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
}
-}
+
+ public IServiceProvider ServiceProvider { get; }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
index dad8b6624..cf8e58740 100644
--- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
+++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
@@ -42,16 +42,16 @@
-
-
+
+
-
-
+
+
diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
index 398e2902d..4e1138a40 100644
--- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
+++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
@@ -2,58 +2,56 @@
// 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;
-using MQTTnet.Server;
-using System;
-using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
-using MQTTnet.Server.Adapter;
+using MQTTnet.Server;
+
+namespace MQTTnet.AspNetCore;
-namespace MQTTnet.AspNetCore
+public sealed class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter
{
- public sealed class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter
- {
- MqttServerOptions _serverOptions;
+ MqttServerOptions _serverOptions;
+
+ public Func ClientHandler { get; set; }
- public Func ClientHandler { get; set; }
+ public void Dispose()
+ {
+ }
- public override async Task OnConnectedAsync(ConnectionContext connection)
+ public override async Task OnConnectedAsync(ConnectionContext connection)
+ {
+ // required for websocket transport to work
+ var transferFormatFeature = connection.Features.Get();
+ if (transferFormatFeature != null)
{
- // required for websocket transport to work
- var transferFormatFeature = connection.Features.Get();
- if (transferFormatFeature != null)
- {
- transferFormatFeature.ActiveFormat = TransferFormat.Binary;
- }
+ transferFormatFeature.ActiveFormat = TransferFormat.Binary;
+ }
- var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax));
- using (var adapter = new MqttConnectionContext(formatter, connection))
+ var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax));
+ using (var adapter = new MqttConnectionContext(formatter, connection))
+ {
+ var clientHandler = ClientHandler;
+ if (clientHandler != null)
{
- var clientHandler = ClientHandler;
- if (clientHandler != null)
- {
- await clientHandler(adapter).ConfigureAwait(false);
- }
+ await clientHandler(adapter).ConfigureAwait(false);
}
}
+ }
- public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
- {
- _serverOptions = options;
-
- return Task.CompletedTask;
- }
+ public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
+ {
+ _serverOptions = options;
- public Task StopAsync()
- {
- return Task.CompletedTask;
- }
+ return Task.CompletedTask;
+ }
- public void Dispose()
- {
- }
+ public Task StopAsync()
+ {
+ return Task.CompletedTask;
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs
index e98be27c2..cb6e5fb5c 100644
--- a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs
+++ b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs
@@ -7,36 +7,33 @@
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
-using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Server;
-using MQTTnet.Server.Adapter;
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
+
+public sealed class MqttHostedServer : MqttServer, IHostedService
{
- public sealed class MqttHostedServer : MqttServer, IHostedService
- {
- readonly MqttServerFactory _mqttServerFactory;
+ readonly MqttServerFactory _mqttServerFactory;
- public MqttHostedServer(MqttServerFactory mqttServerFactory, MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) : base(
- options,
- adapters,
- logger)
- {
- _mqttServerFactory = mqttServerFactory ?? throw new ArgumentNullException(nameof(mqttServerFactory));
- }
+ public MqttHostedServer(MqttServerFactory mqttServerFactory, MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) : base(
+ options,
+ adapters,
+ logger)
+ {
+ _mqttServerFactory = mqttServerFactory ?? throw new ArgumentNullException(nameof(mqttServerFactory));
+ }
- public async Task StartAsync(CancellationToken cancellationToken)
- {
- // The yield makes sure that the hosted service is considered up and running.
- await Task.Yield();
+ public async Task StartAsync(CancellationToken cancellationToken)
+ {
+ // The yield makes sure that the hosted service is considered up and running.
+ await Task.Yield();
- _ = StartAsync();
- }
+ _ = StartAsync();
+ }
- public Task StopAsync(CancellationToken cancellationToken)
- {
- return StopAsync(_mqttServerFactory.CreateMqttServerStopOptionsBuilder().Build());
- }
+ public Task StopAsync(CancellationToken cancellationToken)
+ {
+ return StopAsync(_mqttServerFactory.CreateMqttServerStopOptionsBuilder().Build());
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
index 2d01a9fcf..0d9522949 100644
--- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
+++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
@@ -11,61 +11,59 @@
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Server;
-using MQTTnet.Server.Adapter;
-namespace MQTTnet.AspNetCore
+namespace MQTTnet.AspNetCore;
+
+public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter
{
- public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter
- {
- IMqttNetLogger _logger = MqttNetNullLogger.Instance;
+ IMqttNetLogger _logger = MqttNetNullLogger.Instance;
+
+ public Func ClientHandler { get; set; }
- public Func ClientHandler { get; set; }
+ public void Dispose()
+ {
+ }
- public void Dispose()
+ public async Task RunWebSocketConnectionAsync(WebSocket webSocket, HttpContext httpContext)
+ {
+ if (webSocket == null)
{
+ throw new ArgumentNullException(nameof(webSocket));
}
- public async Task RunWebSocketConnectionAsync(WebSocket webSocket, HttpContext httpContext)
- {
- if (webSocket == null)
- {
- throw new ArgumentNullException(nameof(webSocket));
- }
+ var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}";
- var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}";
+ var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false);
+ try
+ {
+ var isSecureConnection = clientCertificate != null;
- var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false);
- try
+ var clientHandler = ClientHandler;
+ if (clientHandler != null)
{
- var isSecureConnection = clientCertificate != null;
+ var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535));
+ var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection, clientCertificate);
- var clientHandler = ClientHandler;
- if (clientHandler != null)
+ using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger))
{
- var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535));
- var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection, clientCertificate);
-
- using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger))
- {
- await clientHandler(channelAdapter).ConfigureAwait(false);
- }
+ await clientHandler(channelAdapter).ConfigureAwait(false);
}
}
- finally
- {
- clientCertificate?.Dispose();
- }
}
-
- public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
+ finally
{
- _logger = logger ?? throw new ArgumentNullException(nameof(logger));
- return Task.CompletedTask;
+ clientCertificate?.Dispose();
}
+ }
- public Task StopAsync()
- {
- return Task.CompletedTask;
- }
+ public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
+ {
+ _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+ return Task.CompletedTask;
+ }
+
+ public Task StopAsync()
+ {
+ return Task.CompletedTask;
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
index 17ca79272..27b55161e 100644
--- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
+++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
@@ -8,7 +8,7 @@
using Microsoft.Extensions.Hosting;
using MQTTnet.Diagnostics;
using MQTTnet.Server;
-using MQTTnet.Server.Adapter;
+using MQTTnet.Server.Internal.Adapter;
namespace MQTTnet.AspNetCore;
diff --git a/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
index 20fdf0e0a..e43adc753 100644
--- a/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
+++ b/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
@@ -12,15 +12,23 @@
1591;NETSDK1138
+
+ 1591;NETSDK1138
+
+
+
+ 1591;NETSDK1138
+
+
-
+
-
-
-
-
+
+
+
+
\ No newline at end of file
diff --git a/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs b/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs
index b32fc7922..6505abcf7 100644
--- a/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs
@@ -2,233 +2,217 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
-using BenchmarkDotNet.Attributes;
-using MQTTnet.Client;
-using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
+using BenchmarkDotNet.Attributes;
+using MQTTnet.Client;
+using MQTTnet.Packets;
+using MQTTnet.Server;
-namespace MQTTnet.Benchmarks
-{
- [MemoryDiagnoser]
- public class MessageDeliveryBenchmark : BaseBenchmark
- {
- List _topicPublishMessages;
+namespace MQTTnet.Benchmarks;
- [Params(1, 5)]
- public int NumTopicsPerPublisher;
+[MemoryDiagnoser]
+public class MessageDeliveryBenchmark : BaseBenchmark
+{
+ List _allSubscribedTopics; // Keep track of the subset of topics that are subscribed
+ CancellationTokenSource _cancellationTokenSource;
- [Params(1000, 10000)]
- public int NumPublishers;
+ object _lockMsgCount;
+ int _messagesExpectedCount;
+ int _messagesReceivedCount;
+ Dictionary _mqttPublisherClientsByPublisherName;
- [Params(10)]
- public int NumSubscribers;
+ MqttServer _mqttServer;
+ List _mqttSubscriberClients;
+ Dictionary _publisherByTopic;
+ List _topicPublishMessages;
- [Params(5, 10, 20, 50)]
- public int NumSubscribedTopicsPerSubscriber;
+ Dictionary> _topicsByPublisher;
- object _lockMsgCount;
- int _messagesReceivedCount;
- int _messagesExpectedCount;
- CancellationTokenSource _cancellationTokenSource;
+ [Params(1000, 10000)] public int NumPublishers;
- MqttServer _mqttServer;
- List _mqttSubscriberClients;
- Dictionary _mqttPublisherClientsByPublisherName;
+ [Params(5, 10, 20, 50)] public int NumSubscribedTopicsPerSubscriber;
- Dictionary> _topicsByPublisher;
- Dictionary _publisherByTopic;
- List _allSubscribedTopics; // Keep track of the subset of topics that are subscribed
+ [Params(10)] public int NumSubscribers;
+ [Params(1, 5)] public int NumTopicsPerPublisher;
- [GlobalSetup]
- public void Setup()
+ [GlobalCleanup]
+ public void Cleanup()
+ {
+ foreach (var mp in _mqttPublisherClientsByPublisherName)
{
- _lockMsgCount = new object();
+ var mqttPublisherClient = mp.Value;
+ mqttPublisherClient.DisconnectAsync().GetAwaiter().GetResult();
+ mqttPublisherClient.Dispose();
+ }
- Dictionary> singleWildcardTopicsByPublisher;
- Dictionary> multiWildcardTopicsByPublisher;
+ _mqttPublisherClientsByPublisherName.Clear();
- TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
+ foreach (var mqttSubscriber in _mqttSubscriberClients)
+ {
+ mqttSubscriber.DisconnectAsync().GetAwaiter().GetResult();
+ mqttSubscriber.Dispose();
+ }
- var topics = _topicsByPublisher.First().Value;
- _topicPublishMessages = new List();
- // Prepare messages, same for each publisher
- foreach (var topic in topics)
- {
- var message = new MqttApplicationMessageBuilder()
- .WithTopic(topic)
- .WithPayload(new byte[] { 1, 2, 3, 4, 5, 6, 7, 8 })
- .Build();
- _topicPublishMessages.Add(message);
- }
+ _mqttSubscriberClients.Clear();
- // Create server
- var serverFactory = new MqttServerFactory();
- var clientFactory = new MqttClientFactory();
- var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
- _mqttServer = serverFactory.CreateMqttServer(serverOptions);
- _mqttServer.StartAsync().GetAwaiter().GetResult();
+ _mqttServer.StopAsync().GetAwaiter().GetResult();
+ _mqttServer.Dispose();
+ _mqttServer = null;
+ }
- // Create publisher clients
- _mqttPublisherClientsByPublisherName = new Dictionary();
- foreach (var pt in _topicsByPublisher)
- {
- var publisherName = pt.Key;
- var mqttClient = clientFactory.CreateMqttClient();
- var publisherOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost")
- .WithClientId(publisherName)
- .WithKeepAlivePeriod(TimeSpan.FromSeconds(30))
- .Build();
- mqttClient.ConnectAsync(publisherOptions).GetAwaiter().GetResult();
- _mqttPublisherClientsByPublisherName.Add(publisherName, mqttClient);
- }
+ ///
+ /// Publish messages and wait for messages sent to subscribers
+ ///
+ [Benchmark]
+ public void DeliverMessages()
+ {
+ // There should be one message received per publish for each subscribed topic
+ _messagesExpectedCount = NumSubscribedTopicsPerSubscriber * NumSubscribers;
- // Create subscriber clients
- _mqttSubscriberClients = new List();
- for (var i = 0; i < NumSubscribers; i++)
- {
- var mqttSubscriberClient = clientFactory.CreateMqttClient();
- _mqttSubscriberClients.Add(mqttSubscriberClient);
-
- var subscriberOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost")
- .WithClientId("subscriber" + i)
- .Build();
- mqttSubscriberClient.ApplicationMessageReceivedAsync += r =>
- {
- // count messages and signal cancellation when expected message count is reached
- lock (_lockMsgCount)
- {
- ++_messagesReceivedCount;
- if (_messagesReceivedCount == _messagesExpectedCount)
- {
- _cancellationTokenSource.Cancel();
- }
- }
- return Task.CompletedTask;
- };
- mqttSubscriberClient.ConnectAsync(subscriberOptions).GetAwaiter().GetResult();
- }
+ // Loop for a while and exchange messages
+ _messagesReceivedCount = 0;
- List allTopics = new List();
- _publisherByTopic = new Dictionary();
- foreach (var t in _topicsByPublisher)
- {
- foreach (var topic in t.Value)
- {
- _publisherByTopic.Add(topic, t.Key);
- allTopics.Add(topic);
- }
- }
+ _cancellationTokenSource = new CancellationTokenSource();
- // Subscribe to NumSubscribedTopics topics spread across all topics
- _allSubscribedTopics = new List();
+ // same payload for all messages
+ var payload = new byte[] { 1, 2, 3, 4 };
- var totalNumTopics = NumPublishers * NumTopicsPerPublisher;
- int topicIndexStep = totalNumTopics / (NumSubscribedTopicsPerSubscriber * NumSubscribers);
- if (topicIndexStep * NumSubscribedTopicsPerSubscriber * NumSubscribers != totalNumTopics)
- {
- throw new System.Exception(
- String.Format("The total number of topics must be divisible by the number of subscribed topics across all subscribers. Total number of topics: {0}, topic step: {1}",
- totalNumTopics, topicIndexStep
- ));
- }
+ var tasks = new List();
- var topicIndex = 0;
- foreach (var mqttSubscriber in _mqttSubscriberClients)
- {
- for (var i = 0; i < NumSubscribedTopicsPerSubscriber; ++i, topicIndex += topicIndexStep)
- {
- var topic = allTopics[topicIndex];
- _allSubscribedTopics.Add(topic);
- var subOptions = new Client.MqttClientSubscribeOptionsBuilder().WithTopicFilter(
- new Packets.MqttTopicFilter() { Topic = topic })
- .Build();
- mqttSubscriber.SubscribeAsync(subOptions).GetAwaiter().GetResult();
- }
- }
+ // publish a message for each subscribed topic
+ foreach (var topic in _allSubscribedTopics)
+ {
+ var message = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).Build();
+ // pick the correct publisher
+ var publisherName = _publisherByTopic[topic];
+ var publisherClient = _mqttPublisherClientsByPublisherName[publisherName];
+ _ = publisherClient.PublishAsync(message);
+ }
- Task.Delay(1000).GetAwaiter().GetResult();
+ // Wait one message per publish to be received by subscriber (in the subscriber's application message handler)
+ try
+ {
+ Task.Delay(30000, _cancellationTokenSource.Token).GetAwaiter().GetResult();
+ }
+ catch
+ {
}
- ///
- /// Publish messages and wait for messages sent to subscribers
- ///
- [Benchmark]
- public void DeliverMessages()
+ _cancellationTokenSource.Dispose();
+
+ if (_messagesReceivedCount < _messagesExpectedCount)
{
- // There should be one message received per publish for each subscribed topic
- _messagesExpectedCount = NumSubscribedTopicsPerSubscriber * NumSubscribers;
+ throw new Exception(string.Format("Messages Received Count mismatch, expected {0}, received {1}", _messagesExpectedCount, _messagesReceivedCount));
+ }
+ }
- // Loop for a while and exchange messages
- _messagesReceivedCount = 0;
+ [GlobalSetup]
+ public void Setup()
+ {
+ _lockMsgCount = new object();
- _cancellationTokenSource = new CancellationTokenSource();
+ Dictionary> singleWildcardTopicsByPublisher;
+ Dictionary> multiWildcardTopicsByPublisher;
- // same payload for all messages
- var payload = new byte[] { 1, 2, 3, 4 };
+ TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out _topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
- var tasks = new List();
+ var topics = _topicsByPublisher.First().Value;
+ _topicPublishMessages = new List();
+ // Prepare messages, same for each publisher
+ foreach (var topic in topics)
+ {
+ var message = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[] { 1, 2, 3, 4, 5, 6, 7, 8 }).Build();
+ _topicPublishMessages.Add(message);
+ }
- // publish a message for each subscribed topic
- foreach (var topic in _allSubscribedTopics)
- {
- var message = new MqttApplicationMessageBuilder()
- .WithTopic(topic)
- .WithPayload(payload)
- .Build();
- // pick the correct publisher
- var publisherName = _publisherByTopic[topic];
- var publisherClient = _mqttPublisherClientsByPublisherName[publisherName];
- _ = publisherClient.PublishAsync(message);
- }
+ // Create server
+ var serverFactory = new MqttServerFactory();
+ var clientFactory = new MqttClientFactory();
+ var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
+ _mqttServer = serverFactory.CreateMqttServer(serverOptions);
+ _mqttServer.StartAsync().GetAwaiter().GetResult();
- // Wait one message per publish to be received by subscriber (in the subscriber's application message handler)
- try
- {
- Task.Delay(30000, _cancellationTokenSource.Token).GetAwaiter().GetResult();
- }
- catch
+ // Create publisher clients
+ _mqttPublisherClientsByPublisherName = new Dictionary();
+ foreach (var pt in _topicsByPublisher)
+ {
+ var publisherName = pt.Key;
+ var mqttClient = clientFactory.CreateMqttClient();
+ var publisherOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").WithClientId(publisherName).WithKeepAlivePeriod(TimeSpan.FromSeconds(30)).Build();
+ mqttClient.ConnectAsync(publisherOptions).GetAwaiter().GetResult();
+ _mqttPublisherClientsByPublisherName.Add(publisherName, mqttClient);
+ }
+
+ // Create subscriber clients
+ _mqttSubscriberClients = new List();
+ for (var i = 0; i < NumSubscribers; i++)
+ {
+ var mqttSubscriberClient = clientFactory.CreateMqttClient();
+ _mqttSubscriberClients.Add(mqttSubscriberClient);
+
+ var subscriberOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").WithClientId("subscriber" + i).Build();
+ mqttSubscriberClient.ApplicationMessageReceivedAsync += r =>
{
+ // count messages and signal cancellation when expected message count is reached
+ lock (_lockMsgCount)
+ {
+ ++_messagesReceivedCount;
+ if (_messagesReceivedCount == _messagesExpectedCount)
+ {
+ _cancellationTokenSource.Cancel();
+ }
+ }
- }
+ return Task.CompletedTask;
+ };
+ mqttSubscriberClient.ConnectAsync(subscriberOptions).GetAwaiter().GetResult();
+ }
- _cancellationTokenSource.Dispose();
- if (_messagesReceivedCount < _messagesExpectedCount)
+ var allTopics = new List();
+ _publisherByTopic = new Dictionary();
+ foreach (var t in _topicsByPublisher)
+ {
+ foreach (var topic in t.Value)
{
- throw new Exception(string.Format("Messages Received Count mismatch, expected {0}, received {1}", _messagesExpectedCount, _messagesReceivedCount));
+ _publisherByTopic.Add(topic, t.Key);
+ allTopics.Add(topic);
}
}
- [GlobalCleanup]
- public void Cleanup()
+ // Subscribe to NumSubscribedTopics topics spread across all topics
+ _allSubscribedTopics = new List();
+
+ var totalNumTopics = NumPublishers * NumTopicsPerPublisher;
+ var topicIndexStep = totalNumTopics / (NumSubscribedTopicsPerSubscriber * NumSubscribers);
+ if (topicIndexStep * NumSubscribedTopicsPerSubscriber * NumSubscribers != totalNumTopics)
{
- foreach (var mp in _mqttPublisherClientsByPublisherName)
- {
- var mqttPublisherClient = mp.Value;
- mqttPublisherClient.DisconnectAsync().GetAwaiter().GetResult();
- mqttPublisherClient.Dispose();
- }
- _mqttPublisherClientsByPublisherName.Clear();
+ throw new Exception(
+ string.Format(
+ "The total number of topics must be divisible by the number of subscribed topics across all subscribers. Total number of topics: {0}, topic step: {1}",
+ totalNumTopics,
+ topicIndexStep));
+ }
- foreach (var mqttSubscriber in _mqttSubscriberClients)
+ var topicIndex = 0;
+ foreach (var mqttSubscriber in _mqttSubscriberClients)
+ {
+ for (var i = 0; i < NumSubscribedTopicsPerSubscriber; ++i, topicIndex += topicIndexStep)
{
- mqttSubscriber.DisconnectAsync().GetAwaiter().GetResult();
- mqttSubscriber.Dispose();
+ var topic = allTopics[topicIndex];
+ _allSubscribedTopics.Add(topic);
+ var subOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(new MqttTopicFilter { Topic = topic }).Build();
+ mqttSubscriber.SubscribeAsync(subOptions).GetAwaiter().GetResult();
}
- _mqttSubscriberClients.Clear();
-
- _mqttServer.StopAsync().GetAwaiter().GetResult();
- _mqttServer.Dispose();
- _mqttServer = null;
}
+
+ Task.Delay(1000).GetAwaiter().GetResult();
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs
index 43c0745c4..b0c869244 100644
--- a/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs
@@ -7,46 +7,43 @@
using MQTTnet.Client;
using MQTTnet.Server;
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks;
+
+[SimpleJob(RuntimeMoniker.Net60)]
+[RPlotExporter]
+[RankColumn]
+[MemoryDiagnoser]
+public class MessageProcessingBenchmark : BaseBenchmark
{
- [SimpleJob(RuntimeMoniker.Net60)]
- [RPlotExporter, RankColumn]
- [MemoryDiagnoser]
- public class MessageProcessingBenchmark : BaseBenchmark
- {
- MqttServer _mqttServer;
- IMqttClient _mqttClient;
- MqttApplicationMessage _message;
+ MqttApplicationMessage _message;
+ IMqttClient _mqttClient;
+ MqttServer _mqttServer;
- [GlobalSetup]
- public void Setup()
+ [Benchmark]
+ public void Send_10000_Messages()
+ {
+ for (var i = 0; i < 10000; i++)
{
- var serverOptions = new MqttServerOptionsBuilder().Build();
+ _mqttClient.PublishAsync(_message).GetAwaiter().GetResult();
+ }
+ }
- var serverFactory = new MqttServerFactory();
- _mqttServer = serverFactory.CreateMqttServer(serverOptions);
- var clientFactory = new MqttClientFactory();
- _mqttClient = clientFactory.CreateMqttClient();
+ [GlobalSetup]
+ public void Setup()
+ {
+ var serverOptions = new MqttServerOptionsBuilder().Build();
- _mqttServer.StartAsync().GetAwaiter().GetResult();
+ var serverFactory = new MqttServerFactory();
+ _mqttServer = serverFactory.CreateMqttServer(serverOptions);
+ var clientFactory = new MqttClientFactory();
+ _mqttClient = clientFactory.CreateMqttClient();
- var clientOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost").Build();
+ _mqttServer.StartAsync().GetAwaiter().GetResult();
- _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
+ var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").Build();
- _message = new MqttApplicationMessageBuilder()
- .WithTopic("A")
- .Build();
- }
+ _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
- [Benchmark]
- public void Send_10000_Messages()
- {
- for (var i = 0; i < 10000; i++)
- {
- _mqttClient.PublishAsync(_message).GetAwaiter().GetResult();
- }
- }
+ _message = new MqttApplicationMessageBuilder().WithTopic("A").Build();
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs b/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
index 16406087c..1dc822c40 100644
--- a/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
@@ -3,90 +3,86 @@
// See the LICENSE file in the project root for more information.
using System;
-using BenchmarkDotNet.Attributes;
-using MQTTnet.Channel;
-using MQTTnet.Implementations;
-using MQTTnet.Server;
+using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
+using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
+using MQTTnet.Channel;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
-using MQTTnet.Server.Adapter;
+using MQTTnet.Implementations;
+using MQTTnet.Server;
+using MQTTnet.Server.Internal.Adapter;
+
+namespace MQTTnet.Benchmarks;
-namespace MQTTnet.Benchmarks
+[SimpleJob(RuntimeMoniker.Net60)]
+[MemoryDiagnoser]
+public class MqttTcpChannelBenchmark : BaseBenchmark
{
- [SimpleJob(RuntimeMoniker.Net60)]
- [MemoryDiagnoser]
- public class MqttTcpChannelBenchmark : BaseBenchmark
+ IMqttChannel _clientChannel;
+ MqttServer _mqttServer;
+ IMqttChannel _serverChannel;
+
+ [Benchmark]
+ public async Task Send_10000_Chunks()
{
- MqttServer _mqttServer;
- IMqttChannel _serverChannel;
- IMqttChannel _clientChannel;
+ var size = 5;
+ var iterations = 10000;
+
+ await Task.WhenAll(WriteAsync(iterations, size), ReadAsync(iterations, size));
+ }
- [GlobalSetup]
- public void Setup()
+ [GlobalSetup]
+ public void Setup()
+ {
+ var serverFactory = new MqttServerFactory();
+ var tcpServer = new MqttTcpServerAdapter();
+ tcpServer.ClientHandler += args =>
{
- var serverFactory = new MqttServerFactory();
- var tcpServer = new MqttTcpServerAdapter();
- tcpServer.ClientHandler += args =>
- {
- _serverChannel =
- (IMqttChannel)args.GetType().GetField("_channel",
- System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)
- .GetValue(args);
+ _serverChannel = (IMqttChannel)args.GetType().GetField("_channel", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(args);
- return Task.CompletedTask;
- };
+ return Task.CompletedTask;
+ };
- var serverOptions = new MqttServerOptionsBuilder().Build();
- _mqttServer = serverFactory.CreateMqttServer(serverOptions, new[] { tcpServer }, new MqttNetEventLogger());
+ var serverOptions = new MqttServerOptionsBuilder().Build();
+ _mqttServer = serverFactory.CreateMqttServer(serverOptions, new[] { tcpServer }, new MqttNetEventLogger());
- _mqttServer.StartAsync().GetAwaiter().GetResult();
+ _mqttServer.StartAsync().GetAwaiter().GetResult();
- var clientOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost").Build();
+ var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").Build();
- var tcpOptions = (MqttClientTcpOptions)clientOptions.ChannelOptions;
- _clientChannel = new MqttTcpChannel(new MqttClientOptions { ChannelOptions = tcpOptions });
+ var tcpOptions = (MqttClientTcpOptions)clientOptions.ChannelOptions;
+ _clientChannel = new MqttTcpChannel(new MqttClientOptions { ChannelOptions = tcpOptions });
- _clientChannel.ConnectAsync(CancellationToken.None).GetAwaiter().GetResult();
- }
+ _clientChannel.ConnectAsync(CancellationToken.None).GetAwaiter().GetResult();
+ }
- [Benchmark]
- public async Task Send_10000_Chunks()
- {
- var size = 5;
- var iterations = 10000;
+ async Task ReadAsync(int iterations, int size)
+ {
+ await Task.Yield();
- await Task.WhenAll(WriteAsync(iterations, size), ReadAsync(iterations, size));
- }
+ var expected = iterations * size;
+ long read = 0;
- async Task ReadAsync(int iterations, int size)
+ while (read < expected)
{
- await Task.Yield();
-
- var expected = iterations * size;
- long read = 0;
-
- while (read < expected)
- {
- var readResult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
- read += readResult;
- }
+ var readResult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
+ read += readResult;
}
+ }
- async Task WriteAsync(int iterations, int size)
- {
- await Task.Yield();
+ async Task WriteAsync(int iterations, int size)
+ {
+ await Task.Yield();
- var buffer = new ArraySegment(new byte[size]);
+ var buffer = new ArraySegment(new byte[size]);
- for (var i = 0; i < iterations; i++)
- {
- await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
- }
+ for (var i = 0; i < iterations; i++)
+ {
+ await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
}
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.Benchmarks/Program.cs b/Source/MQTTnet.Benchmarks/Program.cs
index be231ff81..bff20e88d 100644
--- a/Source/MQTTnet.Benchmarks/Program.cs
+++ b/Source/MQTTnet.Benchmarks/Program.cs
@@ -71,7 +71,7 @@ static List CollectBenchmarks()
{
if (type != typeof(BaseBenchmark))
{
- benchmarks.Add(type);
+ benchmarks.Add(type);
}
}
}
@@ -87,7 +87,7 @@ static void HandleArguments(string[] arguments)
}
// Allow for preselection to avoid developer frustration.
-
+
if (int.TryParse(arguments[0], out var benchmarkIndex))
{
_selectedBenchmarkIndex = benchmarkIndex;
diff --git a/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs b/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
index 4ac0d4f65..41ba0a821 100644
--- a/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
@@ -31,7 +31,7 @@ public void GlobalSetup()
.WithPayload(new byte[10 * 1024])
.Build();
- var packet = MqttPacketFactories.Publish.Create(mqttMessage);
+ var packet = MqttPublishPacketFactory.Create(mqttMessage);
var buffer = mqttPacketFormatter.Encode(packet);
stream = new MemoryStream();
diff --git a/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs b/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs
index bb2b50a09..db4a8efaf 100644
--- a/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs
@@ -2,66 +2,60 @@
// 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 BenchmarkDotNet.Jobs;
using MQTTnet.Client;
using MQTTnet.Server;
-using System.Collections.Generic;
-using System.Linq;
-namespace MQTTnet.Benchmarks
-{
- [MemoryDiagnoser]
- public class SubscribeBenchmark : BaseBenchmark
- {
- MqttServer _mqttServer;
- IMqttClient _mqttClient;
+namespace MQTTnet.Benchmarks;
- const int NumPublishers = 1;
- const int NumTopicsPerPublisher = 10000;
+[MemoryDiagnoser]
+public class SubscribeBenchmark : BaseBenchmark
+{
+ const int NumPublishers = 1;
+ const int NumTopicsPerPublisher = 10000;
+ IMqttClient _mqttClient;
+ MqttServer _mqttServer;
- List _topics;
+ List _topics;
- [GlobalSetup]
- public void Setup()
- {
- TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
- _topics = topicsByPublisher.Values.First();
+ [GlobalCleanup]
+ public void Cleanup()
+ {
+ _mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+ _mqttServer.StopAsync().GetAwaiter().GetResult();
+ _mqttServer.Dispose();
+ }
- var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
+ [GlobalSetup]
+ public void Setup()
+ {
+ TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
+ _topics = topicsByPublisher.Values.First();
- var serverFactory = new MqttServerFactory();
- _mqttServer = serverFactory.CreateMqttServer(serverOptions);
- var clientFactory = new MqttClientFactory();
- _mqttClient = clientFactory.CreateMqttClient();
+ var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
- _mqttServer.StartAsync().GetAwaiter().GetResult();
+ var serverFactory = new MqttServerFactory();
+ _mqttServer = serverFactory.CreateMqttServer(serverOptions);
+ var clientFactory = new MqttClientFactory();
+ _mqttClient = clientFactory.CreateMqttClient();
- var clientOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost").Build();
+ _mqttServer.StartAsync().GetAwaiter().GetResult();
- _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
- }
+ var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").Build();
- [GlobalCleanup]
- public void Cleanup()
- {
- _mqttClient.DisconnectAsync().GetAwaiter().GetResult();
- _mqttServer.StopAsync().GetAwaiter().GetResult();
- _mqttServer.Dispose();
- }
+ _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
+ }
- [Benchmark]
- public void Subscribe_10000_Topics()
+ [Benchmark]
+ public void Subscribe_10000_Topics()
+ {
+ foreach (var topic in _topics)
{
- foreach (var topic in _topics)
- {
- var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
- .WithTopicFilter(topic, Protocol.MqttQualityOfServiceLevel.AtMostOnce)
- .Build();
+ var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topic).Build();
- _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult();
- }
+ _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult();
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs b/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs
index 2914d2811..48b5edde0 100644
--- a/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs
@@ -2,64 +2,57 @@
// 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.Client;
using MQTTnet.Server;
-using System.Collections.Generic;
-using System.Linq;
-namespace MQTTnet.Benchmarks
-{
- [MemoryDiagnoser]
- public class UnsubscribeBenchmark : BaseBenchmark
- {
- MqttServer _mqttServer;
- IMqttClient _mqttClient;
+namespace MQTTnet.Benchmarks;
- const int NumPublishers = 1;
- const int NumTopicsPerPublisher = 10000;
+[MemoryDiagnoser]
+public class UnsubscribeBenchmark : BaseBenchmark
+{
+ const int NumPublishers = 1;
+ const int NumTopicsPerPublisher = 10000;
+ IMqttClient _mqttClient;
+ MqttServer _mqttServer;
- List _topics;
+ List _topics;
- [GlobalSetup]
- public void Setup()
- {
- TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
- _topics = topicsByPublisher.Values.First();
+ [GlobalSetup]
+ public void Setup()
+ {
+ TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher);
+ _topics = topicsByPublisher.Values.First();
- var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
+ var serverOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
- var serverFactory = new MqttServerFactory();
- _mqttServer = serverFactory.CreateMqttServer(serverOptions);
- var clientFactory = new MqttClientFactory();
- _mqttClient = clientFactory.CreateMqttClient();
+ var serverFactory = new MqttServerFactory();
+ _mqttServer = serverFactory.CreateMqttServer(serverOptions);
+ var clientFactory = new MqttClientFactory();
+ _mqttClient = clientFactory.CreateMqttClient();
- _mqttServer.StartAsync().GetAwaiter().GetResult();
+ _mqttServer.StartAsync().GetAwaiter().GetResult();
- var clientOptions = new MqttClientOptionsBuilder()
- .WithTcpServer("localhost").Build();
+ var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("localhost").Build();
- _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
+ _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
- foreach (var topic in _topics)
- {
- var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
- .WithTopicFilter(topic)
- .Build();
- _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult();
- }
+ foreach (var topic in _topics)
+ {
+ var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topic).Build();
+ _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult();
}
+ }
- [Benchmark]
- public void Unsubscribe_10000_Topics()
+ [Benchmark]
+ public void Unsubscribe_10000_Topics()
+ {
+ foreach (var topic in _topics)
{
- foreach (var topic in _topics)
- {
- var unsubscribeOptions = new MqttClientUnsubscribeOptionsBuilder()
- .WithTopicFilter(topic)
- .Build();
- _mqttClient.UnsubscribeAsync(unsubscribeOptions).GetAwaiter().GetResult();
- }
+ var unsubscribeOptions = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(topic).Build();
+ _mqttClient.UnsubscribeAsync(unsubscribeOptions).GetAwaiter().GetResult();
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj b/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj
index 219a17682..5b236d98f 100644
--- a/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj
+++ b/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj
@@ -41,11 +41,11 @@
-
+
-
+
diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs
index aafa4d171..74c05ec57 100644
--- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs
+++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs
@@ -5,26 +5,24 @@
using System;
using MQTTnet.Client;
-namespace MQTTnet.Extensions.ManagedClient
-{
- public sealed class ManagedMqttClientOptions
- {
- public MqttClientOptions ClientOptions { get; set; }
+namespace MQTTnet.Extensions.ManagedClient;
- public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
+public sealed class ManagedMqttClientOptions
+{
+ public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
+ public MqttClientOptions ClientOptions { get; set; }
- public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1);
+ public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1);
- public IManagedMqttClientStorage Storage { get; set; }
+ public int MaxPendingMessages { get; set; } = int.MaxValue;
- public int MaxPendingMessages { get; set; } = int.MaxValue;
+ ///
+ /// Defines the maximum amount of topic filters which will be sent in a SUBSCRIBE/UNSUBSCRIBE packet.
+ /// Amazon Web Services (AWS) limits this number to 8. The default is int.MaxValue.
+ ///
+ public int MaxTopicFiltersInSubscribeUnsubscribePackets { get; set; } = int.MaxValue;
- public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage;
+ public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage;
- ///
- /// Defines the maximum amount of topic filters which will be sent in a SUBSCRIBE/UNSUBSCRIBE packet.
- /// Amazon Web Services (AWS) limits this number to 8. The default is int.MaxValue.
- ///
- public int MaxTopicFiltersInSubscribeUnsubscribePackets { get; set; } = int.MaxValue;
- }
-}
+ public IManagedMqttClientStorage Storage { get; set; }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
index a9be76699..fad9dee27 100644
--- a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
+++ b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
@@ -41,11 +41,11 @@
-
+
-
+
diff --git a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj.DotSettings b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj.DotSettings
index 1b491ba9f..18d916ce4 100644
--- a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj.DotSettings
+++ b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj.DotSettings
@@ -1,3 +1,7 @@
-
- True
- True
\ No newline at end of file
+
+ True
+ True
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj b/Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj
index f623d461d..5bd02a484 100644
--- a/Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj
+++ b/Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj
@@ -8,9 +8,9 @@
The contributors of MQTTnet
MQTTnet
- This is an extension library which provides mqtt topic templating logic to support dispatch,
- routing and similar functionality based on the well known moustache syntax, aka
- AsyncAPI dynamic channel address.
+ This is an extension library which provides mqtt topic templating logic to support dispatch,
+ routing and similar functionality based on the well known moustache syntax, aka
+ AsyncAPI dynamic channel address.
The contributors of MQTTnet
MQTTnet.Extensions.TopicTemplate
@@ -42,11 +42,11 @@
-
+
-
+
diff --git a/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptions.cs b/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptions.cs
index e9e61e9a6..763217fb1 100644
--- a/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptions.cs
+++ b/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptions.cs
@@ -2,11 +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.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-namespace MQTTnet.Server.Disconnecting
+namespace MQTTnet.Server
{
public sealed class MqttServerClientDisconnectOptions
{
diff --git a/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs b/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs
index 362585c65..7173ac64b 100644
--- a/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs
+++ b/Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs
@@ -2,11 +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.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-namespace MQTTnet.Server.Disconnecting
+namespace MQTTnet.Server
{
public sealed class MqttServerClientDisconnectOptionsBuilder
{
diff --git a/Source/MQTTnet.Server/Events/PreparingSessionEventArgs.cs b/Source/MQTTnet.Server/Events/PreparingSessionEventArgs.cs
index 584c4d4ec..d75ed503a 100644
--- a/Source/MQTTnet.Server/Events/PreparingSessionEventArgs.cs
+++ b/Source/MQTTnet.Server/Events/PreparingSessionEventArgs.cs
@@ -5,6 +5,7 @@
using System;
using System.Collections.Generic;
using MQTTnet.Packets;
+using MQTTnet.Server.Internal;
namespace MQTTnet.Server
{
diff --git a/Source/MQTTnet.Server/Adapter/IMqttServerAdapter.cs b/Source/MQTTnet.Server/IMqttServerAdapter.cs
similarity index 93%
rename from Source/MQTTnet.Server/Adapter/IMqttServerAdapter.cs
rename to Source/MQTTnet.Server/IMqttServerAdapter.cs
index d2322e42a..3f5ac0ce7 100644
--- a/Source/MQTTnet.Server/Adapter/IMqttServerAdapter.cs
+++ b/Source/MQTTnet.Server/IMqttServerAdapter.cs
@@ -5,7 +5,7 @@
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
-namespace MQTTnet.Server.Adapter;
+namespace MQTTnet.Server;
public interface IMqttServerAdapter : IDisposable
{
diff --git a/Source/MQTTnet.Server/InjectedMqttApplicationMessage.cs b/Source/MQTTnet.Server/InjectedMqttApplicationMessage.cs
index ed829b239..3a9add06a 100644
--- a/Source/MQTTnet.Server/InjectedMqttApplicationMessage.cs
+++ b/Source/MQTTnet.Server/InjectedMqttApplicationMessage.cs
@@ -2,27 +2,25 @@
// 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;
-namespace MQTTnet.Server
+namespace MQTTnet.Server;
+
+public sealed class InjectedMqttApplicationMessage
{
- public sealed class InjectedMqttApplicationMessage
+ public InjectedMqttApplicationMessage(MqttApplicationMessage applicationMessage)
{
- public InjectedMqttApplicationMessage(MqttApplicationMessage applicationMessage)
- {
- ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
- }
+ ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+ }
- public MqttApplicationMessage ApplicationMessage { get; }
+ public MqttApplicationMessage ApplicationMessage { get; }
- ///
- /// Gets or sets the session items which should be used for all event handlers which are involved in message
- /// processing.
- /// If _null_ is specified the singleton session items from the server are used instead.
- ///
- public IDictionary CustomSessionItems { get; set; }
+ ///
+ /// Gets or sets the session items which should be used for all event handlers which are involved in message
+ /// processing.
+ /// If _null_ is specified the singleton session items from the server are used instead.
+ ///
+ public IDictionary CustomSessionItems { get; set; }
- public string SenderClientId { get; set; } = string.Empty;
- }
+ public string SenderClientId { get; set; } = string.Empty;
}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/Adapter/MqttTcpServerAdapter.cs b/Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerAdapter.cs
similarity index 98%
rename from Source/MQTTnet.Server/Adapter/MqttTcpServerAdapter.cs
rename to Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerAdapter.cs
index 2751b95a4..921999b65 100644
--- a/Source/MQTTnet.Server/Adapter/MqttTcpServerAdapter.cs
+++ b/Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerAdapter.cs
@@ -8,7 +8,7 @@
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
-namespace MQTTnet.Server.Adapter
+namespace MQTTnet.Server.Internal.Adapter
{
public sealed class MqttTcpServerAdapter : IMqttServerAdapter
{
diff --git a/Source/MQTTnet.Server/Adapter/MqttTcpServerListener.cs b/Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerListener.cs
similarity index 99%
rename from Source/MQTTnet.Server/Adapter/MqttTcpServerListener.cs
rename to Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerListener.cs
index 7ad085c8a..b5cd9be1a 100644
--- a/Source/MQTTnet.Server/Adapter/MqttTcpServerListener.cs
+++ b/Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerListener.cs
@@ -12,7 +12,7 @@
using MQTTnet.Implementations;
using MQTTnet.Internal;
-namespace MQTTnet.Server.Adapter
+namespace MQTTnet.Server.Internal.Adapter
{
public sealed class MqttTcpServerListener : IDisposable
{
diff --git a/Source/MQTTnet.Server/Internal/CheckSubscriptionsResult.cs b/Source/MQTTnet.Server/Internal/CheckSubscriptionsResult.cs
index 7c20e8d8d..298a6b21d 100644
--- a/Source/MQTTnet.Server/Internal/CheckSubscriptionsResult.cs
+++ b/Source/MQTTnet.Server/Internal/CheckSubscriptionsResult.cs
@@ -2,10 +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.Collections.Generic;
using MQTTnet.Protocol;
-namespace MQTTnet.Server
+namespace MQTTnet.Server.Internal
{
public sealed class CheckSubscriptionsResult
{
diff --git a/Source/MQTTnet.Server/Internal/DispatchApplicationMessageResult.cs b/Source/MQTTnet.Server/Internal/DispatchApplicationMessageResult.cs
index 12527f4d7..e7daeecc8 100644
--- a/Source/MQTTnet.Server/Internal/DispatchApplicationMessageResult.cs
+++ b/Source/MQTTnet.Server/Internal/DispatchApplicationMessageResult.cs
@@ -2,10 +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.Collections.Generic;
using MQTTnet.Packets;
-namespace MQTTnet.Server
+namespace MQTTnet.Server.Internal
{
public sealed class DispatchApplicationMessageResult
{
diff --git a/Source/MQTTnet.Server/Internal/EnqueueDataPacketResult.cs b/Source/MQTTnet.Server/Internal/EnqueueDataPacketResult.cs
index 742cb7ad6..0952efbb5 100644
--- a/Source/MQTTnet.Server/Internal/EnqueueDataPacketResult.cs
+++ b/Source/MQTTnet.Server/Internal/EnqueueDataPacketResult.cs
@@ -2,7 +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.
-namespace MQTTnet.Server
+namespace MQTTnet.Server.Internal
{
public enum EnqueueDataPacketResult
{
diff --git a/Source/MQTTnet.Server/Formatter/MqttConnAckPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttConnAckPacketFactory.cs
similarity index 97%
rename from Source/MQTTnet.Server/Formatter/MqttConnAckPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttConnAckPacketFactory.cs
index e65798ba8..39e8a2ae1 100644
--- a/Source/MQTTnet.Server/Formatter/MqttConnAckPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttConnAckPacketFactory.cs
@@ -6,7 +6,7 @@
using MQTTnet.Packets;
using MQTTnet.Protocol;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttConnAckPacketFactory
{
diff --git a/Source/MQTTnet.Server/Formatter/MqttDisconnectPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttDisconnectPacketFactory.cs
similarity index 94%
rename from Source/MQTTnet.Server/Formatter/MqttDisconnectPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttDisconnectPacketFactory.cs
index a2e33e5f0..d7b4199ba 100644
--- a/Source/MQTTnet.Server/Formatter/MqttDisconnectPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttDisconnectPacketFactory.cs
@@ -4,9 +4,8 @@
using MQTTnet.Packets;
using MQTTnet.Protocol;
-using MQTTnet.Server.Disconnecting;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttDisconnectPacketFactory
{
diff --git a/Source/MQTTnet.Server/Formatter/MqttPubAckPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttPubAckPacketFactory.cs
similarity index 96%
rename from Source/MQTTnet.Server/Formatter/MqttPubAckPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttPubAckPacketFactory.cs
index 9e147d381..8e59eb7a9 100644
--- a/Source/MQTTnet.Server/Formatter/MqttPubAckPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttPubAckPacketFactory.cs
@@ -5,7 +5,7 @@
using MQTTnet.Packets;
using MQTTnet.Protocol;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttPubAckPacketFactory
{
diff --git a/Source/MQTTnet.Server/Internal/Formatter/MqttPubCompPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttPubCompPacketFactory.cs
new file mode 100644
index 000000000..958af7282
--- /dev/null
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttPubCompPacketFactory.cs
@@ -0,0 +1,26 @@
+// 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.Client;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server.Internal.Formatter;
+
+public static class MqttPubCompPacketFactory
+{
+ public static MqttPubCompPacket Create(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+ {
+ if (pubRelPacket == null)
+ {
+ throw new ArgumentNullException(nameof(pubRelPacket));
+ }
+
+ return new MqttPubCompPacket
+ {
+ PacketIdentifier = pubRelPacket.PacketIdentifier,
+ ReasonCode = (MqttPubCompReasonCode)(int)reasonCode
+ };
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/Formatter/MqttPubRecPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttPubRecPacketFactory.cs
similarity index 95%
rename from Source/MQTTnet.Server/Formatter/MqttPubRecPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttPubRecPacketFactory.cs
index cb37231d9..ad800a055 100644
--- a/Source/MQTTnet.Server/Formatter/MqttPubRecPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttPubRecPacketFactory.cs
@@ -5,7 +5,7 @@
using MQTTnet.Packets;
using MQTTnet.Protocol;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttPubRecPacketFactory
{
diff --git a/Source/MQTTnet.Server/Internal/Formatter/MqttPubRelPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttPubRelPacketFactory.cs
new file mode 100644
index 000000000..50c6f8852
--- /dev/null
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttPubRelPacketFactory.cs
@@ -0,0 +1,26 @@
+// 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.Client;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server.Internal.Formatter;
+
+public static class MqttPubRelPacketFactory
+{
+ public static MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+ {
+ if (pubRecPacket == null)
+ {
+ throw new ArgumentNullException(nameof(pubRecPacket));
+ }
+
+ return new MqttPubRelPacket
+ {
+ PacketIdentifier = pubRecPacket.PacketIdentifier,
+ ReasonCode = (MqttPubRelReasonCode)(int)reasonCode
+ };
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttPublishPacketFactory.cs
similarity index 59%
rename from Source/MQTTnet.Server/Formatter/MqttPublishPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttPublishPacketFactory.cs
index 33a18584c..6e747f4d0 100644
--- a/Source/MQTTnet.Server/Formatter/MqttPublishPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttPublishPacketFactory.cs
@@ -2,12 +2,48 @@
// 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.Exceptions;
using MQTTnet.Packets;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttPublishPacketFactory
{
+ public static MqttPublishPacket Create(MqttConnectPacket connectPacket)
+ {
+ if (connectPacket == null)
+ {
+ throw new ArgumentNullException(nameof(connectPacket));
+ }
+
+ if (!connectPacket.WillFlag)
+ {
+ throw new MqttProtocolViolationException("The CONNECT packet contains no will message (WillFlag).");
+ }
+
+ ArraySegment willMessageBuffer = default;
+ if (connectPacket.WillMessage?.Length > 0)
+ {
+ willMessageBuffer = new ArraySegment(connectPacket.WillMessage);
+ }
+
+ var packet = new MqttPublishPacket
+ {
+ Topic = connectPacket.WillTopic,
+ PayloadSegment = willMessageBuffer,
+ QualityOfServiceLevel = connectPacket.WillQoS,
+ Retain = connectPacket.WillRetain,
+ ContentType = connectPacket.WillContentType,
+ CorrelationData = connectPacket.WillCorrelationData,
+ MessageExpiryInterval = connectPacket.WillMessageExpiryInterval,
+ PayloadFormatIndicator = connectPacket.WillPayloadFormatIndicator,
+ ResponseTopic = connectPacket.WillResponseTopic,
+ UserProperties = connectPacket.WillUserProperties
+ };
+
+ return packet;
+ }
+
public static MqttPublishPacket Create(MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
diff --git a/Source/MQTTnet.Server/Formatter/MqttSubAckPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttSubAckPacketFactory.cs
similarity index 95%
rename from Source/MQTTnet.Server/Formatter/MqttSubAckPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttSubAckPacketFactory.cs
index 4f42e2664..38f4eeeaa 100644
--- a/Source/MQTTnet.Server/Formatter/MqttSubAckPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttSubAckPacketFactory.cs
@@ -4,7 +4,7 @@
using MQTTnet.Packets;
-namespace MQTTnet.Server.Formatter;
+namespace MQTTnet.Server.Internal.Formatter;
public static class MqttSubAckPacketFactory
{
diff --git a/Source/MQTTnet.Server/Formatter/MqttUnsubAckPacketFactory.cs b/Source/MQTTnet.Server/Internal/Formatter/MqttUnsubAckPacketFactory.cs
similarity index 95%
rename from Source/MQTTnet.Server/Formatter/MqttUnsubAckPacketFactory.cs
rename to Source/MQTTnet.Server/Internal/Formatter/MqttUnsubAckPacketFactory.cs
index 46a517e2c..ad630882b 100644
--- a/Source/MQTTnet.Server/Formatter/MqttUnsubAckPacketFactory.cs
+++ b/Source/MQTTnet.Server/Internal/Formatter/MqttUnsubAckPacketFactory.cs
@@ -4,7 +4,7 @@
using MQTTnet.Packets;
-namespace MQTTnet.Server.Formatter
+namespace MQTTnet.Server.Internal.Formatter
{
public static class MqttUnsubAckPacketFactory
{
diff --git a/Source/MQTTnet.Server/Internal/ISubscriptionChangedNotification.cs b/Source/MQTTnet.Server/Internal/ISubscriptionChangedNotification.cs
index bff6769af..af9538df6 100644
--- a/Source/MQTTnet.Server/Internal/ISubscriptionChangedNotification.cs
+++ b/Source/MQTTnet.Server/Internal/ISubscriptionChangedNotification.cs
@@ -1,6 +1,4 @@
-using System.Collections.Generic;
-
-namespace MQTTnet.Server
+namespace MQTTnet.Server.Internal
{
public interface ISubscriptionChangedNotification
{
diff --git a/Source/MQTTnet.Server/Internal/MqttClient.cs b/Source/MQTTnet.Server/Internal/MqttClient.cs
index 615703966..c2283c45c 100644
--- a/Source/MQTTnet.Server/Internal/MqttClient.cs
+++ b/Source/MQTTnet.Server/Internal/MqttClient.cs
@@ -10,571 +10,568 @@
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-using MQTTnet.Server.Disconnecting;
-using MQTTnet.Server.Formatter;
-using MqttDisconnectPacketFactory = MQTTnet.Server.Formatter.MqttDisconnectPacketFactory;
-using MqttPubAckPacketFactory = MQTTnet.Server.Formatter.MqttPubAckPacketFactory;
-using MqttPublishPacketFactory = MQTTnet.Server.Formatter.MqttPublishPacketFactory;
-using MqttPubRecPacketFactory = MQTTnet.Server.Formatter.MqttPubRecPacketFactory;
-
-namespace MQTTnet.Server
+using MQTTnet.Server.Internal.Formatter;
+using MqttDisconnectPacketFactory = MQTTnet.Server.Internal.Formatter.MqttDisconnectPacketFactory;
+using MqttPubAckPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPubAckPacketFactory;
+using MqttPubCompPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPubCompPacketFactory;
+using MqttPublishPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPublishPacketFactory;
+using MqttPubRecPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPubRecPacketFactory;
+using MqttPubRelPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPubRelPacketFactory;
+
+namespace MQTTnet.Server.Internal;
+
+public sealed class MqttClient : IDisposable
{
- public sealed class MqttClient : IDisposable
+ readonly MqttServerEventContainer _eventContainer;
+ readonly MqttNetSourceLogger _logger;
+ readonly MqttServerOptions _serverOptions;
+ readonly MqttClientSessionsManager _sessionsManager;
+ readonly Dictionary _topicAlias = new();
+
+ CancellationTokenSource _cancellationToken = new();
+ bool _disconnectPacketSent;
+
+ public MqttClient(
+ MqttConnectPacket connectPacket,
+ IMqttChannelAdapter channelAdapter,
+ MqttSession session,
+ MqttServerOptions serverOptions,
+ MqttServerEventContainer eventContainer,
+ MqttClientSessionsManager sessionsManager,
+ IMqttNetLogger logger)
{
- readonly MqttConnectPacket _connectPacket;
- readonly MqttServerEventContainer _eventContainer;
- readonly MqttNetSourceLogger _logger;
- readonly MqttServerOptions _serverOptions;
- readonly MqttClientSessionsManager _sessionsManager;
- readonly Dictionary _topicAlias = new Dictionary();
-
- CancellationTokenSource _cancellationToken = new CancellationTokenSource();
- bool _disconnectPacketSent;
-
- public MqttClient(
- MqttConnectPacket connectPacket,
- IMqttChannelAdapter channelAdapter,
- MqttSession session,
- MqttServerOptions serverOptions,
- MqttServerEventContainer eventContainer,
- MqttClientSessionsManager sessionsManager,
- IMqttNetLogger logger)
- {
- _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
- _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
- _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
- _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
+ _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
+ _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
+ _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
+ ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
- ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
- Endpoint = channelAdapter.Endpoint;
- Session = session ?? throw new ArgumentNullException(nameof(session));
+ ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
+ Endpoint = channelAdapter.Endpoint;
+ Session = session ?? throw new ArgumentNullException(nameof(session));
- if (logger == null)
- {
- throw new ArgumentNullException(nameof(logger));
- }
-
- _logger = logger.WithSource(nameof(MqttClient));
+ if (logger == null)
+ {
+ throw new ArgumentNullException(nameof(logger));
}
- public IMqttChannelAdapter ChannelAdapter { get; }
+ _logger = logger.WithSource(nameof(MqttClient));
+ }
- public MqttDisconnectPacket DisconnectPacket { get; set; }
+ public IMqttChannelAdapter ChannelAdapter { get; }
- public string Endpoint { get; }
+ public MqttConnectPacket ConnectPacket { get; }
- public string Id => _connectPacket.ClientId;
+ public MqttDisconnectPacket DisconnectPacket { get; private set; }
- public bool IsRunning { get; private set; }
+ public string Endpoint { get; }
- public bool IsTakenOver { get; set; }
+ public string Id => ConnectPacket.ClientId;
- public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
+ public bool IsRunning { get; private set; }
- public MqttSession Session { get; }
+ public bool IsTakenOver { get; set; }
- public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
+ public MqttSession Session { get; }
- public void Dispose()
- {
- _cancellationToken?.Dispose();
- }
+ public MqttClientStatistics Statistics { get; } = new();
- public void ResetStatistics()
- {
- ChannelAdapter.ResetStatistics();
- Statistics.ResetStatistics();
- }
+ public void Dispose()
+ {
+ _cancellationToken?.Dispose();
+ }
- public async Task RunAsync()
- {
- _logger.Info("Client '{0}': Session started", Id);
+ public void ResetStatistics()
+ {
+ ChannelAdapter.ResetStatistics();
+ Statistics.ResetStatistics();
+ }
- Session.LatestConnectPacket = _connectPacket;
- Session.WillMessageSent = false;
+ public async Task RunAsync()
+ {
+ _logger.Info("Client '{0}': Session started", Id);
- try
- {
- var cancellationToken = _cancellationToken.Token;
- IsRunning = true;
+ Session.LatestConnectPacket = ConnectPacket;
+ Session.WillMessageSent = false;
- _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default)
- .ConfigureAwait(false);
+ try
+ {
+ var cancellationToken = _cancellationToken.Token;
+ IsRunning = true;
- await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
- }
- finally
- {
- IsRunning = false;
+ _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
- Session.DisconnectedTimestamp = DateTime.UtcNow;
+ await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
+ }
+ finally
+ {
+ IsRunning = false;
- _cancellationToken?.TryCancel();
- _cancellationToken?.Dispose();
- _cancellationToken = null;
- }
+ Session.DisconnectedTimestamp = DateTime.UtcNow;
- var isCleanDisconnect = DisconnectPacket != null;
+ _cancellationToken?.TryCancel();
+ _cancellationToken?.Dispose();
+ _cancellationToken = null;
+ }
- if (!IsTakenOver && !isCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent)
- {
- var willPublishPacket = MqttPacketFactories.Publish.Create(Session.LatestConnectPacket);
- var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket);
+ var isCleanDisconnect = DisconnectPacket != null;
- _ = _sessionsManager.DispatchApplicationMessage(Id, Session.Items, willApplicationMessage, CancellationToken.None);
- Session.WillMessageSent = true;
+ if (!IsTakenOver && !isCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent)
+ {
+ var willPublishPacket = MqttPublishPacketFactory.Create(Session.LatestConnectPacket);
+ var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket);
- _logger.Info("Client '{0}': Published will message", Id);
- }
+ _ = _sessionsManager.DispatchApplicationMessage(Id, Session.Items, willApplicationMessage, CancellationToken.None);
+ Session.WillMessageSent = true;
- _logger.Info("Client '{0}': Connection stopped", Id);
+ _logger.Info("Client '{0}': Published will message", Id);
}
- public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
- {
- packet = await InterceptPacketAsync(packet, cancellationToken).ConfigureAwait(false);
- if (packet == null)
- {
- // The interceptor has decided that this packet will not used at all.
- // This might break the protocol but the user wants that.
- return;
- }
+ _logger.Info("Client '{0}': Connection stopped", Id);
+ }
- await ChannelAdapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
- Statistics.HandleSentPacket(packet);
+ public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
+ {
+ packet = await InterceptPacketAsync(packet, cancellationToken).ConfigureAwait(false);
+ if (packet == null)
+ {
+ // The interceptor has decided that this packet will not used at all.
+ // This might break the protocol but the user wants that.
+ return;
}
- public async Task StopAsync(MqttServerClientDisconnectOptions disconnectOptions)
- {
- IsRunning = false;
+ await ChannelAdapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
+ Statistics.HandleSentPacket(packet);
+ }
+
+ public async Task StopAsync(MqttServerClientDisconnectOptions disconnectOptions)
+ {
+ IsRunning = false;
- if (!_disconnectPacketSent)
+ if (!_disconnectPacketSent)
+ {
+ // // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
+ if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{
- // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
- if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
+ // From RFC: The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
+ // This library does not sent a DISCONNECT packet for a normal disconnection.
+ // TODO: Maybe adding a configuration option is requested in the future.
+ if (disconnectOptions != null)
{
- // From RFC: The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
- // This library does not sent a DISCONNECT packet for a normal disconnection.
- // TODO: Maybe adding a configuration option is requested in the future.
- if (disconnectOptions != null)
+ if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
+ !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
{
- if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
- !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
- {
- // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
- // token because the entire connection is closed (disposed) as soon as the cancellation
- // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
- // at the client!
- await TrySendDisconnectPacket(disconnectOptions).ConfigureAwait(false);
- }
+ // It is very important to send the DISCONNECT packet here BEFORE cancelling the
+ // token because the entire connection is closed (disposed) as soon as the cancellation
+ // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+ // at the client!
+ await TrySendDisconnectPacket(disconnectOptions).ConfigureAwait(false);
}
}
}
-
- StopInternal();
}
- Task ClientAcknowledgedPublishPacket(MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
- {
- if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
- {
- var eventArgs = new ClientAcknowledgedPublishPacketEventArgs(Id, Session.Items, publishPacket, acknowledgePacket);
- return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
- }
-
- return CompletedTask.Instance;
- }
+ StopInternal();
+ }
- void HandleIncomingPingReqPacket()
+ Task ClientAcknowledgedPublishPacket(MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
+ {
+ if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
{
- // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1].
- Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
+ var eventArgs = new ClientAcknowledgedPublishPacketEventArgs(Id, Session.Items, publishPacket, acknowledgePacket);
+ return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
}
- Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket)
- {
- var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
+ return CompletedTask.Instance;
+ }
- if (acknowledgedPublishPacket != null)
- {
- return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket);
- }
+ void HandleIncomingPingReqPacket()
+ {
+ // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1].
+ Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
+ }
- return CompletedTask.Instance;
- }
+ Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket)
+ {
+ var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
- Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket)
+ if (acknowledgedPublishPacket != null)
{
- var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
+ return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket);
+ }
- if (acknowledgedPublishPacket != null)
- {
- return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket);
- }
+ return CompletedTask.Instance;
+ }
- return CompletedTask.Instance;
- }
+ Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket)
+ {
+ var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
- async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+ if (acknowledgedPublishPacket != null)
{
- HandleTopicAlias(publishPacket);
+ return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket);
+ }
+
+ return CompletedTask.Instance;
+ }
+
+ async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+ {
+ HandleTopicAlias(publishPacket);
- var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
+ var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
- var dispatchApplicationMessageResult =
- await _sessionsManager.DispatchApplicationMessage(Id, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);
+ var dispatchApplicationMessageResult = await _sessionsManager.DispatchApplicationMessage(Id, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);
- if (dispatchApplicationMessageResult.CloseConnection)
+ if (dispatchApplicationMessageResult.CloseConnection)
+ {
+ await StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.UnspecifiedError });
+ return;
+ }
+
+ switch (publishPacket.QualityOfServiceLevel)
+ {
+ case MqttQualityOfServiceLevel.AtMostOnce:
{
- await StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.UnspecifiedError });
- return;
+ // Do nothing since QoS 0 has no ACK at all!
+ break;
}
-
- switch (publishPacket.QualityOfServiceLevel)
+ case MqttQualityOfServiceLevel.AtLeastOnce:
{
- case MqttQualityOfServiceLevel.AtMostOnce:
- {
- // Do nothing since QoS 0 has no ACK at all!
- break;
- }
- case MqttQualityOfServiceLevel.AtLeastOnce:
- {
- var pubAckPacket = MqttPubAckPacketFactory.Create(publishPacket, dispatchApplicationMessageResult);
- Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
- break;
- }
- case MqttQualityOfServiceLevel.ExactlyOnce:
- {
- var pubRecPacket = MqttPubRecPacketFactory.Create(publishPacket, dispatchApplicationMessageResult);
- Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
- break;
- }
- default:
- {
- throw new MqttCommunicationException("Received a not supported QoS level");
- }
+ var pubAckPacket = MqttPubAckPacketFactory.Create(publishPacket, dispatchApplicationMessageResult);
+ Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
+ break;
+ }
+ case MqttQualityOfServiceLevel.ExactlyOnce:
+ {
+ var pubRecPacket = MqttPubRecPacketFactory.Create(publishPacket, dispatchApplicationMessageResult);
+ Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
+ break;
+ }
+ default:
+ {
+ throw new MqttCommunicationException("Received a not supported QoS level");
}
}
+ }
- Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
- {
- // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
- // properly when the client has sent the PUBCOMP packet.
- var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
- Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
+ Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
+ {
+ // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
+ // properly when the client has sent the PUBCOMP packet.
+ var pubRelPacket = MqttPubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
- return CompletedTask.Instance;
- }
+ return CompletedTask.Instance;
+ }
- void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
- {
- var pubCompPacket = MqttPacketFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
- Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
- }
+ void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
+ {
+ var pubCompPacket = MqttPubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
+ }
- async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
- {
- var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
+ async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+ {
+ var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
- var subAckPacket = MqttSubAckPacketFactory.Create(subscribePacket, subscribeResult);
+ var subAckPacket = MqttSubAckPacketFactory.Create(subscribePacket, subscribeResult);
- Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
+ Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
- if (subscribeResult.CloseConnection)
- {
- StopInternal();
- return;
- }
+ if (subscribeResult.CloseConnection)
+ {
+ StopInternal();
+ return;
+ }
- if (subscribeResult.RetainedMessages != null)
+ if (subscribeResult.RetainedMessages != null)
+ {
+ foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
{
- foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
- {
- var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
- Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
- }
+ var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+ Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
+ }
- async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
- {
- var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
+ async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+ {
+ var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
- var unsubAckPacket = MqttUnsubAckPacketFactory.Create(unsubscribePacket, unsubscribeResult);
+ var unsubAckPacket = MqttUnsubAckPacketFactory.Create(unsubscribePacket, unsubscribeResult);
- Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
+ Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
- if (unsubscribeResult.CloseConnection)
- {
- StopInternal();
- }
+ if (unsubscribeResult.CloseConnection)
+ {
+ StopInternal();
+ }
+ }
+
+ void HandleTopicAlias(MqttPublishPacket publishPacket)
+ {
+ if (publishPacket.TopicAlias == 0)
+ {
+ return;
}
- void HandleTopicAlias(MqttPublishPacket publishPacket)
+ lock (_topicAlias)
{
- if (publishPacket.TopicAlias == 0)
+ if (!string.IsNullOrEmpty(publishPacket.Topic))
{
- return;
+ _topicAlias[publishPacket.TopicAlias] = publishPacket.Topic;
}
-
- lock (_topicAlias)
+ else
{
- if (!string.IsNullOrEmpty(publishPacket.Topic))
+ if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic))
{
- _topicAlias[publishPacket.TopicAlias] = publishPacket.Topic;
+ publishPacket.Topic = topic;
}
else
{
- if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic))
- {
- publishPacket.Topic = topic;
- }
- else
- {
- _logger.Warning("Client '{0}': Received invalid topic alias ({1})", Id, publishPacket.TopicAlias);
- }
+ _logger.Warning("Client '{0}': Received invalid topic alias ({1})", Id, publishPacket.TopicAlias);
}
}
}
+ }
- async Task InterceptPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
+ async Task InterceptPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
+ {
+ if (!_eventContainer.InterceptingOutboundPacketEvent.HasHandlers)
{
- if (!_eventContainer.InterceptingOutboundPacketEvent.HasHandlers)
- {
- return packet;
- }
-
- var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
- await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
+ return packet;
+ }
- if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
- {
- return null;
- }
+ var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
+ await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
- return interceptingPacketEventArgs.Packet;
+ if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
+ {
+ return null;
}
- async Task ReceivePackagesLoop(CancellationToken cancellationToken)
+ return interceptingPacketEventArgs.Packet;
+ }
+
+ async Task ReceivePackagesLoop(CancellationToken cancellationToken)
+ {
+ MqttPacket currentPacket = null;
+ try
{
- MqttPacket currentPacket = null;
- try
+ // We do not listen for the cancellation token here because the internal buffer might still
+ // contain data to be read even if the TCP connection was already dropped. So we rely on an
+ // own exception in the reading loop!
+ while (!cancellationToken.IsCancellationRequested)
{
- // We do not listen for the cancellation token here because the internal buffer might still
- // contain data to be read even if the TCP connection was already dropped. So we rely on an
- // own exception in the reading loop!
- while (!cancellationToken.IsCancellationRequested)
- {
- await Task.Yield();
+ await Task.Yield();
- currentPacket = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
- if (currentPacket == null)
- {
- return;
- }
+ currentPacket = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
+ if (currentPacket == null)
+ {
+ return;
+ }
- // Check for cancellation again because receive packet might block some time.
- if (cancellationToken.IsCancellationRequested)
- {
- return;
- }
+ // Check for cancellation again because receive packet might block some time.
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return;
+ }
- // The TCP connection of this client may be still open but the client has already been taken over by
- // a new TCP connection. So we must exit here to make sure to no longer process any message.
- if (IsTakenOver || !IsRunning)
- {
- return;
- }
+ // The TCP connection of this client may be still open but the client has already been taken over by
+ // a new TCP connection. So we must exit here to make sure to no longer process any message.
+ if (IsTakenOver || !IsRunning)
+ {
+ return;
+ }
- var processPacket = true;
+ var processPacket = true;
- if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
- {
- var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, currentPacket, Session.Items);
- await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
- currentPacket = interceptingPacketEventArgs.Packet;
- processPacket = interceptingPacketEventArgs.ProcessPacket;
- }
+ if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
+ {
+ var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, currentPacket, Session.Items);
+ await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
+ currentPacket = interceptingPacketEventArgs.Packet;
+ processPacket = interceptingPacketEventArgs.ProcessPacket;
+ }
- if (!processPacket || currentPacket == null)
- {
- // Restart the receiving process to get the next packet ignoring the current one..
- continue;
- }
+ if (!processPacket || currentPacket == null)
+ {
+ // Restart the receiving process to get the next packet ignoring the current one..
+ continue;
+ }
- Statistics.HandleReceivedPacket(currentPacket);
+ Statistics.HandleReceivedPacket(currentPacket);
- if (currentPacket is MqttPublishPacket publishPacket)
- {
- await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false);
- }
- else if (currentPacket is MqttPubAckPacket pubAckPacket)
- {
- await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false);
- }
- else if (currentPacket is MqttPubCompPacket pubCompPacket)
- {
- await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false);
- }
- else if (currentPacket is MqttPubRecPacket pubRecPacket)
- {
- await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false);
- }
- else if (currentPacket is MqttPubRelPacket pubRelPacket)
- {
- HandleIncomingPubRelPacket(pubRelPacket);
- }
- else if (currentPacket is MqttSubscribePacket subscribePacket)
- {
- await HandleIncomingSubscribePacket(subscribePacket, cancellationToken).ConfigureAwait(false);
- }
- else if (currentPacket is MqttUnsubscribePacket unsubscribePacket)
- {
- await HandleIncomingUnsubscribePacket(unsubscribePacket, cancellationToken).ConfigureAwait(false);
- }
- else if (currentPacket is MqttPingReqPacket)
- {
- HandleIncomingPingReqPacket();
- }
- else if (currentPacket is MqttPingRespPacket)
- {
- throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only.");
- }
- else if (currentPacket is MqttDisconnectPacket disconnectPacket)
- {
- DisconnectPacket = disconnectPacket;
- return;
- }
- else
- {
- throw new MqttProtocolViolationException("Packet not allowed");
- }
+ if (currentPacket is MqttPublishPacket publishPacket)
+ {
+ await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false);
}
- }
- catch (OperationCanceledException)
- {
- }
- catch (Exception exception)
- {
- if (exception is MqttCommunicationException)
+ else if (currentPacket is MqttPubAckPacket pubAckPacket)
{
- _logger.Warning(exception, "Client '{0}': Communication exception while receiving packets", Id);
- return;
+ await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false);
}
-
- var logLevel = MqttNetLogLevel.Error;
-
- if (!IsRunning)
+ else if (currentPacket is MqttPubCompPacket pubCompPacket)
{
- // There was an exception but the connection is already closed. So there is no chance to send a response to the client etc.
- logLevel = MqttNetLogLevel.Warning;
+ await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false);
}
-
- if (currentPacket == null)
+ else if (currentPacket is MqttPubRecPacket pubRecPacket)
+ {
+ await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false);
+ }
+ else if (currentPacket is MqttPubRelPacket pubRelPacket)
+ {
+ HandleIncomingPubRelPacket(pubRelPacket);
+ }
+ else if (currentPacket is MqttSubscribePacket subscribePacket)
{
- _logger.Publish(logLevel, exception, "Client '{0}': Error while receiving packets", Id);
+ await HandleIncomingSubscribePacket(subscribePacket, cancellationToken).ConfigureAwait(false);
+ }
+ else if (currentPacket is MqttUnsubscribePacket unsubscribePacket)
+ {
+ await HandleIncomingUnsubscribePacket(unsubscribePacket, cancellationToken).ConfigureAwait(false);
+ }
+ else if (currentPacket is MqttPingReqPacket)
+ {
+ HandleIncomingPingReqPacket();
+ }
+ else if (currentPacket is MqttPingRespPacket)
+ {
+ throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only.");
+ }
+ else if (currentPacket is MqttDisconnectPacket disconnectPacket)
+ {
+ DisconnectPacket = disconnectPacket;
+ return;
}
else
{
- _logger.Publish(logLevel, exception, "Client '{0}': Error while processing {1} packet", Id, currentPacket.GetRfcName());
+ throw new MqttProtocolViolationException("Packet not allowed");
}
}
}
-
- async Task SendPacketsLoop(CancellationToken cancellationToken)
+ catch (OperationCanceledException)
{
- MqttPacketBusItem packetBusItem = null;
-
- try
+ }
+ catch (Exception exception)
+ {
+ if (exception is MqttCommunicationException)
{
- while (!cancellationToken.IsCancellationRequested && !IsTakenOver && IsRunning)
- {
- packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
+ _logger.Warning(exception, "Client '{0}': Communication exception while receiving packets", Id);
+ return;
+ }
- // Also check the cancellation token here because the dequeue is blocking and may take some time.
- if (cancellationToken.IsCancellationRequested)
- {
- return;
- }
+ var logLevel = MqttNetLogLevel.Error;
- if (IsTakenOver || !IsRunning)
- {
- return;
- }
+ if (!IsRunning)
+ {
+ // There was an exception but the connection is already closed. So there is no chance to send a response to the client etc.
+ logLevel = MqttNetLogLevel.Warning;
+ }
- try
- {
- await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
- packetBusItem.Complete();
- }
- catch (OperationCanceledException)
- {
- packetBusItem.Cancel();
- }
- catch (Exception exception)
- {
- packetBusItem.Fail(exception);
- }
- finally
- {
- await Task.Yield();
- }
- }
+ if (currentPacket == null)
+ {
+ _logger.Publish(logLevel, exception, "Client '{0}': Error while receiving packets", Id);
}
- catch (OperationCanceledException)
+ else
{
+ _logger.Publish(logLevel, exception, "Client '{0}': Error while processing {1} packet", Id, currentPacket.GetRfcName());
}
- catch (Exception exception)
+ }
+ }
+
+ async Task SendPacketsLoop(CancellationToken cancellationToken)
+ {
+ MqttPacketBusItem packetBusItem = null;
+
+ try
+ {
+ while (!cancellationToken.IsCancellationRequested && !IsTakenOver && IsRunning)
{
- if (exception is MqttCommunicationTimedOutException)
+ packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
+
+ // Also check the cancellation token here because the dequeue is blocking and may take some time.
+ if (cancellationToken.IsCancellationRequested)
{
- _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to timeout", Id);
+ return;
}
- else if (exception is MqttCommunicationException)
+
+ if (IsTakenOver || !IsRunning)
{
- _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to communication exception", Id);
+ return;
}
- else
+
+ try
{
- _logger.Error(exception, "Client '{0}': Sending PUBLISH packet failed", Id);
+ await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
+ packetBusItem.Complete();
}
-
- if (packetBusItem?.Packet is MqttPublishPacket publishPacket)
+ catch (OperationCanceledException)
{
- if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
- {
- publishPacket.Dup = true;
- Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
- }
+ packetBusItem.Cancel();
+ }
+ catch (Exception exception)
+ {
+ packetBusItem.Fail(exception);
+ }
+ finally
+ {
+ await Task.Yield();
}
-
- StopInternal();
}
}
-
- void StopInternal()
+ catch (OperationCanceledException)
{
- _cancellationToken?.TryCancel();
}
-
- async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options)
+ catch (Exception exception)
{
- try
+ if (exception is MqttCommunicationTimedOutException)
{
- // This also indicates that it was tried at least!
- _disconnectPacketSent = true;
-
- var disconnectPacket = MqttDisconnectPacketFactory.Create(options);
+ _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to timeout", Id);
+ }
+ else if (exception is MqttCommunicationException)
+ {
+ _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to communication exception", Id);
+ }
+ else
+ {
+ _logger.Error(exception, "Client '{0}': Sending PUBLISH packet failed", Id);
+ }
- using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
+ if (packetBusItem?.Packet is MqttPublishPacket publishPacket)
+ {
+ if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
- await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);
+ publishPacket.Dup = true;
+ Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
- catch (Exception exception)
+
+ StopInternal();
+ }
+ }
+
+ void StopInternal()
+ {
+ _cancellationToken?.TryCancel();
+ }
+
+ async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options)
+ {
+ try
+ {
+ // This also indicates that it was tried at least!
+ _disconnectPacketSent = true;
+
+ var disconnectPacket = MqttDisconnectPacketFactory.Create(options);
+
+ using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{
- _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
+ await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);
}
}
+ catch (Exception exception)
+ {
+ _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
+ }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
index b6acbb776..a62aacc55 100644
--- a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
@@ -11,744 +11,743 @@
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-using MQTTnet.Server.Disconnecting;
-using MQTTnet.Server.Formatter;
-using MqttPublishPacketFactory = MQTTnet.Server.Formatter.MqttPublishPacketFactory;
+using MQTTnet.Server.Internal.Formatter;
+using MqttPublishPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPublishPacketFactory;
-namespace MQTTnet.Server
+namespace MQTTnet.Server.Internal;
+
+public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification, IDisposable
{
- public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification, IDisposable
- {
- readonly Dictionary _clients = new Dictionary(4096);
+ readonly Dictionary _clients = new(4096);
- readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
+ readonly AsyncLock _createConnectionSyncRoot = new();
- readonly MqttServerEventContainer _eventContainer;
- readonly MqttNetSourceLogger _logger;
- readonly MqttServerOptions _options;
+ readonly MqttServerEventContainer _eventContainer;
+ readonly MqttNetSourceLogger _logger;
+ readonly MqttServerOptions _options;
- readonly MqttRetainedMessagesManager _retainedMessagesManager;
- readonly IMqttNetLogger _rootLogger;
+ readonly MqttRetainedMessagesManager _retainedMessagesManager;
+ readonly IMqttNetLogger _rootLogger;
- readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
+ readonly ReaderWriterLockSlim _sessionsManagementLock = new();
- // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
- // See the MqttSubscription object for a detailed explanation.
- readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
- readonly HashSet _subscriberSessions = new HashSet();
+ // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
+ // See the MqttSubscription object for a detailed explanation.
+ readonly MqttSessionsStorage _sessionsStorage = new();
+ readonly HashSet _subscriberSessions = new();
- public MqttClientSessionsManager(
- MqttServerOptions options,
- MqttRetainedMessagesManager retainedMessagesManager,
- MqttServerEventContainer eventContainer,
- IMqttNetLogger logger)
+ public MqttClientSessionsManager(MqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServerEventContainer eventContainer, IMqttNetLogger logger)
+ {
+ if (logger == null)
{
- if (logger == null)
- {
- throw new ArgumentNullException(nameof(logger));
- }
+ throw new ArgumentNullException(nameof(logger));
+ }
- _logger = logger.WithSource(nameof(MqttClientSessionsManager));
- _rootLogger = logger;
+ _logger = logger.WithSource(nameof(MqttClientSessionsManager));
+ _rootLogger = logger;
- _options = options ?? throw new ArgumentNullException(nameof(options));
- _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
- _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
- }
+ _options = options ?? throw new ArgumentNullException(nameof(options));
+ _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
+ _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
+ }
- public async Task CloseAllConnections(MqttServerClientDisconnectOptions options)
+ public async Task CloseAllConnections(MqttServerClientDisconnectOptions options)
+ {
+ if (options == null)
{
- if (options == null)
- {
- throw new ArgumentNullException(nameof(options));
- }
+ throw new ArgumentNullException(nameof(options));
+ }
- List connections;
- lock (_clients)
- {
- connections = _clients.Values.ToList();
- _clients.Clear();
- }
+ List connections;
+ lock (_clients)
+ {
+ connections = _clients.Values.ToList();
+ _clients.Clear();
+ }
- foreach (var connection in connections)
- {
- await connection.StopAsync(options).ConfigureAwait(false);
- }
+ foreach (var connection in connections)
+ {
+ await connection.StopAsync(options).ConfigureAwait(false);
}
+ }
+
+ public async Task DeleteSessionAsync(string clientId)
+ {
+ _logger.Verbose("Deleting session for client '{0}'.", clientId);
- public async Task DeleteSessionAsync(string clientId)
+ MqttClient connection;
+ lock (_clients)
{
- _logger.Verbose("Deleting session for client '{0}'.", clientId);
+ _clients.TryGetValue(clientId, out connection);
+ }
- MqttClient connection;
- lock (_clients)
+ MqttSession session;
+ _sessionsManagementLock.EnterWriteLock();
+ try
+ {
+ if (_sessionsStorage.TryRemoveSession(clientId, out session))
{
- _clients.TryGetValue(clientId, out connection);
+ _subscriberSessions.Remove(session);
}
+ }
+ finally
+ {
+ _sessionsManagementLock.ExitWriteLock();
+ }
- MqttSession session;
- _sessionsManagementLock.EnterWriteLock();
- try
- {
- if (_sessionsStorage.TryRemoveSession(clientId, out session))
- {
- _subscriberSessions.Remove(session);
- }
- }
- finally
+ try
+ {
+ if (connection != null)
{
- _sessionsManagementLock.ExitWriteLock();
+ await connection.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.NormalDisconnection }).ConfigureAwait(false);
}
+ }
+ catch (Exception exception)
+ {
+ _logger.Error(exception, "Error while deleting session '{0}'", clientId);
+ }
- try
- {
- if (connection != null)
- {
- await connection.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.NormalDisconnection }).ConfigureAwait(false);
- }
- }
- catch (Exception exception)
+ try
+ {
+ if (_eventContainer.SessionDeletedEvent.HasHandlers && session != null)
{
- _logger.Error(exception, "Error while deleting session '{0}'", clientId);
+ var eventArgs = new SessionDeletedEventArgs(clientId, session.Items);
+ await _eventContainer.SessionDeletedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
}
+ }
+ catch (Exception exception)
+ {
+ _logger.Error(exception, "Error while executing session deleted event for session '{0}'", clientId);
+ }
- try
- {
- if (_eventContainer.SessionDeletedEvent.HasHandlers && session != null)
- {
- var eventArgs = new SessionDeletedEventArgs(clientId, session.Items);
- await _eventContainer.SessionDeletedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
- }
- }
- catch (Exception exception)
+ session?.Dispose();
+
+ _logger.Verbose("Session of client '{0}' deleted", clientId);
+ }
+
+ public async Task DispatchApplicationMessage(
+ string senderId,
+ IDictionary senderSessionItems,
+ MqttApplicationMessage applicationMessage,
+ CancellationToken cancellationToken)
+ {
+ var processPublish = true;
+ var closeConnection = false;
+ string reasonString = null;
+ List userProperties = null;
+ var reasonCode = 0; // The reason code is later converted into several different but compatible enums!
+
+ // Allow the user to intercept application message...
+ if (_eventContainer.InterceptingPublishEvent.HasHandlers)
+ {
+ var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, senderId, senderSessionItems);
+ if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
{
- _logger.Error(exception, "Error while executing session deleted event for session '{0}'", clientId);
+ // This can happen if a topic alias us used but the topic is
+ // unknown to the server.
+ interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid;
+ interceptingPublishEventArgs.ProcessPublish = false;
}
- session?.Dispose();
+ await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
- _logger.Verbose("Session of client '{0}' deleted", clientId);
+ applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
+ closeConnection = interceptingPublishEventArgs.CloseConnection;
+ processPublish = interceptingPublishEventArgs.ProcessPublish;
+ reasonString = interceptingPublishEventArgs.Response.ReasonString;
+ userProperties = interceptingPublishEventArgs.Response.UserProperties;
+ reasonCode = (int)interceptingPublishEventArgs.Response.ReasonCode;
}
- public async Task DispatchApplicationMessage(
- string senderId,
- IDictionary senderSessionItems,
- MqttApplicationMessage applicationMessage,
- CancellationToken cancellationToken)
+ // Process the application message...
+ if (processPublish && applicationMessage != null)
{
- var processPublish = true;
- var closeConnection = false;
- string reasonString = null;
- List userProperties = null;
- var reasonCode = 0; // The reason code is later converted into several different but compatible enums!
-
- // Allow the user to intercept application message...
- if (_eventContainer.InterceptingPublishEvent.HasHandlers)
+ var matchingSubscribersCount = 0;
+ try
{
- var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, senderId, senderSessionItems);
- if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
+ if (applicationMessage.Retain)
{
- // This can happen if a topic alias us used but the topic is
- // unknown to the server.
- interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid;
- interceptingPublishEventArgs.ProcessPublish = false;
+ await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
}
- await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
+ List subscriberSessions;
+ _sessionsManagementLock.EnterReadLock();
+ try
+ {
+ subscriberSessions = _subscriberSessions.ToList();
+ }
+ finally
+ {
+ _sessionsManagementLock.ExitReadLock();
+ }
- applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
- closeConnection = interceptingPublishEventArgs.CloseConnection;
- processPublish = interceptingPublishEventArgs.ProcessPublish;
- reasonString = interceptingPublishEventArgs.Response.ReasonString;
- userProperties = interceptingPublishEventArgs.Response.UserProperties;
- reasonCode = (int)interceptingPublishEventArgs.Response.ReasonCode;
- }
+ // Calculate application message topic hash once for subscription checks
+ MqttTopicHash.Calculate(applicationMessage.Topic, out var topicHash, out _, out _);
- // Process the application message...
- if (processPublish && applicationMessage != null)
- {
- var matchingSubscribersCount = 0;
- try
+ foreach (var session in subscriberSessions)
{
- if (applicationMessage.Retain)
+ if (!session.TryCheckSubscriptions(applicationMessage.Topic, topicHash, applicationMessage.QualityOfServiceLevel, senderId, out var checkSubscriptionsResult))
{
- await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
+ // Checking the subscriptions has failed for the session. The session
+ // will be ignored.
+ continue;
}
- List subscriberSessions;
- _sessionsManagementLock.EnterReadLock();
- try
- {
- subscriberSessions = _subscriberSessions.ToList();
- }
- finally
+ if (!checkSubscriptionsResult.IsSubscribed)
{
- _sessionsManagementLock.ExitReadLock();
+ continue;
}
- // Calculate application message topic hash once for subscription checks
- MqttTopicHash.Calculate(applicationMessage.Topic, out var topicHash, out _, out _);
-
- foreach (var session in subscriberSessions)
+ if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
{
- if (!session.TryCheckSubscriptions(
- applicationMessage.Topic,
- topicHash,
- applicationMessage.QualityOfServiceLevel,
- senderId,
- out var checkSubscriptionsResult))
- {
- // Checking the subscriptions has failed for the session. The session
- // will be ignored.
- continue;
- }
+ var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
+ await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
- if (!checkSubscriptionsResult.IsSubscribed)
+ if (!eventArgs.AcceptEnqueue)
{
+ // Continue checking the other subscriptions
continue;
}
+ }
- if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
- {
- var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
- await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-
- if (!eventArgs.AcceptEnqueue)
- {
- // Continue checking the other subscriptions
- continue;
- }
- }
-
- var publishPacketCopy = MqttPacketFactories.Publish.Create(applicationMessage);
- publishPacketCopy.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
- publishPacketCopy.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
-
- if (publishPacketCopy.QualityOfServiceLevel > 0)
- {
- publishPacketCopy.PacketIdentifier = session.PacketIdentifierProvider.GetNextPacketIdentifier();
- }
-
- if (checkSubscriptionsResult.RetainAsPublished)
- {
- // Transfer the original retain state from the publisher. This is a MQTTv5 feature.
- publishPacketCopy.Retain = applicationMessage.Retain;
- }
- else
- {
- publishPacketCopy.Retain = false;
- }
+ var publishPacketCopy = MqttPublishPacketFactory.Create(applicationMessage);
+ publishPacketCopy.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
+ publishPacketCopy.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
- matchingSubscribersCount++;
+ if (publishPacketCopy.QualityOfServiceLevel > 0)
+ {
+ publishPacketCopy.PacketIdentifier = session.PacketIdentifierProvider.GetNextPacketIdentifier();
+ }
- var result = session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
+ if (checkSubscriptionsResult.RetainAsPublished)
+ {
+ // Transfer the original retain state from the publisher. This is a MQTTv5 feature.
+ publishPacketCopy.Retain = applicationMessage.Retain;
+ }
+ else
+ {
+ publishPacketCopy.Retain = false;
+ }
- if (_eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.HasHandlers)
- {
- var eventArgs = new ApplicationMessageEnqueuedEventArgs(senderId, session.Id, applicationMessage, result == EnqueueDataPacketResult.Dropped);
- await _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
- }
+ matchingSubscribersCount++;
- _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'", session.Id, applicationMessage.Topic);
- }
+ var result = session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
- if (matchingSubscribersCount == 0)
+ if (_eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.HasHandlers)
{
- if (reasonCode == (int)MqttPubAckReasonCode.Success)
- {
- // Only change the value if it was success. Otherwise, we would hide an error or not authorized status.
- reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
- }
-
- await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
+ var eventArgs = new ApplicationMessageEnqueuedEventArgs(senderId, session.Id, applicationMessage, result == EnqueueDataPacketResult.Dropped);
+ await _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
+
+ _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'", session.Id, applicationMessage.Topic);
}
- catch (Exception exception)
+
+ if (matchingSubscribersCount == 0)
{
- _logger.Error(exception, "Error while processing next queued application message");
+ if (reasonCode == (int)MqttPubAckReasonCode.Success)
+ {
+ // Only change the value if it was success. Otherwise, we would hide an error or not authorized status.
+ reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
+ }
+
+ await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
}
}
-
- return new DispatchApplicationMessageResult(reasonCode, closeConnection, reasonString, userProperties);
+ catch (Exception exception)
+ {
+ _logger.Error(exception, "Error while processing next queued application message");
+ }
}
- public void Dispose()
- {
- _createConnectionSyncRoot.Dispose();
+ return new DispatchApplicationMessageResult(reasonCode, closeConnection, reasonString, userProperties);
+ }
- _sessionsManagementLock.EnterWriteLock();
- try
- {
- _sessionsStorage.Dispose();
- }
- finally
- {
- _sessionsManagementLock.ExitWriteLock();
- }
+ public void Dispose()
+ {
+ _createConnectionSyncRoot.Dispose();
- _sessionsManagementLock?.Dispose();
+ _sessionsManagementLock.EnterWriteLock();
+ try
+ {
+ _sessionsStorage.Dispose();
}
-
- public MqttClient GetClient(string id)
+ finally
{
- lock (_clients)
- {
- if (!_clients.TryGetValue(id, out var client))
- {
- throw new InvalidOperationException($"Client with ID '{id}' not found.");
- }
-
- return client;
- }
+ _sessionsManagementLock.ExitWriteLock();
}
- public List GetClients()
+ _sessionsManagementLock?.Dispose();
+ }
+
+ public MqttClient GetClient(string id)
+ {
+ lock (_clients)
{
- lock (_clients)
+ if (!_clients.TryGetValue(id, out var client))
{
- return _clients.Values.ToList();
+ throw new InvalidOperationException($"Client with ID '{id}' not found.");
}
+
+ return client;
}
+ }
- public Task> GetClientsStatus()
+ public List GetClients()
+ {
+ lock (_clients)
{
- var result = new List();
+ return _clients.Values.ToList();
+ }
+ }
+
+ public Task> GetClientsStatus()
+ {
+ var result = new List();
- lock (_clients)
+ lock (_clients)
+ {
+ foreach (var client in _clients.Values)
{
- foreach (var client in _clients.Values)
+ var clientStatus = new MqttClientStatus(client)
{
- var clientStatus = new MqttClientStatus(client)
- {
- Session = new MqttSessionStatus(client.Session)
- };
+ Session = new MqttSessionStatus(client.Session)
+ };
- result.Add(clientStatus);
- }
+ result.Add(clientStatus);
}
-
- return Task.FromResult((IList)result);
}
- public Task> GetSessionsStatus()
- {
- var result = new List();
+ return Task.FromResult((IList)result);
+ }
- _sessionsManagementLock.EnterReadLock();
- try
- {
- foreach (var session in _sessionsStorage.ReadAllSessions())
- {
- var sessionStatus = new MqttSessionStatus(session);
- result.Add(sessionStatus);
- }
- }
- finally
+ public Task> GetSessionsStatus()
+ {
+ var result = new List();
+
+ _sessionsManagementLock.EnterReadLock();
+ try
+ {
+ foreach (var session in _sessionsStorage.ReadAllSessions())
{
- _sessionsManagementLock.ExitReadLock();
+ var sessionStatus = new MqttSessionStatus(session);
+ result.Add(sessionStatus);
}
-
- return Task.FromResult((IList)result);
}
-
- public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+ finally
{
- MqttClient client = null;
+ _sessionsManagementLock.ExitReadLock();
+ }
- try
- {
- var connectPacket = await ReceiveConnectPacket(channelAdapter, cancellationToken).ConfigureAwait(false);
- if (connectPacket == null)
- {
- // Nothing was received in time etc.
- return;
- }
+ return Task.FromResult((IList)result);
+ }
- var validatingConnectionEventArgs = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false);
- var connAckPacket = MqttConnAckPacketFactory.Create(validatingConnectionEventArgs);
+ public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+ {
+ MqttClient client = null;
- if (validatingConnectionEventArgs.ReasonCode != MqttConnectReasonCode.Success)
- {
- // Send failure response here without preparing a connection and session!
- await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
- return;
- }
+ try
+ {
+ var connectPacket = await ReceiveConnectPacket(channelAdapter, cancellationToken).ConfigureAwait(false);
+ if (connectPacket == null)
+ {
+ // Nothing was received in time etc.
+ return;
+ }
- // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists.
- client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, validatingConnectionEventArgs).ConfigureAwait(false);
+ var validatingConnectionEventArgs = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false);
+ var connAckPacket = MqttConnAckPacketFactory.Create(validatingConnectionEventArgs);
- await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
+ if (validatingConnectionEventArgs.ReasonCode != MqttConnectReasonCode.Success)
+ {
+ // Send failure response here without preparing a connection and session!
+ await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
+ return;
+ }
- if (_eventContainer.ClientConnectedEvent.HasHandlers)
- {
- var eventArgs = new ClientConnectedEventArgs(
- connectPacket,
- channelAdapter.PacketFormatterAdapter.ProtocolVersion,
- channelAdapter.Endpoint,
- client.Session.Items);
+ // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists.
+ client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, validatingConnectionEventArgs).ConfigureAwait(false);
- await _eventContainer.ClientConnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
- }
+ await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
- await client.RunAsync().ConfigureAwait(false);
- }
- catch (ObjectDisposedException)
- {
- }
- catch (OperationCanceledException)
- {
- }
- catch (Exception exception)
+ if (_eventContainer.ClientConnectedEvent.HasHandlers)
{
- _logger.Error(exception, exception.Message);
+ var eventArgs = new ClientConnectedEventArgs(connectPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion, channelAdapter.Endpoint, client.Session.Items);
+
+ await _eventContainer.ClientConnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
}
- finally
+
+ await client.RunAsync().ConfigureAwait(false);
+ }
+ catch (ObjectDisposedException)
+ {
+ }
+ catch (OperationCanceledException)
+ {
+ }
+ catch (Exception exception)
+ {
+ _logger.Error(exception, exception.Message);
+ }
+ finally
+ {
+ if (client != null)
{
- if (client != null)
+ if (client.Id != null)
{
- if (client.Id != null)
+ // in case it is a takeover _clientConnections already contains the new connection
+ if (!client.IsTakenOver)
{
- // in case it is a takeover _clientConnections already contains the new connection
- if (!client.IsTakenOver)
+ lock (_clients)
{
- lock (_clients)
- {
- _clients.Remove(client.Id);
- }
-
- if (!_options.EnablePersistentSessions || !client.Session.IsPersistent)
- {
- await DeleteSessionAsync(client.Id).ConfigureAwait(false);
- }
+ _clients.Remove(client.Id);
}
- }
-
- var endpoint = client.Endpoint;
-
- if (client.Id != null && !client.IsTakenOver && _eventContainer.ClientDisconnectedEvent.HasHandlers)
- {
- var disconnectType = client.DisconnectPacket != null ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean;
- var eventArgs = new ClientDisconnectedEventArgs(client.Id, client.DisconnectPacket, disconnectType, endpoint, client.Session.Items);
- await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+ if (!_options.EnablePersistentSessions || !ShouldPersistSession(client))
+ {
+ await DeleteSessionAsync(client.Id).ConfigureAwait(false);
+ }
}
}
- using (var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
- {
- await channelAdapter.DisconnectAsync(timeout.Token).ConfigureAwait(false);
- }
- }
- }
+ var endpoint = client.Endpoint;
- public void OnSubscriptionsAdded(MqttSession clientSession, List topics)
- {
- _sessionsManagementLock.EnterWriteLock();
- try
- {
- if (!clientSession.HasSubscribedTopics)
+ if (client.Id != null && !client.IsTakenOver && _eventContainer.ClientDisconnectedEvent.HasHandlers)
{
- // first subscribed topic
- _subscriberSessions.Add(clientSession);
- }
+ var disconnectType = client.DisconnectPacket != null ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean;
+ var eventArgs = new ClientDisconnectedEventArgs(client.Id, client.DisconnectPacket, disconnectType, endpoint, client.Session.Items);
- foreach (var topic in topics)
- {
- clientSession.AddSubscribedTopic(topic);
+ await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
- finally
+
+ using (var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
{
- _sessionsManagementLock.ExitWriteLock();
+ await channelAdapter.DisconnectAsync(timeout.Token).ConfigureAwait(false);
}
}
+ }
- public void OnSubscriptionsRemoved(MqttSession clientSession, List subscriptionTopics)
+ public void OnSubscriptionsAdded(MqttSession clientSession, List topics)
+ {
+ _sessionsManagementLock.EnterWriteLock();
+ try
{
- _sessionsManagementLock.EnterWriteLock();
- try
+ if (!clientSession.HasSubscribedTopics)
{
- foreach (var subscriptionTopic in subscriptionTopics)
- {
- clientSession.RemoveSubscribedTopic(subscriptionTopic);
- }
-
- if (!clientSession.HasSubscribedTopics)
- {
- // last subscription removed
- _subscriberSessions.Remove(clientSession);
- }
+ // first subscribed topic
+ _subscriberSessions.Add(clientSession);
}
- finally
+
+ foreach (var topic in topics)
{
- _sessionsManagementLock.ExitWriteLock();
+ clientSession.AddSubscribedTopic(topic);
}
}
-
- public void Start()
+ finally
{
- if (!_options.EnablePersistentSessions)
- {
- _sessionsStorage.Clear();
- }
+ _sessionsManagementLock.ExitWriteLock();
}
+ }
- public async Task SubscribeAsync(string clientId, ICollection topicFilters)
+ public void OnSubscriptionsRemoved(MqttSession clientSession, List subscriptionTopics)
+ {
+ _sessionsManagementLock.EnterWriteLock();
+ try
{
- if (clientId == null)
+ foreach (var subscriptionTopic in subscriptionTopics)
{
- throw new ArgumentNullException(nameof(clientId));
+ clientSession.RemoveSubscribedTopic(subscriptionTopic);
}
- if (topicFilters == null)
+ if (!clientSession.HasSubscribedTopics)
{
- throw new ArgumentNullException(nameof(topicFilters));
+ // last subscription removed
+ _subscriberSessions.Remove(clientSession);
}
+ }
+ finally
+ {
+ _sessionsManagementLock.ExitWriteLock();
+ }
+ }
- var fakeSubscribePacket = new MqttSubscribePacket();
- fakeSubscribePacket.TopicFilters.AddRange(topicFilters);
-
- var clientSession = GetClientSession(clientId);
-
- var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
+ public void Start()
+ {
+ if (!_options.EnablePersistentSessions)
+ {
+ _sessionsStorage.Clear();
+ }
+ }
- if (subscribeResult.RetainedMessages != null)
- {
- foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
- {
- var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
- clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
- }
- }
+ public async Task SubscribeAsync(string clientId, ICollection topicFilters)
+ {
+ if (clientId == null)
+ {
+ throw new ArgumentNullException(nameof(clientId));
}
- public Task UnsubscribeAsync(string clientId, ICollection topicFilters)
+ if (topicFilters == null)
{
- if (clientId == null)
- {
- throw new ArgumentNullException(nameof(clientId));
- }
+ throw new ArgumentNullException(nameof(topicFilters));
+ }
- if (topicFilters == null)
- {
- throw new ArgumentNullException(nameof(topicFilters));
- }
+ var fakeSubscribePacket = new MqttSubscribePacket();
+ fakeSubscribePacket.TopicFilters.AddRange(topicFilters);
+
+ var clientSession = GetClientSession(clientId);
- var fakeUnsubscribePacket = new MqttUnsubscribePacket();
- fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters);
+ var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
- return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
+ if (subscribeResult.RetainedMessages != null)
+ {
+ foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
+ {
+ var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+ clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
+ }
}
+ }
- MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
+ public Task UnsubscribeAsync(string clientId, ICollection topicFilters)
+ {
+ if (clientId == null)
{
- return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger);
+ throw new ArgumentNullException(nameof(clientId));
}
- async Task CreateClientConnection(
- MqttConnectPacket connectPacket,
- MqttConnAckPacket connAckPacket,
- IMqttChannelAdapter channelAdapter,
- ValidatingConnectionEventArgs validatingConnectionEventArgs)
+ if (topicFilters == null)
{
- MqttClient client;
+ throw new ArgumentNullException(nameof(topicFilters));
+ }
- bool sessionShouldPersist;
+ var fakeUnsubscribePacket = new MqttUnsubscribePacket();
+ fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters);
- if (validatingConnectionEventArgs.ProtocolVersion == MqttProtocolVersion.V500)
- {
- // MQTT 5.0 section 3.1.2.11.2
- // The Client and Server MUST store the Session State after the Network Connection is closed if the Session Expiry Interval is greater than 0 [MQTT-3.1.2-23].
- //
- // A Client that only wants to process messages while connected will set the Clean Start to 1 and set the Session Expiry Interval to 0.
- // It will not receive Application Messages published before it connected and has to subscribe afresh to any topics that it is interested
- // in each time it connects.
+ return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
+ }
- // Persist if SessionExpiryInterval != 0, but may start with a clean session
- sessionShouldPersist = validatingConnectionEventArgs.SessionExpiryInterval != 0;
- }
- else
- {
- // MQTT 3.1.1 section 3.1.2.4: persist only if 'not CleanSession'
- //
- // If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
- // This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be
- // reused in any subsequent Session [MQTT-3.1.2-6].
+ MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
+ {
+ return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger);
+ }
- sessionShouldPersist = !connectPacket.CleanSession;
- }
+ async Task CreateClientConnection(
+ MqttConnectPacket connectPacket,
+ MqttConnAckPacket connAckPacket,
+ IMqttChannelAdapter channelAdapter,
+ ValidatingConnectionEventArgs validatingConnectionEventArgs)
+ {
+ MqttClient client;
+
+ using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false))
+ {
+ MqttSession oldSession;
+ MqttClient oldClient;
- using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false))
+ _sessionsManagementLock.EnterWriteLock();
+ try
{
- MqttSession oldSession;
- MqttClient oldClient;
+ MqttSession session;
- _sessionsManagementLock.EnterWriteLock();
- try
+ // Create a new session (if required).
+ if (!_sessionsStorage.TryGetSession(connectPacket.ClientId, out oldSession))
{
- MqttSession session;
-
- // Create a new session (if required).
- if (!_sessionsStorage.TryGetSession(connectPacket.ClientId, out oldSession))
+ session = CreateSession(connectPacket, validatingConnectionEventArgs);
+ }
+ else
+ {
+ if (connectPacket.CleanSession)
{
- session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist);
+ _logger.Verbose("Deleting existing session of client '{0}' due to clean start", connectPacket.ClientId);
+ _subscriberSessions.Remove(oldSession);
+ session = CreateSession(connectPacket, validatingConnectionEventArgs);
}
else
{
- if (connectPacket.CleanSession)
- {
- _logger.Verbose("Deleting existing session of client '{0}' due to clean start", connectPacket.ClientId);
- _subscriberSessions.Remove(oldSession);
- session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist);
- }
- else
- {
- _logger.Verbose("Reusing existing session of client '{0}'", connectPacket.ClientId);
- session = oldSession;
- oldSession = null;
+ _logger.Verbose("Reusing existing session of client '{0}'", connectPacket.ClientId);
+ session = oldSession;
+ oldSession = null;
- // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
- session.IsPersistent = sessionShouldPersist;
- session.DisconnectedTimestamp = null;
- session.Recover();
+ session.DisconnectedTimestamp = null;
+ session.Recover();
- connAckPacket.IsSessionPresent = true;
- }
+ connAckPacket.IsSessionPresent = true;
}
+ }
- _sessionsStorage.UpdateSession(connectPacket.ClientId, session);
+ _sessionsStorage.UpdateSession(connectPacket.ClientId, session);
- // Create a new client (always required).
- lock (_clients)
+ // Create a new client (always required).
+ lock (_clients)
+ {
+ _clients.TryGetValue(connectPacket.ClientId, out oldClient);
+ if (oldClient != null)
{
- _clients.TryGetValue(connectPacket.ClientId, out oldClient);
- if (oldClient != null)
- {
- // This will stop the current client from sending and receiving but remains the connection active
- // for a later DISCONNECT packet.
- oldClient.IsTakenOver = true;
- }
-
- client = CreateClient(connectPacket, channelAdapter, session);
- _clients[connectPacket.ClientId] = client;
+ // This will stop the current client from sending and receiving but remains the connection active
+ // for a later DISCONNECT packet.
+ oldClient.IsTakenOver = true;
}
- }
- finally
- {
- _sessionsManagementLock.ExitWriteLock();
- }
- if (!connAckPacket.IsSessionPresent)
- {
- // TODO: This event is not yet final. It can already be used but restoring sessions from storage will be added later!
- var preparingSessionEventArgs = new PreparingSessionEventArgs();
- await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false);
+ client = CreateClient(connectPacket, channelAdapter, session);
+ _clients[connectPacket.ClientId] = client;
}
+ }
+ finally
+ {
+ _sessionsManagementLock.ExitWriteLock();
+ }
- if (oldClient != null)
- {
- // TODO: Consider event here for session takeover to allow manipulation of user properties etc.
- await oldClient.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.SessionTakenOver }).ConfigureAwait(false);
+ if (!connAckPacket.IsSessionPresent)
+ {
+ // TODO: This event is not yet final. It can already be used but restoring sessions from storage will be added later!
+ var preparingSessionEventArgs = new PreparingSessionEventArgs();
+ await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false);
+ }
- if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
- {
- var eventArgs = new ClientDisconnectedEventArgs(oldClient.Id, null, MqttClientDisconnectType.Takeover, oldClient.Endpoint, oldClient.Session.Items);
+ if (oldClient != null)
+ {
+ // TODO: Consider event here for session takeover to allow manipulation of user properties etc.
+ await oldClient.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.SessionTakenOver }).ConfigureAwait(false);
- await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
- }
- }
+ if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
+ {
+ var eventArgs = new ClientDisconnectedEventArgs(oldClient.Id, null, MqttClientDisconnectType.Takeover, oldClient.Endpoint, oldClient.Session.Items);
- oldSession?.Dispose();
+ await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
+ }
}
- return client;
+ oldSession?.Dispose();
}
- MqttSession CreateSession(MqttConnectPacket connectPacket, ValidatingConnectionEventArgs validatingConnectionEventArgs, bool isPersistent)
- {
- _logger.Verbose("Created new session for client '{0}'", connectPacket.ClientId);
+ return client;
+ }
- return new MqttSession(connectPacket, validatingConnectionEventArgs.SessionItems, _options, _eventContainer, _retainedMessagesManager, this)
- {
- IsPersistent = isPersistent
- };
+ MqttSession CreateSession(MqttConnectPacket connectPacket, ValidatingConnectionEventArgs validatingConnectionEventArgs)
+ {
+ _logger.Verbose("Created new session for client '{0}'", connectPacket.ClientId);
+
+ return new MqttSession(connectPacket, validatingConnectionEventArgs.SessionItems, _options, _eventContainer, _retainedMessagesManager, this);
+ }
+
+ async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId)
+ {
+ if (!_eventContainer.ApplicationMessageNotConsumedEvent.HasHandlers)
+ {
+ return;
}
- async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId)
+ var eventArgs = new ApplicationMessageNotConsumedEventArgs(applicationMessage, senderId);
+ await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+ }
+
+ MqttSession GetClientSession(string clientId)
+ {
+ _sessionsManagementLock.EnterReadLock();
+ try
{
- if (!_eventContainer.ApplicationMessageNotConsumedEvent.HasHandlers)
+ if (!_sessionsStorage.TryGetSession(clientId, out var session))
{
- return;
+ throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
}
- var eventArgs = new ApplicationMessageNotConsumedEventArgs(applicationMessage, senderId);
- await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+ return session;
}
+ finally
+ {
+ _sessionsManagementLock.ExitReadLock();
+ }
+ }
- MqttSession GetClientSession(string clientId)
+ async Task ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+ {
+ try
{
- _sessionsManagementLock.EnterReadLock();
- try
+ using (var timeoutToken = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
+ using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, cancellationToken))
{
- if (!_sessionsStorage.TryGetSession(clientId, out var session))
+ var firstPacket = await channelAdapter.ReceivePacketAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
+ if (firstPacket is MqttConnectPacket connectPacket)
{
- throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
+ return connectPacket;
}
-
- return session;
- }
- finally
- {
- _sessionsManagementLock.ExitReadLock();
}
}
+ catch (OperationCanceledException)
+ {
+ _logger.Warning("Client '{0}': Connected but did not sent a CONNECT packet.", channelAdapter.Endpoint);
+ }
+ catch (MqttCommunicationTimedOutException)
+ {
+ _logger.Warning("Client '{0}': Connected but did not sent a CONNECT packet.", channelAdapter.Endpoint);
+ }
- async Task ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+ _logger.Warning("Client '{0}': First received packet was no 'CONNECT' packet [MQTT-3.1.0-1].", channelAdapter.Endpoint);
+ return null;
+ }
+
+ static bool ShouldPersistSession(MqttClient client)
+ {
+ switch (client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion)
{
- try
+ case MqttProtocolVersion.V500:
{
- using (var timeoutToken = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
- using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, cancellationToken))
+ // MQTT 5.0 section 3.1.2.11.2
+ // The Client and Server MUST store the Session State after the Network Connection is closed if the Session Expiry Interval is greater than 0 [MQTT-3.1.2-23].
+ //
+ // A Client that only wants to process messages while connected will set the Clean Start to 1 and set the Session Expiry Interval to 0.
+ // It will not receive Application Messages published before it connected and has to subscribe afresh to any topics that it is interested
+ // in each time it connects.
+
+ var effectiveSessionExpiryInterval = client.DisconnectPacket?.SessionExpiryInterval ?? 0U;
+ if (effectiveSessionExpiryInterval == 0U)
{
- var firstPacket = await channelAdapter.ReceivePacketAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
- if (firstPacket is MqttConnectPacket connectPacket)
- {
- return connectPacket;
- }
+ // From RFC: If the Session Expiry Interval is absent, the Session Expiry Interval in the CONNECT packet is used.
+ effectiveSessionExpiryInterval = client.ConnectPacket.SessionExpiryInterval;
}
+
+ return effectiveSessionExpiryInterval != 0U;
}
- catch (OperationCanceledException)
+
+ case MqttProtocolVersion.V311:
{
- _logger.Warning("Client '{0}': Connected but did not sent a CONNECT packet.", channelAdapter.Endpoint);
+ // MQTT 3.1.1 section 3.1.2.4: persist only if 'not CleanSession'
+ //
+ // If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
+ // This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be
+ // reused in any subsequent Session [MQTT-3.1.2-6].
+
+ return !client.ConnectPacket.CleanSession;
}
- catch (MqttCommunicationTimedOutException)
+
+ case MqttProtocolVersion.V310:
{
- _logger.Warning("Client '{0}': Connected but did not sent a CONNECT packet.", channelAdapter.Endpoint);
+ return true;
}
- _logger.Warning("Client '{0}': First received packet was no 'CONNECT' packet [MQTT-3.1.0-1].", channelAdapter.Endpoint);
- return null;
+ default:
+ throw new NotSupportedException();
}
+ }
- async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
- {
- // TODO: Load session items from persisted sessions in the future.
- var sessionItems = new ConcurrentDictionary