Add more conversion methos for application message payloads

.github/workflows/ReleaseNotes.md

@@ -1,5 +1 @@
-* [Core] Fixed several dead locks by implementing a new async signal (#1552)
-* [Client] All waiting tasks will now fail when the client connection gets closed. This fixes several freezes (#1561)
-* [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)
+

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -43,7 +43,7 @@ namespace MQTTnet.Extensions.ManagedClient
         
         Task StartAsync(ManagedMqttClientOptions options);
         
-        Task StopAsync();
+        Task StopAsync(bool cleanDisconnect = true);
         
         Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters);
         

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -62,6 +62,7 @@ namespace MQTTnet.Extensions.ManagedClient
         CancellationTokenSource _publishingCancellationToken;
 
         ManagedMqttClientStorageManager _storageManager;
+        bool _isCleanDisconnect;
 
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
         {
@@ -160,7 +161,7 @@ namespace MQTTnet.Extensions.ManagedClient
                 throw new InvalidOperationException("call StartAsync before publishing messages");
             }
 
-            MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
+            MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage);
 
             ManagedMqttApplicationMessage removedMessage = null;
             ApplicationMessageSkippedEventArgs applicationMessageSkippedEventArgs = null;
@@ -255,10 +256,12 @@ namespace MQTTnet.Extensions.ManagedClient
             _logger.Info("Started");
         }
 
-        public async Task StopAsync()
+        public async Task StopAsync(bool cleanDisconnect = true)
         {
             ThrowIfDisposed();
 
+            _isCleanDisconnect = cleanDisconnect;
+            
             StopPublishing();
             StopMaintainingConnection();
 
@@ -382,9 +385,12 @@ namespace MQTTnet.Extensions.ManagedClient
                 {
                     try
                     {
-                        using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
+                        if (_isCleanDisconnect)
                         {
-                            await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
+                            using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
+                            {
+                                await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
+                            }
                         }
                     }
                     catch (OperationCanceledException)

Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs

@@ -64,7 +64,7 @@ namespace MQTTnet.Tests.Client
 
             await client.ConnectAsync(options, CancellationToken.None).ConfigureAwait(false);
         }
-        
+
         [TestMethod]
         [ExpectedException(typeof(MqttCommunicationException))]
         public async Task Connect_To_Wrong_Host()
@@ -80,6 +80,8 @@ namespace MQTTnet.Tests.Client
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
+                testEnvironment.IgnoreServerLogErrors = true;
+
                 testEnvironment.ServerPort = 8364;
                 var server = await testEnvironment.StartServer();
 
@@ -116,6 +118,8 @@ namespace MQTTnet.Tests.Client
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
+                testEnvironment.IgnoreServerLogErrors = true;
+
                 var server = await testEnvironment.StartServer();
 
                 using (var lowLevelClient = testEnvironment.CreateLowLevelClient())
@@ -123,7 +127,7 @@ namespace MQTTnet.Tests.Client
                     Assert.IsFalse(lowLevelClient.IsConnected);
 
                     var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithTimeout(TimeSpan.FromSeconds(1)).Build();
-                    
+
                     await lowLevelClient.ConnectAsync(clientOptions, CancellationToken.None);
 
                     Assert.IsTrue(lowLevelClient.IsConnected);
@@ -148,7 +152,7 @@ namespace MQTTnet.Tests.Client
                     catch
                     {
                     }
-           
+
                     Assert.IsFalse(lowLevelClient.IsConnected);
                 }
             }

Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs

@@ -11,6 +11,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Extensions.ManagedClient;
+using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
@@ -22,51 +23,64 @@ namespace MQTTnet.Tests.Client
     public sealed class ManagedMqttClient_Tests : BaseTestClass
     {
         [TestMethod]
-        public async Task Subscribe_When_Offline_But_Existing_Session()
+        public async Task Receive_While_Not_Cleanly_Disconnected()
         {
-            using (var testEnvironment = CreateTestEnvironment())
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
             {
-                var server = await testEnvironment.StartServer(o => o.WithPersistentSessions());
-                
-                // The connected client will crate a new session.
-                var mqttClient = await testEnvironment.ConnectClient();
-                mqttClient.Dispose();
-                
-                mqttClient = testEnvironment.CreateClient();
-                var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient);
-                
-                // Subscribe before starting!
-                await managedClient.SubscribeAsync("a");
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
 
-                MqttApplicationMessage applicationMessage = null;
-                managedClient.ApplicationMessageReceivedAsync += e =>
+                var senderClient = await testEnvironment.ConnectClient();
+
+                // Prepare managed client.
+                var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
+                await managedClient.SubscribeAsync("#");
+                var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
+                
+                var managedClientOptions = new ManagedMqttClientOptions
                 {
-                    applicationMessage = e.ApplicationMessage;
-                    return CompletedTask.Instance;
+                    ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
+                        .WithCleanSession(false)
+                        .Build()
                 };
-                
-                // Now the client must send the new subscription. 
-                await managedClient.StartAsync(
-                    new ManagedMqttClientOptions
-                    {
-                        ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder().WithCleanSession(false).WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build()
-                    });
 
+                await managedClient.StartAsync(managedClientOptions);
                 await LongTestDelay();
                 
-                await server.InjectApplicationMessage(
-                    new InjectedMqttApplicationMessage(
-                        new MqttApplicationMessage
-                        {
-                            Topic = "a"
-                        }));
-
+                // Send test data.
+                await senderClient.PublishStringAsync("topic1");
                 await LongTestDelay();
                 
-                Assert.IsNotNull(applicationMessage);
+                receivedMessages.AssertReceivedCountEquals(1);
+
+                // Stop the managed client but keep session at server (not clean disconnect required).
+                await managedClient.StopAsync(false);
+                await LongTestDelay();
+                
+                // Send new messages in the meantime.
+                await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
+                await LongTestDelay();
+                
+                // Start the managed client, it should receive the new message.
+                await managedClient.StartAsync(managedClientOptions);
+                await LongTestDelay();
+                
+                receivedMessages.AssertReceivedCountEquals(2);
+                
+                // Stop and start again, no new message should be received.
+                for (var i = 0; i < 3; i++)
+                {
+                    await managedClient.StopAsync(false);
+                    await LongTestDelay();
+                    await managedClient.StartAsync(managedClientOptions);
+                    await LongTestDelay();
+                }
+
+                receivedMessages.AssertReceivedCountEquals(2);
             }
         }
