From 13762eedd44c0f31ac5b8535a20cd03c76671e19 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Tue, 19 Jul 2022 17:33:01 +0200 Subject: [PATCH 1/9] Update Server_Asp_Net_Samples.cs --- Samples/Server/Server_Asp_Net_Samples.cs | 43 +++++++++++++++++------- 1 file changed, 30 insertions(+), 13 deletions(-) diff --git a/Samples/Server/Server_Asp_Net_Samples.cs b/Samples/Server/Server_Asp_Net_Samples.cs index a343159e..d085196d 100644 --- a/Samples/Server/Server_Asp_Net_Samples.cs +++ b/Samples/Server/Server_Asp_Net_Samples.cs @@ -5,12 +5,15 @@ // ReSharper disable UnusedType.Global // ReSharper disable UnusedMember.Global // ReSharper disable InconsistentNaming +// ReSharper disable EmptyConstructor +// ReSharper disable MemberCanBeMadeStatic.Local using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using MQTTnet.AspNetCore; +using MQTTnet.Server; namespace MQTTnet.Samples.Server; @@ -30,7 +33,7 @@ public static class Server_Asp_Net_Samples { // This will allow MQTT connections based on TCP port 1883. o.ListenAnyIP(1883, l => l.UseMqtt()); - + // This will allow MQTT connections based on HTTP WebSockets with URI "localhost:5000/mqtt" // See code below for URI configuration. o.ListenAnyIP(5000); // Default HTTP pipeline @@ -42,9 +45,30 @@ public static class Server_Asp_Net_Samples return host.RunConsoleAsync(); } + sealed class MqttController + { + public MqttController() + { + // Inject other services via constructor. + } + + public Task OnClientConnected(ClientConnectedEventArgs eventArgs) + { + Console.WriteLine($"Client '{eventArgs.ClientId}' connected."); + return Task.CompletedTask; + } + + + public Task ValidateConnection(ValidatingConnectionEventArgs eventArgs) + { + Console.WriteLine($"Client '{eventArgs.ClientId}' wants to connect. Accepting!"); + return Task.CompletedTask; + } + } + sealed class Startup { - public void Configure(IApplicationBuilder app, IWebHostEnvironment env) + public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController) { app.UseRouting(); @@ -64,17 +88,8 @@ public static class Server_Asp_Net_Samples * Attach event handlers etc. if required. */ - server.ValidatingConnectionAsync += e => - { - Console.WriteLine($"Client '{e.ClientId}' wants to connect. Accepting!"); - return Task.CompletedTask; - }; - - server.ClientConnectedAsync += e => - { - Console.WriteLine($"Client '{e.ClientId}' connected."); - return Task.CompletedTask; - }; + server.ValidatingConnectionAsync += mqttController.ValidateConnection; + server.ClientConnectedAsync += mqttController.OnClientConnected; }); } @@ -88,6 +103,8 @@ public static class Server_Asp_Net_Samples services.AddMqttConnectionHandler(); services.AddConnections(); + + services.AddSingleton(); } } } \ No newline at end of file From ed7c150cc5ebdae4305d312dc2f4cfe2347d431d Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Tue, 19 Jul 2022 20:47:16 +0200 Subject: [PATCH 2/9] Add new options. --- Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs | 7 +++++++ Source/MQTTnet/Implementations/MqttTcpChannel.cs | 5 +++++ 2 files changed, 12 insertions(+) diff --git a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs index 479bb1c5..879d8ac7 100644 --- a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System.Net; using System.Net.Sockets; namespace MQTTnet.Client @@ -10,6 +11,12 @@ namespace MQTTnet.Client { public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified; + /// + /// Gets the local endpoint (network card) which is used by the client. + /// Set it to _null_ to let the OS select the network card. + /// + public EndPoint LocalEndpoint { get; set; } + public int BufferSize { get; set; } = 8192; public bool? DualMode { get; set; } diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 295cee6b..f1391bc1 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -68,6 +68,11 @@ namespace MQTTnet.Implementations socket = new CrossPlatformSocket(_tcpOptions.AddressFamily); } + if (_tcpOptions.LocalEndpoint != null) + { + socket.Bind(_tcpOptions.LocalEndpoint); + } + socket.ReceiveBufferSize = _tcpOptions.BufferSize; socket.SendBufferSize = _tcpOptions.BufferSize; socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds; From 6c29ea56fce333a3f3930597ece15945ea294aa2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20K=C3=B6pke?= Date: Sat, 23 Jul 2022 10:37:57 +0200 Subject: [PATCH 3/9] Add PackageIdentifier to MqttApplicationMessageReceivedEventArgs (#1466) Provide `PackageIdentifier` as read only property in `MqttApplicationMessageReceivedEventArgs` --- .../MqttApplicationMessageReceivedEventArgs.cs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs index c288ab26..4bf1c17c 100644 --- a/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs +++ b/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs @@ -25,6 +25,7 @@ namespace MQTTnet.Client ClientId = clientId; ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); PublishPacket = publishPacket; + PacketIdentifier = publishPacket.PacketIdentifier; _acknowledgeHandler = acknowledgeHandler; } @@ -62,12 +63,17 @@ namespace MQTTnet.Client public bool IsHandled { get; set; } /// - /// Gets ir sets whether the library should send MQTT ACK packets automatically if required. + /// Gets or sets whether the library should send MQTT ACK packets automatically if required. /// public bool AutoAcknowledge { get; set; } = true; public object Tag { get; set; } - + + /// + /// Gets the indentifier of MQTT packet + /// + public ushort PacketIdentifier { get; } + public Task AcknowledgeAsync(CancellationToken cancellationToken) { if (_acknowledgeHandler == null) From d3848d8cb0d3760cadbff8fcabdf1445dd9f65ad Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 11:00:21 +0200 Subject: [PATCH 4/9] Add comments. --- .github/workflows/ReleaseNotes.md | 3 ++- .../WebSocket4NetMqttChannel.cs | 1 - .../Client/Options/MqttClientOptionsBuilder.cs | 3 +-- .../MQTTnet/Client/Options/MqttClientTcpOptions.cs | 5 +++++ .../Client/Options/MqttClientWebSocketOptions.cs | 14 +++++++------- .../Options/MqttClientWebSocketProxyOptions.cs | 2 +- .../Implementations/MqttWebSocketChannel.cs | 1 - 7 files changed, 16 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 897cf698..e1b28022 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -5,4 +5,5 @@ * [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442). * [Server] Fix cross thread issue in session message storage for QoS 1 and 2. * [Server] The event _ClientSubscribedTopicAsync_ is now fired after the subscription is completely processed internally (#1435). -* [Server] Improved CPU usage on lower end machines (#788). \ No newline at end of file +* [Server] Improved CPU usage on lower end machines (#788). +* [Client] Added support for passing the local endpoint which should be used (network card). \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs index c628f542..47e78eec 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs @@ -98,7 +98,6 @@ namespace MQTTnet.Extensions.WebSocket4Net #else certificates.Add(certificate); #endif - } } diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index d8568770..7e3d2547 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -264,8 +264,7 @@ namespace MQTTnet.Client return this; } - - // TODO: Consider creating _MqttClientTcpOptionsBuilder_ as overload. + public MqttClientOptionsBuilder WithTcpServer(Action optionsBuilder) { if (optionsBuilder == null) diff --git a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs index 879d8ac7..586cab51 100644 --- a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs @@ -19,6 +19,11 @@ namespace MQTTnet.Client public int BufferSize { get; set; } = 8192; + /// + /// Gets or sets whether the underlying socket should run in dual mode. + /// Leaving this _null_ will avoid setting this value at socket level. + /// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine. + /// public bool? DualMode { get; set; } public LingerOption LingerState { get; set; } = new LingerOption(true, 0); diff --git a/Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs b/Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs index 03160a86..e9de2da8 100644 --- a/Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs @@ -7,23 +7,23 @@ using System.Net; namespace MQTTnet.Client { - public class MqttClientWebSocketOptions : IMqttClientChannelOptions + public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions { - public string Uri { get; set; } + public CookieContainer CookieContainer { get; set; } + + public MqttClientWebSocketProxyOptions ProxyOptions { get; set; } public IDictionary RequestHeaders { get; set; } public ICollection SubProtocols { get; set; } = new List { "mqtt" }; - public CookieContainer CookieContainer { get; set; } - - public MqttClientWebSocketProxyOptions ProxyOptions { get; set; } - public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); + + public string Uri { get; set; } public override string ToString() { return Uri; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs b/Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs index 17a6e2de..92968984 100644 --- a/Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs @@ -4,7 +4,7 @@ namespace MQTTnet.Client { - public class MqttClientWebSocketProxyOptions + public sealed class MqttClientWebSocketProxyOptions { public string Address { get; set; } diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index 7a3b4a9f..3a54843d 100644 --- a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs +++ b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs @@ -118,7 +118,6 @@ namespace MQTTnet.Implementations void SetupClientWebSocket(ClientWebSocket clientWebSocket) { - if (_options.ProxyOptions != null) { clientWebSocket.Options.Proxy = CreateProxy(); From a94776eeead86b0a9d147a375623a003bd6faa8e Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 11:02:00 +0200 Subject: [PATCH 5/9] Update version information. --- .github/workflows/ReleaseNotes.md | 8 -------- .github/workflows/ci.yml | 2 +- 2 files changed, 1 insertion(+), 9 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 897cf698..e69de29b 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,8 +0,0 @@ -* [Core] Updated nuget packages. -* [Core] The option _IgnoreCertificateChainErrors_ is now respected (thanks to @GodVenn, #1447). -* [ManagedClient] The managed client now sends the entire topic filter including new MQTTv5 properties when subscribing. -* [ManagedClient] Fixed wrong firing of _ApplicationMessageSkippedAsync_ event (thanks to @quackgizmo, #1460). -* [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442). -* [Server] Fix cross thread issue in session message storage for QoS 1 and 2. -* [Server] The event _ClientSubscribedTopicAsync_ is now fired after the subscription is completely processed internally (#1435). -* [Server] Improved CPU usage on lower end machines (#788). \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 359fb294..cb2c15da 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,7 @@ name: CI on: [push, pull_request] env: - VERSION: "4.0.2.${{github.run_number}}" + VERSION: "4.1.0.${{github.run_number}}" jobs: build: From d71ff3d61591dc3fe62a6d7c691f80e83f548466 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 11:05:14 +0200 Subject: [PATCH 6/9] Update ReleaseNotes.md --- .github/workflows/ReleaseNotes.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index e8811df2..a5254e05 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,2 @@ * [Client] Added support for passing the local endpoint which should be used (network card). +* [Client] Exposed _PackageIdentifier_ in _MqttApplicationMessageReceivedEventArgs_ (thanks to @koepalex, #1466). From 0d7b0209942ce6c83cdb0ed5254caca41d5d08a9 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 11:07:32 +0200 Subject: [PATCH 7/9] Squashed commit of the following: 487-any-way-to-generate-events-on-receipt-of-puback-or-pubrec-packets commit 3f2f8d4b0a31b20734eca3fff86c43bd1ac3e3ad Merge: 34babbf2 d71ff3d6 Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 23 11:06:22 2022 +0200 Merge branch 'master' into 487-any-way-to-generate-events-on-receipt-of-puback-or-pubrec-packets commit 34babbf2714b742bc4f3802804a1a07159d38e22 Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Mon Jul 11 22:56:01 2022 +0200 Fix typo. commit add6277458f528fc03bf4a1ba228a55c5a8d1a80 Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 2 12:50:13 2022 +0200 Add new event for client acknowledges. --- .github/workflows/ReleaseNotes.md | 1 + Source/MQTTnet.Tests/Server/General.cs | 146 +----- Source/MQTTnet.Tests/Server/QoS_Tests.cs | 124 +++++ Source/MQTTnet.Tests/Server/Security_Tests.cs | 2 - Source/MQTTnet.Tests/Server/Session_Tests.cs | 85 +++- Source/MQTTnet.Tests/Server/Will_Tests.cs | 71 +++ ...lientAcknowledgedPublishPacketEventArgs.cs | 38 ++ .../Events/ValidatingConnectionEventArgs.cs | 3 + Source/MQTTnet/Server/Internal/MqttClient.cs | 452 ++++++++++-------- .../Internal/MqttServerEventContainer.cs | 63 +-- Source/MQTTnet/Server/Internal/MqttSession.cs | 13 +- Source/MQTTnet/Server/MqttServer.cs | 6 + 12 files changed, 625 insertions(+), 379 deletions(-) create mode 100644 Source/MQTTnet.Tests/Server/QoS_Tests.cs create mode 100644 Source/MQTTnet.Tests/Server/Will_Tests.cs create mode 100644 Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index a5254e05..7104a49f 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,2 +1,3 @@ * [Client] Added support for passing the local endpoint which should be used (network card). * [Client] Exposed _PackageIdentifier_ in _MqttApplicationMessageReceivedEventArgs_ (thanks to @koepalex, #1466). +* [Server] Added a new event (_ClientAcknowledgedPublishPacketAsync_) which is fired whenever a client acknowledges a PUBLISH packet (QoS 1 and 2, #487). diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index f7bf4a72..2d9cf418 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -798,38 +798,7 @@ namespace MQTTnet.Tests.Server Assert.IsTrue(longBody.SequenceEqual(receivedBody ?? new byte[0])); } } - - [TestMethod] - public async Task Session_Takeover() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var options = new MqttClientOptionsBuilder().WithCleanSession(false) - .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+ - .WithClientId("a"); - - var client1 = await testEnvironment.ConnectClient(options); - await Task.Delay(500); - - var disconnectReason = MqttClientDisconnectReason.NormalDisconnection; - client1.DisconnectedAsync += c => - { - disconnectReason = c.Reason; - return PlatformAbstractionLayer.CompletedTask; - }; - - var client2 = await testEnvironment.ConnectClient(options); - await Task.Delay(500); - - Assert.IsFalse(client1.IsConnected); - Assert.IsTrue(client2.IsConnected); - - Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason); - } - } - + [TestMethod] public async Task Set_Subscription_At_Server() { @@ -944,26 +913,6 @@ namespace MQTTnet.Tests.Server } } - [TestMethod] - public async Task Use_Clean_Session() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var client = testEnvironment.CreateClient(); - var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build()); - - // Create the session including the subscription. - var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false)); - await client1.SubscribeAsync("x"); - await client1.DisconnectAsync(); - await Task.Delay(500); - - Assert.IsFalse(connectResult.IsSessionPresent); - } - } - [TestMethod] public async Task Use_Empty_Client_ID() { @@ -986,98 +935,7 @@ namespace MQTTnet.Tests.Server Assert.IsTrue(client.IsConnected); } } - - [TestMethod] - public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will"); - - var c1 = await testEnvironment.ConnectClient(); - - c1.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); - - var c2 = await testEnvironment.ConnectClient(clientOptions); - await c2.DisconnectAsync().ConfigureAwait(false); - - await Task.Delay(1000); - - Assert.AreEqual(0, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Will_Message_Do_Not_Send_On_Takeover() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - // C1 will receive the last will! - var c1 = await testEnvironment.ConnectClient(); - c1.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); - - // C2 has the last will defined. - var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner"); - - var c2 = await testEnvironment.ConnectClient(clientOptions); - - // C3 will do the connection takeover. - var c3 = await testEnvironment.ConnectClient(clientOptions); - - await Task.Delay(1000); - - Assert.AreEqual(0, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Will_Message_Send() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); - - var receivedMessagesCount = 0; - c1.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); - - var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce); - var c2 = await testEnvironment.ConnectClient(clientOptions); - c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. - - await Task.Delay(1000); - - Assert.AreEqual(1, receivedMessagesCount); - } - } - + void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs) { if (_connected.ContainsKey(eventArgs.ClientId)) diff --git a/Source/MQTTnet.Tests/Server/QoS_Tests.cs b/Source/MQTTnet.Tests/Server/QoS_Tests.cs new file mode 100644 index 00000000..483c8d07 --- /dev/null +++ b/Source/MQTTnet.Tests/Server/QoS_Tests.cs @@ -0,0 +1,124 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Implementations; +using MQTTnet.Protocol; +using MQTTnet.Server; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class QoS_Tests : BaseTestClass + { + [TestMethod] + public async Task Fire_Event_On_Client_Acknowledges_QoS_0() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + ClientAcknowledgedPublishPacketEventArgs eventArgs = null; + server.ClientAcknowledgedPublishPacketAsync += args => + { + eventArgs = args; + return PlatformAbstractionLayer.CompletedTask; + }; + + var client1 = await testEnvironment.ConnectClient(); + await client1.SubscribeAsync("A"); + + var client2 = await testEnvironment.ConnectClient(); + await client2.PublishStringAsync("A"); + + await LongTestDelay(); + + // Must be null because no event should be fired for QoS 0. + Assert.IsNull(eventArgs); + } + } + + [TestMethod] + public async Task Fire_Event_On_Client_Acknowledges_QoS_1() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + ClientAcknowledgedPublishPacketEventArgs eventArgs = null; + server.ClientAcknowledgedPublishPacketAsync += args => + { + eventArgs = args; + return PlatformAbstractionLayer.CompletedTask; + }; + + var client1 = await testEnvironment.ConnectClient(); + await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtLeastOnce); + + var client2 = await testEnvironment.ConnectClient(); + await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await LongTestDelay(); + + Assert.IsNotNull(eventArgs); + Assert.IsNotNull(eventArgs.PublishPacket); + Assert.IsNotNull(eventArgs.AcknowledgePacket); + Assert.IsTrue(eventArgs.IsCompleted); + + Assert.AreEqual("A", eventArgs.PublishPacket.Topic); + } + } + + [TestMethod] + public async Task Fire_Event_On_Client_Acknowledges_QoS_2() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + var eventArgs = new List(); + server.ClientAcknowledgedPublishPacketAsync += args => + { + lock (eventArgs) + { + eventArgs.Add(args); + } + + return PlatformAbstractionLayer.CompletedTask; + }; + + var client1 = await testEnvironment.ConnectClient(); + await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.ExactlyOnce); + + var client2 = await testEnvironment.ConnectClient(); + await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce); + + await LongTestDelay(); + + Assert.AreEqual(2, eventArgs.Count); + + var firstEvent = eventArgs[0]; + + Assert.IsNotNull(firstEvent); + Assert.IsNotNull(firstEvent.PublishPacket); + Assert.IsNotNull(firstEvent.AcknowledgePacket); + Assert.IsFalse(firstEvent.IsCompleted); + + Assert.AreEqual("A", firstEvent.PublishPacket.Topic); + + var secondEvent = eventArgs[1]; + + Assert.IsNotNull(secondEvent); + Assert.IsNotNull(secondEvent.PublishPacket); + Assert.IsNotNull(secondEvent.AcknowledgePacket); + Assert.IsTrue(secondEvent.IsCompleted); + + Assert.AreEqual("A", secondEvent.PublishPacket.Topic); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Server/Security_Tests.cs b/Source/MQTTnet.Tests/Server/Security_Tests.cs index 226752d0..7d2336b0 100644 --- a/Source/MQTTnet.Tests/Server/Security_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Security_Tests.cs @@ -2,10 +2,8 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using System; using System.Collections.Generic; using System.Threading.Tasks; -using Microsoft.CodeAnalysis; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.Client; diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs index b4b9a8ab..a791c4c5 100644 --- a/Source/MQTTnet.Tests/Server/Session_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs @@ -14,7 +14,6 @@ using MQTTnet.Implementations; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; -using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet.Tests.Server { @@ -220,6 +219,37 @@ namespace MQTTnet.Tests.Server } } + [TestMethod] + public async Task Session_Takeover() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var options = new MqttClientOptionsBuilder().WithCleanSession(false) + .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+ + .WithClientId("a"); + + var client1 = await testEnvironment.ConnectClient(options); + await Task.Delay(500); + + var disconnectReason = MqttClientDisconnectReason.NormalDisconnection; + client1.DisconnectedAsync += c => + { + disconnectReason = c.Reason; + return PlatformAbstractionLayer.CompletedTask; + }; + + var client2 = await testEnvironment.ConnectClient(options); + await Task.Delay(500); + + Assert.IsFalse(client1.IsConnected); + Assert.IsTrue(client2.IsConnected); + + Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason); + } + } + [TestMethod] public async Task Set_Session_Item() { @@ -277,6 +307,59 @@ namespace MQTTnet.Tests.Server } } + [TestMethod] + public async Task Use_Clean_Session() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var client = testEnvironment.CreateClient(); + var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build()); + + // Create the session including the subscription. + var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false)); + await client1.SubscribeAsync("x"); + await client1.DisconnectAsync(); + await Task.Delay(500); + + Assert.IsFalse(connectResult.IsSessionPresent); + } + } + + [TestMethod] + public async Task Will_Message_Do_Not_Send_On_Takeover() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + // C1 will receive the last will! + var c1 = await testEnvironment.ConnectClient(); + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return PlatformAbstractionLayer.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + // C2 has the last will defined. + var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner"); + + var c2 = await testEnvironment.ConnectClient(clientOptions); + + // C3 will do the connection takeover. + var c3 = await testEnvironment.ConnectClient(clientOptions); + + await Task.Delay(1000); + + Assert.AreEqual(0, receivedMessagesCount); + } + } + async Task ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive) { try diff --git a/Source/MQTTnet.Tests/Server/Will_Tests.cs b/Source/MQTTnet.Tests/Server/Will_Tests.cs new file mode 100644 index 00000000..d464cd7e --- /dev/null +++ b/Source/MQTTnet.Tests/Server/Will_Tests.cs @@ -0,0 +1,71 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Implementations; +using MQTTnet.Protocol; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class Will_Tests : BaseTestClass + { + [TestMethod] + public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will"); + + var c1 = await testEnvironment.ConnectClient(); + + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return PlatformAbstractionLayer.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + var c2 = await testEnvironment.ConnectClient(clientOptions); + await c2.DisconnectAsync().ConfigureAwait(false); + + await Task.Delay(1000); + + Assert.AreEqual(0, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Will_Message_Send() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); + + var receivedMessagesCount = 0; + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return PlatformAbstractionLayer.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce); + var c2 = await testEnvironment.ConnectClient(clientOptions); + c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. + + await Task.Delay(1000); + + Assert.AreEqual(1, receivedMessagesCount); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs b/Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs new file mode 100644 index 00000000..e0c09ab1 --- /dev/null +++ b/Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs @@ -0,0 +1,38 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections; +using MQTTnet.Packets; + +namespace MQTTnet.Server +{ + public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs + { + /// + /// Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet. + /// + public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; } + + /// + /// Gets the ID of the client which acknowledged a PUBLISH packet. + /// + public string ClientId { get; internal set; } + + /// + /// Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2. + /// + public bool IsCompleted => AcknowledgePacket is MqttPubAckPacket || AcknowledgePacket is MqttPubCompPacket; + + /// + /// Gets the PUBLISH packet which was acknowledged. + /// + public MqttPublishPacket PublishPacket { get; internal set; } + + /// + /// Gets the session items which contain custom user data per session. + /// + public IDictionary SessionItems { get; internal set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index 61262800..8a5b48b6 100644 --- a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -161,7 +161,10 @@ namespace MQTTnet.Server /// public ushort TopicAliasMaximum => _connectPacket.TopicAliasMaximum; + [Obsolete("This property name has a typo. Use 'UserName' instead. This one will be removed soon.")] public string Username => _connectPacket.Username; + + public string UserName => _connectPacket.Username; /// /// Gets or sets the user properties. diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 6d2f59e1..483ff8af 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -11,6 +11,7 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; +using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; @@ -20,20 +21,20 @@ namespace MQTTnet.Server { public sealed class MqttClient { - readonly Dictionary _topicAlias = new Dictionary(); + readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); + readonly MqttConnectPacket _connectPacket; + readonly MqttServerEventContainer _eventContainer; + readonly MqttNetSourceLogger _logger; readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); - readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); - - readonly MqttClientSessionsManager _sessionsManager; - readonly MqttNetSourceLogger _logger; - + readonly MqttServerOptions _serverOptions; - readonly MqttServerEventContainer _eventContainer; - readonly MqttConnectPacket _connectPacket; + + readonly MqttClientSessionsManager _sessionsManager; + readonly Dictionary _topicAlias = new Dictionary(); CancellationTokenSource _cancellationToken; - + public MqttClient( MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, @@ -53,43 +54,31 @@ namespace MQTTnet.Server Session = session ?? throw new ArgumentNullException(nameof(session)); _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + _logger = logger.WithSource(nameof(MqttClient)); } - public string Id => _connectPacket.ClientId; + public IMqttChannelAdapter ChannelAdapter { get; } public string Endpoint { get; } - public IMqttChannelAdapter ChannelAdapter { get; } + public string Id => _connectPacket.ClientId; - public MqttClientStatistics Statistics { get; } = new MqttClientStatistics(); + public bool IsCleanDisconnect { get; private set; } public bool IsRunning { get; private set; } + public bool IsTakenOver { get; set; } + public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod; public MqttSession Session { get; } - public bool IsTakenOver { get; set; } - - public bool IsCleanDisconnect { get; private set; } - - public async Task StopAsync(MqttDisconnectReasonCode reason) - { - IsRunning = false; - - if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout) - { - // 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(reason).ConfigureAwait(false); - } - - StopInternal(); - } + public MqttClientStatistics Statistics { get; } = new MqttClientStatistics(); public void ResetStatistics() { @@ -110,7 +99,7 @@ namespace MQTTnet.Server try { Task.Run(() => SendPacketsLoop(cancellationToken), cancellationToken).RunInBackground(_logger); - + IsRunning = true; await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false); @@ -130,10 +119,10 @@ namespace MQTTnet.Server { var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket); var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket); - - _= _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage); + + _ = _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage); Session.WillMessageSent = true; - + _logger.Info("Client '{0}': Published will message.", Id); } @@ -149,18 +138,233 @@ namespace MQTTnet.Server // This might break the protocol but the user wants that. return; } - + await ChannelAdapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false); Statistics.HandleSentPacket(packet); } + public async Task StopAsync(MqttDisconnectReasonCode reason) + { + IsRunning = false; + + if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout) + { + // 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(reason).ConfigureAwait(false); + } + + StopInternal(); + } + + Task ClientAcknowledgedPublishPacket(MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket) + { + if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers) + { + var eventArgs = new ClientAcknowledgedPublishPacketEventArgs + { + PublishPacket = publishPacket, + AcknowledgePacket = acknowledgePacket, + ClientId = Id, + SessionItems = Session.Items + }; + + return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket) + { + var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier); + + if (acknowledgedPublishPacket != null) + { + return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket) + { + var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier); + + if (acknowledgedPublishPacket != null) + { + return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + HandleTopicAlias(publishPacket); + + InterceptingPublishEventArgs interceptingPublishEventArgs = null; + var applicationMessage = _applicationMessageFactory.Create(publishPacket); + var closeConnection = false; + var processPublish = true; + + if (_eventContainer.InterceptingPublishEvent.HasHandlers) + { + interceptingPublishEventArgs = new InterceptingPublishEventArgs + { + ClientId = Id, + ApplicationMessage = applicationMessage, + SessionItems = Session.Items, + ProcessPublish = true, + CloseConnection = false, + CancellationToken = cancellationToken + }; + + if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic)) + { + // 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 _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); + + applicationMessage = interceptingPublishEventArgs.ApplicationMessage; + closeConnection = interceptingPublishEventArgs.CloseConnection; + processPublish = interceptingPublishEventArgs.ProcessPublish; + } + + if (closeConnection) + { + await StopAsync(MqttDisconnectReasonCode.UnspecifiedError); + return; + } + + if (processPublish && applicationMessage != null) + { + await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false); + } + + switch (publishPacket.QualityOfServiceLevel) + { + case MqttQualityOfServiceLevel.AtMostOnce: + { + // Do nothing since QoS 0 has no ACK at all! + break; + } + case MqttQualityOfServiceLevel.AtLeastOnce: + { + var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket)); + break; + } + case MqttQualityOfServiceLevel.ExactlyOnce: + { + var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket)); + break; + } + default: + { + throw new MqttCommunicationException("Received a not supported QoS level."); + } + } + } + + async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) + { + var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier); + + if (acknowledgedPublishPacket != null) + { + await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false); + } + + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); + } + + void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) + { + var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket)); + } + + async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) + { + var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); + + var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); + + Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket)); + + if (subscribeResult.CloseConnection) + { + StopInternal(); + return; + } + + if (subscribeResult.RetainedMessages != null) + { + foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) + { + var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); + Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); + } + } + } + + async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) + { + var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); + + var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); + + Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket)); + + if (unsubscribeResult.CloseConnection) + { + StopInternal(); + } + } + + void HandleTopicAlias(MqttPublishPacket publishPacket) + { + if (publishPacket.TopicAlias == 0) + { + return; + } + + lock (_topicAlias) + { + if (!string.IsNullOrEmpty(publishPacket.Topic)) + { + _topicAlias[publishPacket.TopicAlias] = publishPacket.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); + } + } + } + } + async Task InterceptPacketAsync(MqttPacket packet, CancellationToken cancellationToken) { if (!_eventContainer.InterceptingOutboundPacketEvent.HasHandlers) { return packet; } - + var interceptingPacketEventArgs = new InterceptingPacketEventArgs { ClientId = Id, @@ -168,14 +372,14 @@ namespace MQTTnet.Server Packet = packet, CancellationToken = cancellationToken }; - + await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); - + if (!interceptingPacketEventArgs.ProcessPacket || packet == null) { return null; } - + return interceptingPacketEventArgs.Packet; } @@ -197,7 +401,7 @@ namespace MQTTnet.Server } var processPacket = true; - + if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers) { var interceptingPacketEventArgs = new InterceptingPacketEventArgs @@ -207,12 +411,12 @@ namespace MQTTnet.Server Packet = packet, CancellationToken = cancellationToken }; - + await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); packet = interceptingPacketEventArgs.Packet; processPacket = interceptingPacketEventArgs.ProcessPacket; } - + if (!processPacket || packet == null) { // Restart the receiving process to get the next packet ignoring the current one.. @@ -227,15 +431,15 @@ namespace MQTTnet.Server } else if (packet is MqttPubAckPacket pubAckPacket) { - Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier); + await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false); } else if (packet is MqttPubCompPacket pubCompPacket) { - Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier); + await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false); } else if (packet is MqttPubRecPacket pubRecPacket) { - HandleIncomingPubRecPacket(pubRecPacket); + await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false); } else if (packet is MqttPubRelPacket pubRelPacket) { @@ -353,7 +557,7 @@ namespace MQTTnet.Server StopInternal(); } } - + void StopInternal() { try @@ -367,162 +571,12 @@ namespace MQTTnet.Server } } - void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) - { - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); - } - - void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) - { - var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket)); - } - - async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) - { - var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); - - var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); - - Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket)); - - if (subscribeResult.CloseConnection) - { - StopInternal(); - return; - } - - if (subscribeResult.RetainedMessages != null) - { - foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) - { - var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); - Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); - } - } - } - - async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) - { - var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); - - var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); - - Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket)); - - if (unsubscribeResult.CloseConnection) - { - StopInternal(); - } - } - - async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - HandleTopicAlias(publishPacket); - - InterceptingPublishEventArgs interceptingPublishEventArgs = null; - var applicationMessage = _applicationMessageFactory.Create(publishPacket); - var closeConnection = false; - var processPublish = true; - - if (_eventContainer.InterceptingPublishEvent.HasHandlers) - { - interceptingPublishEventArgs = new InterceptingPublishEventArgs - { - ClientId = Id, - ApplicationMessage = applicationMessage, - SessionItems = Session.Items, - ProcessPublish = true, - CloseConnection = false, - CancellationToken = cancellationToken - }; - - if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic)) - { - // 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 _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); - - applicationMessage = interceptingPublishEventArgs.ApplicationMessage; - closeConnection = interceptingPublishEventArgs.CloseConnection; - processPublish = interceptingPublishEventArgs.ProcessPublish; - } - - if (closeConnection) - { - await StopAsync(MqttDisconnectReasonCode.UnspecifiedError); - return; - } - - if (processPublish && applicationMessage != null) - { - await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false); - } - - switch (publishPacket.QualityOfServiceLevel) - { - case MqttQualityOfServiceLevel.AtMostOnce: - { - // Do nothing since QoS 0 has no ack at all! - break; - } - case MqttQualityOfServiceLevel.AtLeastOnce: - { - var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket)); - break; - } - case MqttQualityOfServiceLevel.ExactlyOnce: - { - var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket)); - break; - } - default: - { - throw new MqttCommunicationException("Received a not supported QoS level."); - } - } - } - - void HandleTopicAlias(MqttPublishPacket publishPacket) - { - if (publishPacket.TopicAlias == 0) - { - return; - } - - lock (_topicAlias) - { - if (!string.IsNullOrEmpty(publishPacket.Topic)) - { - _topicAlias[publishPacket.TopicAlias] = publishPacket.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); - } - } - } - } - async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode) { try { var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode); - + using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) { await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false); diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index f42d23d5..d363ca54 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -9,40 +9,41 @@ namespace MQTTnet.Server { public sealed class MqttServerEventContainer { - public AsyncEvent InterceptingSubscriptionEvent { get; } = new AsyncEvent(); - - public AsyncEvent InterceptingUnsubscriptionEvent { get; } = new AsyncEvent(); - - public AsyncEvent InterceptingPublishEvent { get; } = new AsyncEvent(); - - public AsyncEvent ValidatingConnectionEvent { get; } = new AsyncEvent(); - - public AsyncEvent ClientConnectedEvent { get; } = new AsyncEvent(); - - public AsyncEvent ClientDisconnectedEvent { get; } = new AsyncEvent(); - - public AsyncEvent ClientSubscribedTopicEvent { get; } = new AsyncEvent(); - - public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); - - public AsyncEvent PreparingSessionEvent { get; } = new AsyncEvent(); - - public AsyncEvent SessionDeletedEvent { get; } = new AsyncEvent(); - public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); - - public AsyncEvent RetainedMessageChangedEvent { get; } = new AsyncEvent(); - - public AsyncEvent LoadingRetainedMessagesEvent { get; } = new AsyncEvent(); - - public AsyncEvent RetainedMessagesClearedEvent { get; } = new AsyncEvent(); - + + public AsyncEvent ClientAcknowledgedPublishPacketEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientConnectedEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientDisconnectedEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientSubscribedTopicEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingInboundPacketEvent { get; } = new AsyncEvent(); - + public AsyncEvent InterceptingOutboundPacketEvent { get; } = new AsyncEvent(); - + + public AsyncEvent InterceptingPublishEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingSubscriptionEvent { get; } = new AsyncEvent(); + + public AsyncEvent InterceptingUnsubscriptionEvent { get; } = new AsyncEvent(); + + public AsyncEvent LoadingRetainedMessagesEvent { get; } = new AsyncEvent(); + + public AsyncEvent PreparingSessionEvent { get; } = new AsyncEvent(); + + public AsyncEvent RetainedMessageChangedEvent { get; } = new AsyncEvent(); + + public AsyncEvent RetainedMessagesClearedEvent { get; } = new AsyncEvent(); + + public AsyncEvent SessionDeletedEvent { get; } = new AsyncEvent(); + public AsyncEvent StartedEvent { get; } = new AsyncEvent(); - + public AsyncEvent StoppedEvent { get; } = new AsyncEvent(); + + public AsyncEvent ValidatingConnectionEvent { get; } = new AsyncEvent(); } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index e152106a..de9ff74d 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -71,9 +71,18 @@ namespace MQTTnet.Server public bool WillMessageSent { get; set; } - public void AcknowledgePublishPacket(ushort packetIdentifier) + public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier) { - _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out _); + // This will only return the matching PUBLISH packet but does not remove it. + // This is required for QoS 2. + _unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket); + return publishPacket; + } + + public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier) + { + _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket); + return publishPacket; } public void AddSubscribedTopic(string topic) diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index fa3f49f2..9903be8a 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -60,6 +60,12 @@ namespace MQTTnet.Server add => _eventContainer.ClientConnectedEvent.AddHandler(value); remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); } + + public event Func ClientAcknowledgedPublishPacketAsync + { + add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value); + remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value); + } public event Func ClientDisconnectedAsync { From 11249d006f781ad75412f443a0a55f1bdea6632a Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 15:10:20 +0200 Subject: [PATCH 8/9] Add samples. --- Samples/Client/Client_Connection_Samples.cs | 25 +++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/Samples/Client/Client_Connection_Samples.cs b/Samples/Client/Client_Connection_Samples.cs index 75ef6e61..62feb648 100644 --- a/Samples/Client/Client_Connection_Samples.cs +++ b/Samples/Client/Client_Connection_Samples.cs @@ -8,6 +8,7 @@ using System.Security.Authentication; using MQTTnet.Client; +using MQTTnet.Extensions.WebSocket4Net; using MQTTnet.Formatter; using MQTTnet.Samples.Helpers; @@ -154,6 +155,30 @@ public static class Client_Connection_Samples response.DumpToConsole(); } } + + public static async Task Connect_Client_Using_WebSocket4Net() + { + /* + * This sample creates a simple MQTT client and connects to a public broker using a WebSocket connection. + * Instead of the .NET implementation of WebSockets the implementaion from WebSocket4Net is used. It provides more + * encryption algorithms and supports more platforms. + * + * This is a modified version of the sample _Connect_Client_! See other sample for more details. + */ + + var mqttFactory = new MqttFactory().UseWebSocket4Net(); + + using (var mqttClient = mqttFactory.CreateMqttClient()) + { + var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").Build(); + + var response = await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); + + Console.WriteLine("The MQTT client is connected."); + + response.DumpToConsole(); + } + } public static async Task Connect_Client_With_TLS_Encryption() { From cbf413f04e9c92c9685529276b73049b473895d7 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 15:33:13 +0200 Subject: [PATCH 9/9] Squashed commit of the following: 1125-expose-http-context-in-validateconnectionasync commit 353c9671a04a273798b0b3775be5f0764bfd7edf Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 23 15:33:01 2022 +0200 Remove usage of obsolete stuff. commit 1f3c35685c7a65453c37a37e8c54a2ca4305ff6d Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 23 15:30:54 2022 +0200 Update ValidatingConnectionEventArgs.cs commit a5e1b8938974b5b57e95ec41d74634ca130a8ad5 Merge: 8878efee 11249d00 Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 23 15:30:30 2022 +0200 Merge branch 'master' into 1125-expose-http-context-in-validateconnectionasync commit 8878efee0da3c67d90bb7fb6946453f62329cd8e Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat Jul 23 15:30:03 2022 +0200 Expose channel adapter when validating connections. --- .github/workflows/ReleaseNotes.md | 1 + MQTTnet.sln.DotSettings | 5 ++++- Samples/Server/Server_Simple_Samples.cs | 2 +- Source/MQTTnet.TestApp/ServerTest.cs | 2 +- Source/MQTTnet.Tests/Server/Security_Tests.cs | 4 ++-- .../Events/ValidatingConnectionEventArgs.cs | 17 +++++++++++------ .../Internal/MqttClientSessionsManager.cs | 10 +++++----- 7 files changed, 25 insertions(+), 16 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 7104a49f..67631049 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,3 +1,4 @@ * [Client] Added support for passing the local endpoint which should be used (network card). * [Client] Exposed _PackageIdentifier_ in _MqttApplicationMessageReceivedEventArgs_ (thanks to @koepalex, #1466). * [Server] Added a new event (_ClientAcknowledgedPublishPacketAsync_) which is fired whenever a client acknowledges a PUBLISH packet (QoS 1 and 2, #487). +* [Server] Exposed channel adapter (HTTP context etc.) in connection validation (#1125). diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings index 504fc3a2..87d205d0 100644 --- a/MQTTnet.sln.DotSettings +++ b/MQTTnet.sln.DotSettings @@ -29,7 +29,7 @@ OnDifferentLines OnDifferentLines <?xml version="1.0" encoding="utf-16"?> -<Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"> +<Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"> <TypePattern DisplayName="Non-reorderable types"> <TypePattern.Match> <Or> @@ -226,6 +226,9 @@ </Entry> </TypePattern> </Patterns> + // 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. True True True diff --git a/Samples/Server/Server_Simple_Samples.cs b/Samples/Server/Server_Simple_Samples.cs index e4d63eb4..0ea4b6b7 100644 --- a/Samples/Server/Server_Simple_Samples.cs +++ b/Samples/Server/Server_Simple_Samples.cs @@ -138,7 +138,7 @@ public static class Server_Simple_Samples e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; } - if (e.Username != "ValidUser") + if (e.UserName != "ValidUser") { e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; } diff --git a/Source/MQTTnet.TestApp/ServerTest.cs b/Source/MQTTnet.TestApp/ServerTest.cs index 341b3dda..e71f9761 100644 --- a/Source/MQTTnet.TestApp/ServerTest.cs +++ b/Source/MQTTnet.TestApp/ServerTest.cs @@ -118,7 +118,7 @@ namespace MQTTnet.TestApp { if (e.ClientId == "SpecialClient") { - if (e.Username != "USER" || e.Password != "PASS") + if (e.UserName != "USER" || e.Password != "PASS") { e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; } diff --git a/Source/MQTTnet.Tests/Server/Security_Tests.cs b/Source/MQTTnet.Tests/Server/Security_Tests.cs index 7d2336b0..0e144df8 100644 --- a/Source/MQTTnet.Tests/Server/Security_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Security_Tests.cs @@ -38,7 +38,7 @@ namespace MQTTnet.Tests.Server testEnvironment.Server.ValidatingConnectionAsync += eventArgs => { - if (eventArgs.Username == "SECRET") + if (eventArgs.UserName == "SECRET") { eventArgs.ReasonCode = MqttConnectReasonCode.Success; } @@ -147,7 +147,7 @@ namespace MQTTnet.Tests.Server server.ValidatingConnectionAsync += e => { - if (e.Username != "UserName1") + if (e.UserName != "UserName1") { e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; } diff --git a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index 8a5b48b6..eaa916e1 100644 --- a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -17,13 +17,12 @@ namespace MQTTnet.Server { public sealed class ValidatingConnectionEventArgs : EventArgs { - readonly IMqttChannelAdapter _clientAdapter; readonly MqttConnectPacket _connectPacket; public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) { _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); - _clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter)); + ChannelAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter)); } /// @@ -44,6 +43,12 @@ namespace MQTTnet.Server /// public string AuthenticationMethod => _connectPacket.AuthenticationMethod; + /// + /// Gets the channel adapter. This can be a _MqttConnectionContext_ (used in ASP.NET), a _MqttChannelAdapter_ (used for + /// TCP or WebSockets) or a custom implementation. + /// + public IMqttChannelAdapter ChannelAdapter { get; } + /// /// Gets or sets a value indicating whether clean sessions are used or not. /// When a client connects to a broker it can connect using either a non persistent connection (clean session) or a @@ -56,7 +61,7 @@ namespace MQTTnet.Server /// public bool? CleanSession => _connectPacket.CleanSession; - public X509Certificate2 ClientCertificate => _clientAdapter.ClientCertificate; + public X509Certificate2 ClientCertificate => ChannelAdapter.ClientCertificate; /// /// Gets the client identifier. @@ -64,9 +69,9 @@ namespace MQTTnet.Server /// public string ClientId => _connectPacket.ClientId; - public string Endpoint => _clientAdapter.Endpoint; + public string Endpoint => ChannelAdapter.Endpoint; - public bool IsSecureConnection => _clientAdapter.IsSecureConnection; + public bool IsSecureConnection => ChannelAdapter.IsSecureConnection; /// /// Gets or sets the keep alive period. @@ -85,7 +90,7 @@ namespace MQTTnet.Server public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray); - public MqttProtocolVersion ProtocolVersion => _clientAdapter.PacketFormatterAdapter.ProtocolVersion; + public MqttProtocolVersion ProtocolVersion => ChannelAdapter.PacketFormatterAdapter.ProtocolVersion; public byte[] RawPassword => _connectPacket.Password; diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 7d2e47ec..cea0e381 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -623,25 +623,25 @@ namespace MQTTnet.Server async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { - var context = new ValidatingConnectionEventArgs(connectPacket, channelAdapter) + var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter) { SessionItems = new ConcurrentDictionary() }; - await _eventContainer.ValidatingConnectionEvent.InvokeAsync(context).ConfigureAwait(false); + await _eventContainer.ValidatingConnectionEvent.InvokeAsync(eventArgs).ConfigureAwait(false); // Check the client ID and set a random one if supported. if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) { - connectPacket.ClientId = context.AssignedClientIdentifier; + connectPacket.ClientId = eventArgs.AssignedClientIdentifier; } if (string.IsNullOrEmpty(connectPacket.ClientId)) { - context.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; + eventArgs.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; } - return context; + return eventArgs; } } } \ No newline at end of file