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] 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 {