-        
+
         [TestMethod]
         public async Task Connect_To_Invalid_Server()
         {
@@ -75,10 +89,10 @@ namespace MQTTnet.Tests.Client
                 testEnvironment.IgnoreClientLogErrors = true;
 
                 var mqttClient = testEnvironment.CreateClient();
+
                 var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient);
 
                 ConnectingFailedEventArgs connectingFailedEventArgs = null;
-
                 managedClient.ConnectingFailedAsync += args =>
                 {
                     connectingFailedEventArgs = args;

Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs

@@ -52,12 +52,26 @@ namespace MQTTnet.Tests.Internal
         }
 
         [TestMethod]
-        public async Task Loop_Signal()
+        public async Task Reset_Signal()
         {
             var asyncSignal = new AsyncSignal();
 
+            // WaitAsync should fail because no signal is available.
             for (var i = 0; i < 10; i++)
             {
+                try
+                {
+                    using (var timeout = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)))
+                    {
+                        await asyncSignal.WaitAsync(timeout.Token);
+                    }
+                    
+                    Assert.Fail("This must fail because the signal is not yet set.");
+                }
+                catch (OperationCanceledException)
+                {
+                }
+                
                 asyncSignal.Set();
                 
                 // WaitAsync should return directly because the signal is available.

Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs

@@ -0,0 +1,43 @@
+// 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.Internal;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Tests.Internal
+{
+    [TestClass]
+    public sealed class MqttPacketBusItem_Tests
+    {
+        [TestMethod]
+        public void Fire_Completed_Event()
+        {
+            var eventFired = false;
+
+            var item = new MqttPacketBusItem(new MqttPublishPacket());
+            item.Completed += (_, __) =>
+            {
+                eventFired = true;
+            };
+
+            item.Complete();
+
+            Assert.IsTrue(eventFired);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(TaskCanceledException))]
+        public async Task Wait_Packet_Bus_Item_After_Already_Canceled()
+        {
+            var item = new MqttPacketBusItem(new MqttPublishPacket());
+
+            // Finish the item before the actual 
+            item.Cancel();
+
+            await item.WaitAsync();
+        }
+    }
+}

Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs

