Fix upgrading and downgrading of retained messages QoS for subscriptions (#1564)

.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)
+* [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)

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.

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();

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;
+        }
     }
 }

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);

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));
                 }
             }

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -169,10 +169,7 @@ namespace MQTTnet.Server
             }
 
             var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false);
-            var result = new SubscribeResult
+            var result = new SubscribeResult(subscribePacket.TopicFilters.Count);
-            {
-                ReasonCodes = new List<MqttSubscribeReasonCode>(subscribePacket.TopicFilters.Count)
-            };
 
             var addedSubscriptions = new List<string>();
             var finalTopicFilters = new List<MqttTopicFilter>();
@@ -412,14 +409,14 @@ namespace MQTTnet.Server
         }
 
         static void FilterRetainedApplicationMessages(
-            IList<MqttApplicationMessage> retainedApplicationMessages,
+            IList<MqttApplicationMessage> 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];
+                var retainedMessage = retainedMessages[index];
-                if (retainedApplicationMessage == null)
+                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,
+                    // UPGRADING the QoS is not allowed! 
-                    SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel
+                    // 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;
             }
         }
 

Source/MQTTnet/Server/Internal/SubscribeResult.cs

@@ -10,8 +10,13 @@ namespace MQTTnet.Server
 {
     public sealed class SubscribeResult
     {
+        public SubscribeResult(int topicsCount)
+        {
+            ReasonCodes = new List<MqttSubscribeReasonCode>(topicsCount);
+        }
+        
         public bool CloseConnection { get; set; }
-
+        
         public List<MqttSubscribeReasonCode> ReasonCodes { get; set; }
 
         public string ReasonString { get; set; }

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; }
     }