From 3fc6982e9a08bb8a78690ac9645724dc0a314d50 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 6 Nov 2022 10:35:59 +0100 Subject: [PATCH] Fix upgrading and downgrading of retained messages QoS for subscriptions (#1564) --- .github/workflows/ReleaseNotes.md | 3 +- .../Server/Retained_Messages_Tests.cs | 62 ++++++++++++++++++- .../MQTTnet/Formatter/MqttPacketFactories.cs | 1 + .../Formatter/MqttPublishPacketFactory.cs | 13 ++++ Source/MQTTnet/Server/Internal/MqttClient.cs | 5 +- .../Internal/MqttClientSessionsManager.cs | 4 +- .../MqttClientSubscriptionsManager.cs | 30 ++++----- .../Server/Internal/SubscribeResult.cs | 7 ++- .../Server/MqttRetainedMessageMatch.cs | 9 ++- 9 files changed, 111 insertions(+), 23 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index b2cc9759..95fe5078 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,2 +1,3 @@ * [Core] Fixed several dead locks by implementing a new async signal (#1552) -* [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550) \ No newline at end of file +* [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) \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs b/Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs index 1a6b0996..b2774157 100644 --- a/Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs @@ -65,6 +65,66 @@ namespace MQTTnet.Tests.Server } } + [TestMethod] + public async Task Downgrade_QoS_Level() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + + // Add the retained message with QoS 2! + await c1.PublishAsync( + new MqttApplicationMessageBuilder().WithTopic("retained") + .WithPayload(new byte[3]) + .WithRetainFlag() + .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce) + .Build()); + + // The second client uses QoS 1 so a downgrade is required. + var c2 = await testEnvironment.ConnectClient(); + var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2); + await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); + + await LongTestDelay(); + + messageHandler.AssertReceivedCountEquals(1); + + Assert.AreEqual(MqttQualityOfServiceLevel.AtLeastOnce, messageHandler.ReceivedEventArgs.First().ApplicationMessage.QualityOfServiceLevel); + } + } + + [TestMethod] + public async Task No_Upgrade_QoS_Level() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + + // Add the retained message with QoS 1! + await c1.PublishAsync( + new MqttApplicationMessageBuilder().WithTopic("retained") + .WithPayload(new byte[3]) + .WithRetainFlag() + .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) + .Build()); + + // The second client uses QoS 2 so an upgrade is expected but according to the MQTT spec this is not supported! + var c2 = await testEnvironment.ConnectClient(); + var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2); + await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }); + + await LongTestDelay(); + + messageHandler.AssertReceivedCountEquals(1); + + Assert.AreEqual(MqttQualityOfServiceLevel.AtLeastOnce, messageHandler.ReceivedEventArgs.First().ApplicationMessage.QualityOfServiceLevel); + } + } + [TestMethod] public async Task Receive_No_Retained_Message_After_Subscribe() { @@ -124,7 +184,7 @@ namespace MQTTnet.Tests.Server .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag() .Build()); - + await c1.DisconnectAsync(); // Subscribe using a new client. diff --git a/Source/MQTTnet/Formatter/MqttPacketFactories.cs b/Source/MQTTnet/Formatter/MqttPacketFactories.cs index a13c2aa6..70de69d2 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFactories.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFactories.cs @@ -7,6 +7,7 @@ namespace MQTTnet.Formatter public sealed class MqttPacketFactories { public MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory(); + public MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory(); public MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory(); diff --git a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs index 90325e9e..ab33f7a9 100644 --- a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs @@ -5,6 +5,7 @@ using System; using MQTTnet.Exceptions; using MQTTnet.Packets; +using MQTTnet.Server; namespace MQTTnet.Formatter { @@ -80,5 +81,17 @@ namespace MQTTnet.Formatter return packet; } + + public MqttPublishPacket Create(MqttRetainedMessageMatch retainedMessage) + { + if (retainedMessage == null) + { + throw new ArgumentNullException(nameof(retainedMessage)); + } + + var publishPacket = Create(retainedMessage.ApplicationMessage); + publishPacket.QualityOfServiceLevel = retainedMessage.SubscriptionQualityOfServiceLevel; + return publishPacket; + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index cb003f83..1bc51152 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -296,15 +296,14 @@ namespace MQTTnet.Server if (subscribeResult.RetainedMessages != null) { - foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) + foreach (var retainedMessageMatch in subscribeResult.RetainedMessages) { - var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); + var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch); Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } } } - async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 125af7fd..87fcdd34 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -418,9 +418,9 @@ namespace MQTTnet.Server if (subscribeResult.RetainedMessages != null) { - foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) + foreach (var retainedMessageMatch in subscribeResult.RetainedMessages) { - var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); + var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch); clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 3a505a40..a6542d14 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -169,10 +169,7 @@ namespace MQTTnet.Server } var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false); - var result = new SubscribeResult - { - ReasonCodes = new List(subscribePacket.TopicFilters.Count) - }; + var result = new SubscribeResult(subscribePacket.TopicFilters.Count); var addedSubscriptions = new List(); var finalTopicFilters = new List(); @@ -412,14 +409,14 @@ namespace MQTTnet.Server } static void FilterRetainedApplicationMessages( - IList retainedApplicationMessages, + IList retainedMessages, CreateSubscriptionResult createSubscriptionResult, SubscribeResult subscribeResult) { - for (var index = retainedApplicationMessages.Count - 1; index >= 0; index--) + for (var index = retainedMessages.Count - 1; index >= 0; index--) { - var retainedApplicationMessage = retainedApplicationMessages[index]; - if (retainedApplicationMessage == null) + var retainedMessage = retainedMessages[index]; + if (retainedMessage == null) { continue; } @@ -436,16 +433,21 @@ namespace MQTTnet.Server continue; } - if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch) + if (MqttTopicFilterComparer.Compare(retainedMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch) { continue; } - var retainedMessageMatch = new MqttRetainedMessageMatch + var retainedMessageMatch = new MqttRetainedMessageMatch(retainedMessage, createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel); + if (retainedMessageMatch.SubscriptionQualityOfServiceLevel > retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel) { - ApplicationMessage = retainedApplicationMessage, - SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel - }; + // UPGRADING the QoS is not allowed! + // From MQTT spec: Subscribing to a Topic Filter at QoS 2 is equivalent to saying + // "I would like to receive Messages matching this filter at the QoS with which they were published". + // This means a publisher is responsible for determining the maximum QoS a Message can be delivered at, + // but a subscriber is able to require that the Server downgrades the QoS to one more suitable for its usage. + retainedMessageMatch.SubscriptionQualityOfServiceLevel = retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel; + } if (subscribeResult.RetainedMessages == null) { @@ -456,7 +458,7 @@ namespace MQTTnet.Server // Clear the retained message from the list because the client should receive every message only // one time even if multiple subscriptions affect them. - retainedApplicationMessages[index] = null; + retainedMessages[index] = null; } } diff --git a/Source/MQTTnet/Server/Internal/SubscribeResult.cs b/Source/MQTTnet/Server/Internal/SubscribeResult.cs index beff649b..9c828eda 100644 --- a/Source/MQTTnet/Server/Internal/SubscribeResult.cs +++ b/Source/MQTTnet/Server/Internal/SubscribeResult.cs @@ -10,8 +10,13 @@ namespace MQTTnet.Server { public sealed class SubscribeResult { + public SubscribeResult(int topicsCount) + { + ReasonCodes = new List(topicsCount); + } + public bool CloseConnection { get; set; } - + public List ReasonCodes { get; set; } public string ReasonString { get; set; } diff --git a/Source/MQTTnet/Server/MqttRetainedMessageMatch.cs b/Source/MQTTnet/Server/MqttRetainedMessageMatch.cs index a54e5b7f..59bfbc2c 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessageMatch.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessageMatch.cs @@ -2,13 +2,20 @@ // 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 MQTTnet.Protocol; namespace MQTTnet.Server { public sealed class MqttRetainedMessageMatch { - public MqttApplicationMessage ApplicationMessage { get; set; } + public MqttRetainedMessageMatch(MqttApplicationMessage applicationMessage, MqttQualityOfServiceLevel subscriptionQualityOfServiceLevel) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + SubscriptionQualityOfServiceLevel = subscriptionQualityOfServiceLevel; + } + + public MqttApplicationMessage ApplicationMessage { get; } public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; } }