@@ -13,7 +13,129 @@ namespace MQTTnet.Tests.Internal
 {
     [TestClass]
     public sealed class MqttPacketBus_Tests
-    {
+    {       
+        [TestMethod]
+        public void Alternate_Priorities()
+        {
+            var bus = new MqttPacketBus();
+
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+
+            Assert.AreEqual(9, bus.TotalItemsCount);
+
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+
+            Assert.AreEqual(0, bus.TotalItemsCount);
+        }
+
+        [TestMethod]
+        public void Await_Single_Packet()
+        {
+            var bus = new MqttPacketBus();
+
+            var delivered = false;
+
+            var item1 = new MqttPacketBusItem(new MqttPublishPacket());
+            var item2 = new MqttPacketBusItem(new MqttPublishPacket());
+
+            var item3 = new MqttPacketBusItem(new MqttPublishPacket());
+            item3.Completed += (_, __) =>
+            {
+                delivered = true;
+            };
+
+            bus.EnqueueItem(item1, MqttPacketBusPartition.Data);
+            bus.EnqueueItem(item2, MqttPacketBusPartition.Data);
+            bus.EnqueueItem(item3, MqttPacketBusPartition.Data);
+
+            Assert.IsFalse(delivered);
+
+            bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
+
+            Assert.IsFalse(delivered);
+
+            bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
+
+            Assert.IsFalse(delivered);
+
+            bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
+
+            // The third packet has the event attached.
+            Assert.IsTrue(delivered);
+        }
+
+        [TestMethod]
+        public void Export_Packets_Without_Dequeue()
+        {
+            var bus = new MqttPacketBus();
+
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+
+            Assert.AreEqual(3, bus.TotalItemsCount);
+
+            var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
+            Assert.AreEqual(0, exportedPackets.Count);
+
+            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
+            Assert.AreEqual(0, exportedPackets.Count);
+
+            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
+            Assert.AreEqual(3, exportedPackets.Count);
+
+            Assert.AreEqual(3, bus.TotalItemsCount);
+        }
+
+        [TestMethod]
+        public async Task Fill_From_Different_Task()
+        {
+            const int MessageCount = 500;
+
+            var delayRandom = new Random();
+
+            var bus = new MqttPacketBus();
+
+            _ = Task.Run(
+                () =>
+                {
+                    for (var i = 0; i < MessageCount; i++)
+                    {
+                        bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
+
+                        Thread.Sleep(delayRandom.Next(0, 10));
+                    }
+                });
+
+            for (var i = 0; i < MessageCount; i++)
+            {
+                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
+                {
+                    await bus.DequeueItemAsync(timeout.Token);
+                }
+            }
+
+            Assert.AreEqual(0, bus.TotalItemsCount);
+        }
+
         [TestMethod]
         [ExpectedException(typeof(TaskCanceledException))]
         public async Task Wait_With_Empty_Bus()
@@ -25,127 +147,5 @@ namespace MQTTnet.Tests.Internal
                 await bus.DequeueItemAsync(timeout.Token);
             }
         }
-        
-        [TestMethod]
-        public async Task Fill_From_Different_Task()
-        {
-            const int MessageCount = 500;
-
-            var delayRandom = new Random();
-            
-            var bus = new MqttPacketBus();
-
-            _ = Task.Run(
-                () =>
-                {
-                    for (int i = 0; i < MessageCount; i++)
-                    {
-                        bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
-                        
-                        Thread.Sleep(delayRandom.Next(0, 10));
-                    }
-                });
-            
-            for (var i = 0; i < MessageCount; i++)
-            {
-                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
-                {
-                    await bus.DequeueItemAsync(timeout.Token);
-                }    
-            }
-            
-            Assert.AreEqual(0, bus.TotalItemsCount);
-        }
-        
-        [TestMethod]
-        public void Alternate_Priorities()
-        {
-            var bus = new MqttPacketBus();
-            
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
-            
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
-
-            Assert.AreEqual(9, bus.TotalItemsCount);
-            
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
-            Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
-            
-            Assert.AreEqual(0, bus.TotalItemsCount);
-        }
-        
-        [TestMethod]
-        public void Export_Packets_Without_Dequeue()
-        {
-            var bus = new MqttPacketBus();
-
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
-            
-            Assert.AreEqual(3, bus.TotalItemsCount);
-
-            var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
-            Assert.AreEqual(0, exportedPackets.Count);
-            
-            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
-            Assert.AreEqual(0, exportedPackets.Count);
-            
-            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
-            Assert.AreEqual(3, exportedPackets.Count);
-            
-            Assert.AreEqual(3, bus.TotalItemsCount);
-        }
-        
-        [TestMethod]
-        public void Await_Single_Packet()
-        {
-            var bus = new MqttPacketBus();
-
-            var delivered = false;
-
-            var item1 = new MqttPacketBusItem(new MqttPublishPacket());
-            var item2 = new MqttPacketBusItem(new MqttPublishPacket());
-            
-            var item3 = new MqttPacketBusItem(new MqttPublishPacket());
-            item3.Delivered += (_, __) =>
-            {
-                delivered = true;
-            };
-            
-            bus.EnqueueItem(item1, MqttPacketBusPartition.Data);
-            bus.EnqueueItem(item2, MqttPacketBusPartition.Data);
-            bus.EnqueueItem(item3, MqttPacketBusPartition.Data);
-            
-            Assert.IsFalse(delivered);
-
-            bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
-            
-            Assert.IsFalse(delivered);
-            
-            bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
-            
-            Assert.IsFalse(delivered);
-            
-            bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
-            
-            // The third packet has the event attached.
-            Assert.IsTrue(delivered);
-        }
     }
 }

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

Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs

@@ -2,8 +2,10 @@
 // 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.Linq;
+using System.Text;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
@@ -17,7 +19,12 @@ namespace MQTTnet.Tests.Mockups
 
         public TestApplicationMessageReceivedHandler(IMqttClient mqttClient)
         {
-            mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;
+            if (mqttClient == null)
+            {
+                throw new ArgumentNullException(nameof(mqttClient));
+            }
+
+            mqttClient.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
         }
 
         public List<MqttApplicationMessageReceivedEventArgs> ReceivedEventArgs
@@ -39,7 +46,25 @@ namespace MQTTnet.Tests.Mockups
             }
         }
 
