From bfec9dc5ad58bc08a64432b1497f4e2cbc3f1346 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Wed, 9 Nov 2022 21:14:53 +0100 Subject: [PATCH] 802 client isnt being notified if message is not delivered (#1568) --- .github/workflows/ReleaseNotes.md | 1 + Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs | 4 +- Source/MQTTnet.Tests/Server/Publish_Tests.cs | 58 +++++++++++++++++++ .../Publishing/MqttClientPublishResult.cs | 31 ++++++---- .../Formatter/MqttPubAckPacketFactory.cs | 38 ++++++------ .../Formatter/MqttPubRecPacketFactory.cs | 10 +++- .../MQTTnet/Protocol/MqttPubAckReasonCode.cs | 5 ++ .../DispatchApplicationMessageResult.cs | 16 +++++ Source/MQTTnet/Server/Internal/MqttClient.cs | 7 ++- .../Internal/MqttClientSessionsManager.cs | 25 ++++---- 10 files changed, 146 insertions(+), 49 deletions(-) create mode 100644 Source/MQTTnet.Tests/Server/Publish_Tests.cs create mode 100644 Source/MQTTnet/Server/Internal/DispatchApplicationMessageResult.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 1614de15..608f1589 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -4,3 +4,4 @@ * [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550) * [Server] Fixed issue in upgrading and downgrading of QoS levels for subscriptions and retained messages (#1560) * [Server] Fixed memory leak when old sessions are discarded (#1553) +* [Server] When using MQTTv5 the reason code _NoMatchingSubscribers_ is returned when no client is subscribed to a published topic (#802) diff --git a/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs b/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs index ac7b8792..29fd74e9 100644 --- a/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs +++ b/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs @@ -154,7 +154,7 @@ namespace MQTTnet.Tests.MQTTv5 var result = await client.PublishStringAsync("a", "b", MqttQualityOfServiceLevel.AtLeastOnce); await client.DisconnectAsync(); - Assert.AreEqual(MqttClientPublishReasonCode.Success, result.ReasonCode); + Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, result.ReasonCode); } } @@ -169,7 +169,7 @@ namespace MQTTnet.Tests.MQTTv5 var result = await client.PublishStringAsync("a", "b", MqttQualityOfServiceLevel.ExactlyOnce); await client.DisconnectAsync(); - Assert.AreEqual(MqttClientPublishReasonCode.Success, result.ReasonCode); + Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, result.ReasonCode); } } diff --git a/Source/MQTTnet.Tests/Server/Publish_Tests.cs b/Source/MQTTnet.Tests/Server/Publish_Tests.cs new file mode 100644 index 00000000..c2293f0a --- /dev/null +++ b/Source/MQTTnet.Tests/Server/Publish_Tests.cs @@ -0,0 +1,58 @@ +// 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.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Formatter; +using MQTTnet.Protocol; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class Publish_Tests : BaseTestClass + { + [TestMethod] + public async Task Return_Success_When_Subscribed() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(); + + var sender = await testEnvironment.ConnectClient(); + var receiver = await testEnvironment.ConnectClient(); + + await receiver.SubscribeAsync("A"); + + // AtLeastOnce is required to get an ACK packet from the server. + var publishResult = await sender.PublishStringAsync("A", "Payload", MqttQualityOfServiceLevel.AtLeastOnce); + + Assert.AreEqual(MqttClientPublishReasonCode.Success, publishResult.ReasonCode); + + Assert.AreEqual(true, publishResult.IsSuccess); + } + } + + [TestMethod] + public async Task Return_NoMatchingSubscribers_When_Not_Subscribed() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(); + + var sender = await testEnvironment.ConnectClient(); + var receiver = await testEnvironment.ConnectClient(); + + await receiver.SubscribeAsync("A"); + + // AtLeastOnce is required to get an ACK packet from the server. + var publishResult = await sender.PublishStringAsync("B", "Payload", MqttQualityOfServiceLevel.AtLeastOnce); + + Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, publishResult.ReasonCode); + + Assert.AreEqual(true, publishResult.IsSuccess); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index 47d338eb..6fd2dcd8 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -10,29 +10,38 @@ namespace MQTTnet.Client public sealed class MqttClientPublishResult { /// - /// Gets the packet identifier which was used for this publish. + /// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or + /// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit + /// to the server etc. was a success. + /// + public bool IsSuccess => ReasonCode == MqttClientPublishReasonCode.Success || ReasonCode == MqttClientPublishReasonCode.NoMatchingSubscribers; + + /// + /// Gets the packet identifier which was used for this publish. /// public ushort? PacketIdentifier { get; set; } /// - /// Gets or sets the reason code. - /// Hint: MQTT 5 feature only. + /// Gets or sets the reason code. + /// Hint: MQTT 5 feature only. /// public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success; /// - /// Gets or sets the reason string. - /// Hint: MQTT 5 feature only. + /// Gets or sets the reason string. + /// Hint: MQTT 5 feature only. /// public string ReasonString { get; set; } /// - /// Gets or sets the user properties. - /// In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT packet. - /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add metadata to MQTT messages and pass information between publisher, broker, and subscriber. - /// The feature is very similar to the HTTP header concept. - /// Hint: MQTT 5 feature only. + /// Gets or sets the user properties. + /// In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT + /// packet. + /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add + /// metadata to MQTT messages and pass information between publisher, broker, and subscriber. + /// The feature is very similar to the HTTP header concept. + /// Hint: MQTT 5 feature only. /// public IReadOnlyCollection UserProperties { get; internal set; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs index 4d6c6870..536ec011 100644 --- a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs @@ -12,7 +12,10 @@ namespace MQTTnet.Formatter { public sealed class MqttPubAckPacketFactory { - public MqttPubAckPacket Create(MqttPublishPacket publishPacket, InterceptingPublishEventArgs interceptingPublishEventArgs) + public MqttPubAckPacket Create( + MqttPublishPacket publishPacket, + InterceptingPublishEventArgs interceptingPublishEventArgs, + DispatchApplicationMessageResult dispatchApplicationMessageResult) { if (publishPacket == null) { @@ -21,10 +24,19 @@ namespace MQTTnet.Formatter var pubAckPacket = new MqttPubAckPacket { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubAckReasonCode.Success + PacketIdentifier = publishPacket.PacketIdentifier }; + if (dispatchApplicationMessageResult.MatchingSubscribersCount == 0) + { + // NoMatchingSubscribers is ONLY sent by the server! + pubAckPacket.ReasonCode = MqttPubAckReasonCode.NoMatchingSubscribers; + } + else + { + pubAckPacket.ReasonCode = MqttPubAckReasonCode.Success; + } + if (interceptingPublishEventArgs != null) { pubAckPacket.ReasonCode = (MqttPubAckReasonCode)(int)interceptingPublishEventArgs.Response.ReasonCode; @@ -42,24 +54,12 @@ namespace MQTTnet.Formatter throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs)); } - var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode); - pubAckPacket.UserProperties = applicationMessageReceivedEventArgs.ResponseUserProperties; - pubAckPacket.ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString; - - return pubAckPacket; - } - - static MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) - { - if (publishPacket == null) - { - throw new ArgumentNullException(nameof(publishPacket)); - } - var pubAckPacket = new MqttPubAckPacket { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = (MqttPubAckReasonCode)(int)applicationMessageReceivedReasonCode + PacketIdentifier = applicationMessageReceivedEventArgs.PublishPacket.PacketIdentifier, + ReasonCode = (MqttPubAckReasonCode)(int)applicationMessageReceivedEventArgs.ReasonCode, + UserProperties = applicationMessageReceivedEventArgs.ResponseUserProperties, + ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString }; return pubAckPacket; diff --git a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs index 5c0726d8..0a6bba24 100644 --- a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs @@ -25,7 +25,10 @@ namespace MQTTnet.Formatter return pubRecPacket; } - public MqttPacket Create(MqttPublishPacket publishPacket, InterceptingPublishEventArgs interceptingPublishEventArgs) + public MqttPacket Create( + MqttPublishPacket publishPacket, + InterceptingPublishEventArgs interceptingPublishEventArgs, + DispatchApplicationMessageResult dispatchApplicationMessageResult) { if (publishPacket == null) { @@ -38,6 +41,11 @@ namespace MQTTnet.Formatter ReasonCode = MqttPubRecReasonCode.Success }; + if (dispatchApplicationMessageResult.MatchingSubscribersCount == 0) + { + pubRecPacket.ReasonCode = MqttPubRecReasonCode.NoMatchingSubscribers; + } + if (interceptingPublishEventArgs != null) { pubRecPacket.ReasonCode = (MqttPubRecReasonCode)(int)interceptingPublishEventArgs.Response.ReasonCode; diff --git a/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs b/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs index 9eded18c..50de497f 100644 --- a/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs +++ b/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs @@ -7,7 +7,12 @@ namespace MQTTnet.Protocol public enum MqttPubAckReasonCode { Success = 0, + + /// + /// The message is accepted but there are no subscribers. This is sent only by the Server. If the Server knows that there are no matching subscribers, it MAY use this Reason Code instead of 0x00 (Success). + /// NoMatchingSubscribers = 16, + UnspecifiedError = 128, ImplementationSpecificError = 131, NotAuthorized = 135, diff --git a/Source/MQTTnet/Server/Internal/DispatchApplicationMessageResult.cs b/Source/MQTTnet/Server/Internal/DispatchApplicationMessageResult.cs new file mode 100644 index 00000000..bc4637a1 --- /dev/null +++ b/Source/MQTTnet/Server/Internal/DispatchApplicationMessageResult.cs @@ -0,0 +1,16 @@ +// 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. + +namespace MQTTnet.Server +{ + public sealed class DispatchApplicationMessageResult + { + public DispatchApplicationMessageResult(int matchingSubscribersCount) + { + MatchingSubscribersCount = matchingSubscribersCount; + } + + public int MatchingSubscribersCount { get; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 674a262f..f9f7e695 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -237,9 +237,10 @@ namespace MQTTnet.Server return; } + DispatchApplicationMessageResult dispatchResult = null; if (processPublish && applicationMessage != null) { - await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false); + dispatchResult = await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false); } switch (publishPacket.QualityOfServiceLevel) @@ -251,13 +252,13 @@ namespace MQTTnet.Server } case MqttQualityOfServiceLevel.AtLeastOnce: { - var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs); + var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs, dispatchResult); Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket)); break; } case MqttQualityOfServiceLevel.ExactlyOnce: { - var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs); + var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs, dispatchResult); Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket)); break; } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 186a15ca..9fb9ca18 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -127,21 +127,20 @@ namespace MQTTnet.Server _logger.Verbose("Session for client '{0}' deleted.", clientId); } - - public async Task DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage) + + public async Task DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage) { + var matchingSubscribersCount = 0; try { if (applicationMessage.Retain) { await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false); } - - var deliveryCount = 0; + List subscriberSessions; lock (_sessionsManagementLock) { - // only subscriber clients are of interest here. subscriberSessions = _subscriberSessions.ToList(); } @@ -187,17 +186,22 @@ namespace MQTTnet.Server } session.EnqueueDataPacket(new MqttPacketBusItem(newPublishPacket)); - deliveryCount++; + matchingSubscribersCount++; _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic); } - await FireApplicationMessageNotConsumedEvent(applicationMessage, deliveryCount, senderId); + if (matchingSubscribersCount == 0) + { + await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false); + } } catch (Exception exception) { _logger.Error(exception, "Unhandled exception while processing next queued application message."); } + + return new DispatchApplicationMessageResult(matchingSubscribersCount); } public void Dispose() @@ -575,13 +579,8 @@ namespace MQTTnet.Server return new MqttSession(clientId, isPersistent, sessionItems, _options, _eventContainer, _retainedMessagesManager, this); } - async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, int deliveryCount, string senderId) + async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId) { - if (deliveryCount > 0) - { - return; - } - if (!_eventContainer.ApplicationMessageNotConsumedEvent.HasHandlers) { return;