Fix handling of retained messages with QoS > 0.

.github/workflows/ReleaseNotes.md

@@ -1,5 +1,6 @@
-* [Server] Connections with _try_private_ flag (MQTT 3.1.1 Bridge) are now accepted (#1413).
 * [Client] Added the flag _try_private_ (MQTT 3.1.1 Bridge) to client options (#1413).
 * [Client] Fixed MQTTv5 protocol violation in PUBLISH packets (#1423).
 * [Client] Added missing "WithWill..." methods in _MqttClientOptionsBuilder_.
-* [ManagedClient] Fixed wrong event args type for connected and disconnected events (#1432).
+* [ManagedClient] Fixed wrong event args type for connected and disconnected events (#1432).
+* [Server] Fixed wrong handling of retained messages with QoS > 0 (#1434).
+* [Server] Connections with _try_private_ flag (MQTT 3.1.1 Bridge) are now accepted (#1413).

Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs

@@ -16,67 +16,52 @@ namespace MQTTnet.Tests.Server
     public sealed class Retained_Messages_Tests : BaseTestClass
     {
         [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V3()
+        public async Task Clear_Retained_Message_With_Empty_Payload()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
-                var client = testEnvironment.CreateClient();
-                
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V311)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V5()
-        {
-            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-            {
-                await testEnvironment.StartServer();
-
-                var client = testEnvironment.CreateClient();
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V500)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Retained_Messages_Flow()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
-
-                await testEnvironment.StartServer();
                 var c1 = await testEnvironment.ConnectClient();
-    
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
+
+                await c1.DisconnectAsync();
+
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
-                await c1.PublishAsync(retainedMessage);
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
+            }
+        }
+
+        [TestMethod]
+        public async Task Clear_Retained_Message_With_Null_Payload()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var c1 = await testEnvironment.ConnectClient();
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+
                 await c1.DisconnectAsync();
-                await LongTestDelay();
 
-                for (var i = 0; i < 5; i++)
+                var c2 = await testEnvironment.ConnectClient();
-                {
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                    await c2.UnsubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i);
-                    
-                    await c2.SubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i + 1);
-                }
 
-                await c2.DisconnectAsync();
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
             }
         }
 
@@ -114,7 +99,7 @@ namespace MQTTnet.Tests.Server
 
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                 await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained").Build());
 
                 await Task.Delay(500);
@@ -125,52 +110,102 @@ namespace MQTTnet.Tests.Server
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message_With_Empty_Payload()
+        public async Task Receive_Retained_Messages_From_Higher_Qos_Level()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
+                // Upload retained message.
                 var c1 = await testEnvironment.ConnectClient();
-
+                await c1.PublishAsync(
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                    new MqttApplicationMessageBuilder().WithTopic("retained")
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
+                        .WithPayload(new byte[1])
-
+                        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
+                        .WithRetainFlag()
+                        .Build());
+                
                 await c1.DisconnectAsync();
 
+                // Subscribe using a new client.
                 var c2 = await testEnvironment.ConnectClient();
                 var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                 await Task.Delay(200);
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                // Using QoS 2 will lead to 1 instead because the publish was made with QoS level 1 (see 3.8.4 SUBSCRIBE Actions)!
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
                 await Task.Delay(500);
 
-                messageHandler.AssertReceivedCountEquals(0);
+                messageHandler.AssertReceivedCountEquals(1);
             }
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message_With_Null_Payload()
+        public async Task Retained_Messages_Flow()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
+
+                await testEnvironment.StartServer();
+                var c1 = await testEnvironment.ConnectClient();
+
+                var c2 = await testEnvironment.ConnectClient();
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+
+                await c1.PublishAsync(retainedMessage);
+                await c1.DisconnectAsync();
+                await LongTestDelay();
+
+                for (var i = 0; i < 5; i++)
+                {
+                    await c2.UnsubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i);
+
+                    await c2.SubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i + 1);
+                }
+
+                await c2.DisconnectAsync();
+            }
+        }
+
+        [TestMethod]
+        public async Task Server_Reports_Retained_Messages_Supported_V3()
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
                 await testEnvironment.StartServer();
 
-                var c1 = await testEnvironment.ConnectClient();
+                var client = testEnvironment.CreateClient();
 
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                var connectResult = await client.ConnectAsync(
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V311)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());
 
-                await c1.DisconnectAsync();
+                Assert.IsTrue(connectResult.RetainAvailable);
+            }
+        }
 
-                var c2 = await testEnvironment.ConnectClient();
+        [TestMethod]
-                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+        public async Task Server_Reports_Retained_Messages_Supported_V5()
-                
+        {
-                await Task.Delay(200);
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+            {
-                await Task.Delay(500);
+                await testEnvironment.StartServer();
 
-                messageHandler.AssertReceivedCountEquals(0);
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V500)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());
+
+                Assert.IsTrue(connectResult.RetainAvailable);
             }
         }
     }

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -392,7 +392,7 @@ namespace MQTTnet.Server
                 foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
                 {
                     var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
-                    Session.EnqueueControlPacket(new MqttPacketBusItem(publishPacket));
+                    Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
                 }
             }
         }

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -19,6 +19,7 @@ namespace MQTTnet.Server
     {
         readonly MqttClientSessionsManager _clientSessionsManager;
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
+        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         readonly MqttServerOptions _serverOptions;
 
@@ -100,6 +101,11 @@ namespace MQTTnet.Server
             SubscriptionsManager.Dispose();
         }
 
+        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
+        {
+            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
+        }
+
         public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
         {
             if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
@@ -118,28 +124,27 @@ namespace MQTTnet.Server
             }
 
             var publishPacket = (MqttPublishPacket)packetBusItem.Packet;
+
             if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
                 _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
         }
 
-        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
-        {
-            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
-        }
-
         public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
         {
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
         }
-        
+
         public void Recover()
         {
             // TODO: Keep the bus and only insert pending items again.
             // TODO: Check if packet identifier must be restarted or not.
+            // TODO: Recover package identifier.
             _packetBus.Clear();
 
             foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())