-        Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
+        public string GeneratePayloadSequence()
+        {
+            var sequence = new StringBuilder();
+
+            lock (_receivedEventArgs)
+            {
+                foreach (var receivedEventArg in _receivedEventArgs)
+                {
+                    var payload = receivedEventArg.ApplicationMessage.ConvertPayloadToString();
+
+                    // An empty payload is not part of the sequence!
+                    sequence.Append(payload);
+                }
+            }
+
+            return sequence.ToString();
+        }
+
+        Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
             lock (_receivedEventArgs)
             {

Source/MQTTnet.Tests/Mockups/TestEnvironment.cs

@@ -11,6 +11,7 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
+using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Extensions.Rpc;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
@@ -109,7 +110,7 @@ namespace MQTTnet.Tests.Mockups
             configureOptions.Invoke(optionsBuilder);
 
             var options = optionsBuilder.Build();
-            
+
             var client = CreateClient();
 
             if (timeout == TimeSpan.Zero)
@@ -137,7 +138,7 @@ namespace MQTTnet.Tests.Mockups
             options = options.WithTcpServer("127.0.0.1", ServerPort);
 
             var client = CreateClient();
-            
+
             if (timeout == TimeSpan.Zero)
             {
                 await client.ConnectAsync(options.Build()).ConfigureAwait(false);
@@ -161,7 +162,7 @@ namespace MQTTnet.Tests.Mockups
             }
 
             var client = CreateClient();
-            
+
             if (timeout == TimeSpan.Zero)
             {
                 await client.ConnectAsync(options).ConfigureAwait(false);
@@ -199,6 +200,16 @@ namespace MQTTnet.Tests.Mockups
             return new TestApplicationMessageReceivedHandler(mqttClient);
         }
 
+        public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IManagedMqttClient managedClient)
+        {
+            if (managedClient == null)
+            {
+                throw new ArgumentNullException(nameof(managedClient));
+            }
+
+            return new TestApplicationMessageReceivedHandler(managedClient.InternalClient);
+        }
+
         public IMqttClient CreateClient()
         {
             lock (_clients)
@@ -325,7 +336,7 @@ namespace MQTTnet.Tests.Mockups
             var options = optionsBuilder.Build();
             var server = CreateServer(options);
             await server.StartAsync();
-            
+
             // The OS has chosen the port to we have to properly expose it to the tests.
             ServerPort = options.DefaultEndpointOptions.Port;
             return server;
@@ -344,7 +355,7 @@ namespace MQTTnet.Tests.Mockups
             var options = optionsBuilder.Build();
             var server = CreateServer(options);
             await server.StartAsync();
-            
+
             // The OS has chosen the port to we have to properly expose it to the tests.
             ServerPort = options.DefaultEndpointOptions.Port;
             return server;

Source/MQTTnet.Tests/Server/Load_Tests.cs

@@ -55,7 +55,7 @@ namespace MQTTnet.Tests.Server
                             });
                     }
 
