diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 897cf698..67631049 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,8 +1,4 @@ -* [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 +* [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/.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: 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/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() { 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 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.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.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/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..0e144df8 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; @@ -40,7 +38,7 @@ namespace MQTTnet.Tests.Server testEnvironment.Server.ValidatingConnectionAsync += eventArgs => { - if (eventArgs.Username == "SECRET") + if (eventArgs.UserName == "SECRET") { eventArgs.ReasonCode = MqttConnectReasonCode.Success; } @@ -149,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.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/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 479bb1c5..586cab51 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,8 +11,19 @@ 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; + /// + /// 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/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) 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; 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(); 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..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; @@ -161,7 +166,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 9e97b671..3d4c5d82 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,11 +138,226 @@ 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) @@ -163,12 +367,12 @@ namespace MQTTnet.Server var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items); await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); - + if (!interceptingPacketEventArgs.ProcessPacket || packet == null) { return null; } - + return interceptingPacketEventArgs.Packet; } @@ -190,7 +394,7 @@ namespace MQTTnet.Server } var processPacket = true; - + if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers) { var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items); @@ -198,7 +402,7 @@ namespace MQTTnet.Server packet = interceptingPacketEventArgs.Packet; processPacket = interceptingPacketEventArgs.ProcessPacket; } - + if (!processPacket || packet == null) { // Restart the receiving process to get the next packet ignoring the current one.. @@ -213,15 +417,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) { @@ -339,7 +543,7 @@ namespace MQTTnet.Server StopInternal(); } } - + void StopInternal() { try @@ -499,7 +703,7 @@ namespace MQTTnet.Server 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/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 4787e28c..eb2359c7 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -603,25 +603,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 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 {