-                    SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(90));
+                    SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(120));
 
                     Assert.AreEqual(100000, receivedMessages);
                 }

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

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -15,6 +15,57 @@ namespace MQTTnet.Tests.Server
     [TestClass]
     public sealed class QoS_Tests : BaseTestClass
     {
+        [TestMethod]
+        public async Task Preserve_Message_Order_For_Queued_Messages()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                // Create a session which will contain the messages.
+                var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
+                await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
+                dummyClient.Dispose();
+                
+                // Now inject messages which are appended to the queue of the client.
+                await server.InjectApplicationMessage("T", "0", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await server.InjectApplicationMessage("T", "2", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                await server.InjectApplicationMessage("T", "1", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await server.InjectApplicationMessage("T", "4", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                await server.InjectApplicationMessage("T", "3", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await server.InjectApplicationMessage("T", "6", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                await server.InjectApplicationMessage("T", "5", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await server.InjectApplicationMessage("T", "8", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                await server.InjectApplicationMessage("T", "7", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await server.InjectApplicationMessage("T", "9", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                
+                await LongTestDelay();
+                
+                // Create a new client for the existing message.
+                var client = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
+                var messages = testEnvironment.CreateApplicationMessageHandler(client);
+                
+                await LongTestDelay();
+                
+                var payloadSequence = messages.GeneratePayloadSequence();
+                Assert.AreEqual("0214365879", payloadSequence);
+
+                // Disconnect and reconnect to make sure that the server will not send the messages twice.
+                await client.DisconnectAsync();
+                await LongTestDelay();
+                await client.ReconnectAsync();
+                await LongTestDelay();
+                
+                payloadSequence = messages.GeneratePayloadSequence();
+                Assert.AreEqual("0214365879", payloadSequence);
+            }
+        }
+        
         [TestMethod]
         public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
         {

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -150,7 +150,7 @@ namespace MQTTnet.Tests.Server
                 // Try to connect 50 clients at the same time.
                 var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
 
-                await Task.Delay(5000);
+                await Task.Delay(TimeSpan.FromSeconds(10));
                 
                 var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList();
 
@@ -162,7 +162,7 @@ namespace MQTTnet.Tests.Server
                 var sendClient = await testEnvironment.ConnectClient(option2);
                 await sendClient.PublishStringAsync("aaa", "1");
 
-                await Task.Delay(3000);
+                await Task.Delay(TimeSpan.FromSeconds(5));
 
                 Assert.AreEqual(true, hasReceive);
             }

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -86,7 +86,7 @@ namespace MQTTnet.Adapter
                     await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
                     if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
                     {
-                        throw new OperationCanceledException("MQTT connect cancelled.", cancellationToken);
+                        throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
                     }
 
                     // Make sure that the exception from the connect task gets thrown.
@@ -298,7 +298,7 @@ namespace MQTTnet.Adapter
                 // Check two times for cancellation because the call to _ReadAsync_ might block for some time.
                 if (cancellationToken.IsCancellationRequested)
                 {
-                    return ReadFixedHeaderResult.Cancelled;
+                    return ReadFixedHeaderResult.Canceled;
                 }
 
                 int bytesRead;
@@ -308,7 +308,7 @@ namespace MQTTnet.Adapter
                 }
                 catch (OperationCanceledException)
                 {
-                    return ReadFixedHeaderResult.Cancelled;
+                    return ReadFixedHeaderResult.Canceled;
                 }
                 catch (SocketException)
                 {
@@ -317,7 +317,7 @@ namespace MQTTnet.Adapter
                
                 if (cancellationToken.IsCancellationRequested)
                 {
-                    return ReadFixedHeaderResult.Cancelled;
+                    return ReadFixedHeaderResult.Canceled;
                 }
 
                 if (bytesRead == 0)

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -14,26 +14,6 @@ namespace MQTTnet.Client
 {
     public static class MqttClientExtensions
     {
-        public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
-        {
-            if (client == null)
-            {
-                throw new ArgumentNullException(nameof(client));
-            }
-
-            try
-            {
-                await client.PingAsync(cancellationToken).ConfigureAwait(false);
-                return true;
-            }
-            catch
-            {
-                // Ignore errors.
-            }
-
-            return false;
-        }
-        
         public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null)
         {
             if (client == null)
@@ -89,6 +69,17 @@ namespace MQTTnet.Client
             return mqttClient.PublishBinaryAsync(topic, payloadBuffer, qualityOfServiceLevel, retain, cancellationToken);
         }
 
+        public static Task ReconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default)
+        {
+            if (client.Options == null)
+            {
+                throw new InvalidOperationException(
+                    "The MQTT client was not connected before. A reconnect is only permitted when the client was already connected or at least tried to.");
+            }
+
+            return client.ConnectAsync(client.Options, cancellationToken);
+        }
+
         public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
         {
             if (client == null)
@@ -137,6 +128,26 @@ namespace MQTTnet.Client
             return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken);
         }
 
+        public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
+        {
+            if (client == null)
+            {
+                throw new ArgumentNullException(nameof(client));
+            }
+
+            try
+            {
+                await client.PingAsync(cancellationToken).ConfigureAwait(false);
+                return true;
+            }
+            catch
+            {
+                // Ignore errors.
+            }
+
+            return false;
+        }
+
         public static Task<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient mqttClient, string topic, CancellationToken cancellationToken = default)
         {
             if (mqttClient == null)

Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs

@@ -10,29 +10,38 @@ namespace MQTTnet.Client
     public sealed class MqttClientPublishResult
     {
         /// <summary>
-        /// 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.
+        /// </summary>
+        public bool IsSuccess => ReasonCode == MqttClientPublishReasonCode.Success || ReasonCode == MqttClientPublishReasonCode.NoMatchingSubscribers;
+
+        /// <summary>
+        ///     Gets the packet identifier which was used for this publish.
         /// </summary>
         public ushort? PacketIdentifier { get; set; }
 
         /// <summary>
-        /// Gets or sets the reason code.
-        /// Hint: MQTT 5 feature only.
+        ///     Gets or sets the reason code.
+        ///     Hint: MQTT 5 feature only.
         /// </summary>
         public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success;
 
         /// <summary>
-        /// Gets or sets the reason string.
-        /// Hint: MQTT 5 feature only.
+        ///     Gets or sets the reason string.
+        ///     Hint: MQTT 5 feature only.
         /// </summary>
         public string ReasonString { get; set; }
 
         /// <summary>
-        /// 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.
         /// </summary>
         public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
     }
-}
+}

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;

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;

Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs

@@ -6,9 +6,9 @@ namespace MQTTnet.Formatter
 {
     public struct ReadFixedHeaderResult
     {
-        public static ReadFixedHeaderResult Cancelled { get; } = new ReadFixedHeaderResult
+        public static ReadFixedHeaderResult Canceled { get; } = new ReadFixedHeaderResult
         {
-            IsCancelled = true
+            IsCanceled = true
         };
         
         public static ReadFixedHeaderResult ConnectionClosed { get; } = new ReadFixedHeaderResult
@@ -16,7 +16,7 @@ namespace MQTTnet.Formatter
             IsConnectionClosed = true
         };
         
-        public bool IsCancelled { get; set; }
+        public bool IsCanceled { get; set; }
         
         public bool IsConnectionClosed { get; set; }
 

Source/MQTTnet/Internal/AsyncQueue.cs

@@ -29,7 +29,7 @@ namespace MQTTnet.Internal
         {
             lock (_syncRoot)
             {
-                _signal?.Dispose();
+                _signal.Dispose();
 
                 _isDisposed = true;
             }
@@ -48,10 +48,10 @@ namespace MQTTnet.Internal
         {
             if (_queue.TryDequeue(out var item))
             {
-                return new AsyncQueueDequeueResult<TItem>(true, item);
+                return AsyncQueueDequeueResult<TItem>.Success(item);
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default);
+            return AsyncQueueDequeueResult<TItem>.NonSuccess;
         }
 
         public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -60,36 +60,42 @@ namespace MQTTnet.Internal
             {
                 try
                 {
-                    Task task;
+                    Task task = null;
                     lock (_syncRoot)
                     {
                         if (_isDisposed)
                         {
-                            return new AsyncQueueDequeueResult<TItem>(false, default);
+                            return AsyncQueueDequeueResult<TItem>.NonSuccess;
                         }
 
-                        task = _signal.WaitAsync(cancellationToken);
+                        if (_queue.IsEmpty)
+                        {
+                            task = _signal.WaitAsync(cancellationToken);
+                        }
                     }
 
-                    await task.ConfigureAwait(false);
-
+                    if (task != null)
+                    {
+                        await task.ConfigureAwait(false);    
+                    }
+                    
                     if (cancellationToken.IsCancellationRequested)
                     {
-                        return new AsyncQueueDequeueResult<TItem>(false, default);
+                        return AsyncQueueDequeueResult<TItem>.NonSuccess;
                     }
 
                     if (_queue.TryDequeue(out var item))
                     {
-                        return new AsyncQueueDequeueResult<TItem>(true, item);
+                        return AsyncQueueDequeueResult<TItem>.Success(item);
                     }
                 }
                 catch (OperationCanceledException)
                 {
-                    return new AsyncQueueDequeueResult<TItem>(false, default);
+                    return AsyncQueueDequeueResult<TItem>.NonSuccess;
                 }
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default);
+            return AsyncQueueDequeueResult<TItem>.NonSuccess;
         }
     }
 }

Source/MQTTnet/Internal/AsyncQueueDequeueResult.cs

@@ -4,8 +4,10 @@
 
 namespace MQTTnet.Internal
 {
-    public class AsyncQueueDequeueResult<TItem>
+    public sealed class AsyncQueueDequeueResult<TItem>
     {
+        public static readonly AsyncQueueDequeueResult<TItem> NonSuccess = new AsyncQueueDequeueResult<TItem>(false, default);
+
         public AsyncQueueDequeueResult(bool isSuccess, TItem item)
         {
             IsSuccess = isSuccess;
@@ -15,5 +17,10 @@ namespace MQTTnet.Internal
         public bool IsSuccess { get; }
 
         public TItem Item { get; }
+
+        public static AsyncQueueDequeueResult<TItem> Success(TItem item)
+        {
+            return new AsyncQueueDequeueResult<TItem>(true, item);
+        }
     }
 }

Source/MQTTnet/Internal/AsyncSignal.cs

@@ -33,9 +33,17 @@ namespace MQTTnet.Internal
             {
                 _isSignaled = true;
 
+                Cleanup();
+                
                 // If there is already a waiting task let it run.
-                _waiter?.Approve();
-                _waiter = null;
+                if (_waiter != null)
+                {
+                    _waiter.Approve();
+                    _waiter = null;
+
+                    // Since we already got a waiter the signal must be reset right now!
+                    _isSignaled = false;
+                }
             }
         }
 
@@ -47,8 +55,11 @@ namespace MQTTnet.Internal
             {
                 ThrowIfDisposed();
 
+                Cleanup();
+
                 if (_isSignaled)
                 {
+                    _isSignaled = false;
                     return CompletedTask.Instance;
                 }
 
@@ -58,7 +69,7 @@ namespace MQTTnet.Internal
                     {
                         throw new InvalidOperationException("Only one waiting task is permitted per async signal.");
                     }
-                    
+
                     _waiter.Dispose();
                 }
 
@@ -67,6 +78,15 @@ namespace MQTTnet.Internal
             }
         }
 
+        void Cleanup()
+        {
+            // Cleanup if the previous waiter was cancelled.
+            if (_waiter != null && _waiter.Task.IsCanceled)
+            {
+                _waiter = null;
+            }
+        }
+
         void ThrowIfDisposed()
         {
             if (_isDisposed)

Source/MQTTnet/Internal/MqttPacketBusItem.cs

@@ -10,40 +10,36 @@ namespace MQTTnet.Internal
 {
     public sealed class MqttPacketBusItem
     {
-        readonly AsyncTaskCompletionSource<int> _promise = new AsyncTaskCompletionSource<int>();
-
+        readonly AsyncTaskCompletionSource<bool> _promise = new AsyncTaskCompletionSource<bool>();
+        
         public MqttPacketBusItem(MqttPacket packet)
         {
             Packet = packet ?? throw new ArgumentNullException(nameof(packet));
         }
-        
+
+        public event EventHandler Completed;
+
         public MqttPacket Packet { get; }
 
-        public event EventHandler Delivered;
-
-        public Task WaitForDeliveryAsync()
-        {
-            return _promise.Task;
-        }
-        
-        public void MarkAsDelivered()
-        {
-            if (_promise.TrySetResult(0))
-            {
-                Delivered?.Invoke(this, EventArgs.Empty);
-            }
-        }
-
-        public void MarkAsFailed(Exception exception)
-        {
-            if (exception == null) throw new ArgumentNullException(nameof(exception));
-            
-            _promise.TrySetException(exception);
-        }
-
-        public void MarkAsCancelled()
+        public void Cancel()
         {
             _promise.TrySetCanceled();
         }
+
+        public void Complete()
+        {
+            _promise.TrySetResult(true);
+            Completed?.Invoke(this, EventArgs.Empty);
+        }
+
+        public void Fail(Exception exception)
+        {
+            _promise.TrySetException(exception);
+        }
+
+        public Task WaitAsync()
+        {
+            return _promise.Task;
+        }
     }
 }

Source/MQTTnet/MqttApplicationMessageExtensions.cs

@@ -5,13 +5,78 @@
 using System;
 using System.Text;
 
+// ReSharper disable MemberCanBePrivate.Global
+
 namespace MQTTnet
 {
     public static class MqttApplicationMessageExtensions
     {
+        public static short ConvertPayloadToByte(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            if (applicationMessage.Payload.Length != 1)
+            {
+                throw new ArgumentException("Payload requires to be 1 bytes long.");
+            }
+
+            return applicationMessage.Payload[0];
+        }
+
+        public static short ConvertPayloadToInt16(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            if (applicationMessage.Payload.Length != 2)
+            {
+                throw new ArgumentException("Payload requires to be 2 bytes long.");
+            }
+
+            return BitConverter.ToInt16(applicationMessage.Payload, 0);
+        }
+
+        public static int ConvertPayloadToInt32(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            if (applicationMessage.Payload?.Length != 4)
+            {
+                throw new ArgumentException("Payload requires to be 4 bytes long.");
+            }
+
+            return BitConverter.ToInt32(applicationMessage.Payload, 0);
+        }
+
+        public static long ConvertPayloadToInt64(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            if (applicationMessage.Payload.Length != 8)
+            {
+                throw new ArgumentException("Payload requires to be 8 bytes long.");
+            }
+
+            return BitConverter.ToInt64(applicationMessage.Payload, 0);
+        }
+
         public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
 
             if (applicationMessage.Payload == null)
             {
@@ -25,5 +90,65 @@ namespace MQTTnet
 
             return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length);
         }
+
+        public static decimal ParsePayloadToDecimal(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return decimal.Parse(ConvertPayloadToString(applicationMessage));
+        }
+
+        public static double ParsePayloadToDouble(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return double.Parse(ConvertPayloadToString(applicationMessage));
+        }
+
+        public static short ParsePayloadToInt16(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return short.Parse(ConvertPayloadToString(applicationMessage));
+        }
+
+        public static int ParsePayloadToInt32(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return int.Parse(ConvertPayloadToString(applicationMessage));
+        }
+
+        public static long ParsePayloadToInt64(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return long.Parse(ConvertPayloadToString(applicationMessage));
+        }
+
+        public static float ParsePayloadToSingle(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return float.Parse(ConvertPayloadToString(applicationMessage));
+        }
     }
-}
+}

Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs

@@ -7,7 +7,12 @@ namespace MQTTnet.Protocol
     public enum MqttPubAckReasonCode
     {
         Success = 0,
+        
+        /// <summary>
+        /// 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).
+        /// </summary>
         NoMatchingSubscribers = 16,
+        
         UnspecifiedError = 128,
         ImplementationSpecificError = 131,
         NotAuthorized = 135,

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

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;
                 }
@@ -506,15 +507,15 @@ namespace MQTTnet.Server
                     try
                     {
                         await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
-                        packetBusItem.MarkAsDelivered();
+                        packetBusItem.Complete();
                     }
                     catch (OperationCanceledException)
                     {
-                        packetBusItem.MarkAsCancelled();
+                        packetBusItem.Cancel();
                     }
                     catch (Exception exception)
                     {
-                        packetBusItem.MarkAsFailed(exception);
+                        packetBusItem.Fail(exception);
                     }
                     finally
                     {
@@ -562,7 +563,7 @@ namespace MQTTnet.Server
             catch (ObjectDisposedException)
             {
                 // This can happen when connections are created and dropped very quickly.
-                // It is not an issue if the cancellation token cannot be cancelled multiple times.
+                // It is not an issue if the cancellation token cannot be canceled multiple times.
             }
         }
 

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<DispatchApplicationMessageResult> DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage)
         {
+            var matchingSubscribersCount = 0;
             try
             {
                 if (applicationMessage.Retain)
                 {
                     await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
                 }
-
-                var deliveryCount = 0;
+                
                 List<MqttSession> 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;

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -4,7 +4,6 @@
 
 using System;
 using System.Collections;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -25,7 +24,8 @@ namespace MQTTnet.Server
         readonly MqttServerOptions _serverOptions;
         readonly MqttClientSubscriptionsManager _subscriptionsManager;
 
-        readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
+        // Do not use a dictionary in order to keep the ordering of the messages.
+        readonly List<MqttPublishPacket> _unacknowledgedPublishPackets = new List<MqttPublishPacket>();
 
         // Bookkeeping to know if this is a subscribing client; lazy initialize later.
         HashSet<string> _subscribedTopics;
@@ -72,7 +72,14 @@ namespace MQTTnet.Server
 
         public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
         {
-            _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
+            MqttPublishPacket publishPacket;
+
+            lock (_unacknowledgedPublishPackets)
+            {
+                publishPacket = _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
+                _unacknowledgedPublishPackets.Remove(publishPacket);
+            }
+
             return publishPacket;
         }
 
@@ -130,7 +137,10 @@ namespace MQTTnet.Server
             {
                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
 
-                _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
+                lock (_unacknowledgedPublishPackets)
+                {
+                    _unacknowledgedPublishPackets.Add(publishPacket);
+                }
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
@@ -145,8 +155,10 @@ namespace MQTTnet.Server
         {
             // 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;
+            lock (_unacknowledgedPublishPackets)
+            {
+                return _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
+            }
         }
 
         public void Recover()
@@ -154,9 +166,33 @@ namespace MQTTnet.Server
             // TODO: Keep the bus and only insert pending items again.
             // TODO: Check if packet identifier must be restarted or not.
             // TODO: Recover package identifier.
+            
+            /*
+                The Session state in the Client consists of:
+                ·         QoS 1 and QoS 2 messages which have been sent to the Server, but have not been completely acknowledged.
+                ·         QoS 2 messages which have been received from the Server, but have not been completely acknowledged. 
+
+                The Session state in the Server consists of:
+                ·         The existence of a Session, even if the rest of the Session state is empty.
+                ·         The Client’s subscriptions.
+                ·         QoS 1 and QoS 2 messages which have been sent to the Client, but have not been completely acknowledged.
+                ·         QoS 1 and QoS 2 messages pending transmission to the Client.
+                ·         QoS 2 messages which have been received from the Client, but have not been completely acknowledged.
+                ·         Optionally, QoS 0 messages pending transmission to the Client. 
+             */
+
+            // Create a copy of all currently unacknowledged publish packets and clear the storage.
+            // We must re-enqueue them in order to trigger other code.
+            List<MqttPublishPacket> unacknowledgedPublishPackets;
+            lock (_unacknowledgedPublishPackets)
+            {
+                unacknowledgedPublishPackets = _unacknowledgedPublishPackets.ToList();
+                _unacknowledgedPublishPackets.Clear();    
+            }
+
             _packetBus.Clear();
 
-            foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())
+            foreach (var publishPacket in unacknowledgedPublishPackets)
             {
                 EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
             }

Source/MQTTnet/Server/MqttServerExtensions.cs

@@ -3,29 +3,89 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Text;
 using System.Threading.Tasks;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
     public static class MqttServerExtensions
     {
+        public static Task InjectApplicationMessage(
+            this MqttServer server,
+            string topic,
+            string payload = null,
+            MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
+            bool retain = false)
+        {
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
+
+            if (topic == null)
+            {
+                throw new ArgumentNullException(nameof(topic));
+            }
+
+            var payloadBuffer = EmptyBuffer.Array;
+            if (payload is string stringPayload)
+            {
+                payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
+            }
+
+            return server.InjectApplicationMessage(
+                new InjectedMqttApplicationMessage(
+                    new MqttApplicationMessage
+                    {
+                        Topic = topic,
+                        Payload = payloadBuffer,
+                        QualityOfServiceLevel = qualityOfServiceLevel,
+                        Retain = retain
+                    }));
+        }
+
         public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
         {
-            if (server == null) throw new ArgumentNullException(nameof(server));
-            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
+
+            if (clientId == null)
+            {
+                throw new ArgumentNullException(nameof(clientId));
+            }
+
+            if (topicFilters == null)
+            {
+                throw new ArgumentNullException(nameof(topicFilters));
+            }
 
             return server.SubscribeAsync(clientId, topicFilters);
         }
-        
+
         public static Task SubscribeAsync(this MqttServer server, string clientId, string topic)
         {
-            if (server == null) throw new ArgumentNullException(nameof(server));
-            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
-            if (topic == null) throw new ArgumentNullException(nameof(topic));
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
 
-            return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
+            if (clientId == null)
+            {
+                throw new ArgumentNullException(nameof(clientId));
+            }
+
+            if (topic == null)
+            {
+                throw new ArgumentNullException(nameof(topic));
+            }
+
+            var topicFilters = new MqttTopicFilterBuilder().WithTopic(topic).Build();
+            return server.SubscribeAsync(clientId, topicFilters);
         }
     }
-}
+}

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -45,7 +45,7 @@ namespace MQTTnet.Server
             var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage));
             _session.EnqueueDataPacket(packetBusItem);
 
-            return packetBusItem.WaitForDeliveryAsync();
+            return packetBusItem.WaitAsync();
         }
         
         public Task DeleteAsync()