Add unit test to reproduce issue

.github/workflows/ReleaseNotes.md

@@ -1,2 +1,5 @@
-* [Core] Fixed dead lock and race conditions in new _AsyncLock_ implementation (#1542).
-* [Client] Fixed connection freeze when using Xamarin etc.
+* [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)

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.1.2.${{github.run_number}}"
+  VERSION: "4.1.3.${{github.run_number}}"
 
 jobs:
   build:

MQTTnet.sln.DotSettings

@@ -242,5 +242,6 @@ See the LICENSE file in the project root for more information.</s:String>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBCOMP/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Tnet/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

Samples/Server/Server_Intercepting_Samples.cs

@@ -0,0 +1,47 @@
+// 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.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+// ReSharper disable EmptyConstructor
+// ReSharper disable MemberCanBeMadeStatic.Local
+
+using MQTTnet.Internal;
+using MQTTnet.Server;
+
+namespace MQTTnet.Samples.Server;
+
+public static class Server_Intercepting_Samples
+{
+    public static async Task Intercept_Application_Messages()
+    {
+        /*
+         * This sample starts a simple MQTT server which manipulate all processed application messages.
+         * Please see _Server_Simple_Samples_ for more details on how to start a server.
+         */
+
+        var mqttFactory = new MqttFactory();
+        var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
+
+        using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
+        {
+            mqttServer.InterceptingPublishAsync += args =>
+            {
+                // Here we only change the topic of the received application message.
+                // but also changing the payload etc. is required. Changing the QoS after
+                // transmitting is not supported and makes no sense at all.
+                args.ApplicationMessage.Topic += "/manipulated";
+
+                return CompletedTask.Instance;
+            };
+
+            await mqttServer.StartAsync();
+
+            Console.WriteLine("Press Enter to exit.");
+            Console.ReadLine();
+            await mqttServer.StopAsync();
+        }
+    }
+}

Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs

@@ -1,26 +0,0 @@
-// 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.Collections.Generic;
-using System.IO.Pipelines;
-using Microsoft.AspNetCore.Connections;
-using Microsoft.AspNetCore.Http.Features;
-
-namespace MQTTnet.AspNetCore.Tests.Mockups
-{
-    public class ConnectionContextMockup : ConnectionContext
-    {
-        public override string ConnectionId { get; set; }
-
-        public override IFeatureCollection Features { get; }
-
-        public override IDictionary<object, object> Items { get; set; }
-        public override IDuplexPipe Transport { get; set; }
-
-        public ConnectionContextMockup()
-        {
-            //Transport = new DefaultConnectionContext
-        }
-    }
-}

Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs

@@ -2,20 +2,24 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using Microsoft.AspNetCore.Connections;
-using MQTTnet.Adapter;
-using MQTTnet.Formatter;
-using MQTTnet.Server;
 using System;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
+using MQTTnet.Formatter;
+using MQTTnet.Server;
 
 namespace MQTTnet.AspNetCore.Tests.Mockups
 {
-    public class ConnectionHandlerMockup: IMqttServerAdapter
+    public sealed class ConnectionHandlerMockup : IMqttServerAdapter
     {
-        public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
         public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
+        public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
+
+        public void Dispose()
+        {
+        }
 
         public async Task OnConnectedAsync(ConnectionContext connection)
         {
@@ -42,9 +46,5 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
         {
             return Task.CompletedTask;
         }
-
-        public void Dispose()
-        {
-        }
     }
-}
+}

Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs

@@ -6,7 +6,7 @@ using System.IO.Pipelines;
 
 namespace MQTTnet.AspNetCore.Tests.Mockups
 {
-    public class DuplexPipeMockup : IDuplexPipe
+    public sealed class DuplexPipeMockup : IDuplexPipe
     {
         public DuplexPipeMockup()
         {
@@ -20,9 +20,8 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
 
         PipeWriter IDuplexPipe.Output => Send.Writer;
 
+        public Pipe Receive { get; } 
 
-        public Pipe Receive { get; set; } 
-
-        public Pipe Send { get; set; }
+        public Pipe Send { get; }
     }
 }

Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs

@@ -6,7 +6,7 @@ using System.Buffers;
 
 namespace MQTTnet.AspNetCore.Tests.Mockups
 {
-    public class LimitedMemoryPool : MemoryPool<byte>
+    public sealed class LimitedMemoryPool : MemoryPool<byte>
     {
         protected override void Dispose(bool disposing)
         {

Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs

@@ -7,9 +7,9 @@ using System.Buffers;
 
 namespace MQTTnet.AspNetCore.Tests.Mockups
 {
-    public class MemoryOwner : IMemoryOwner<byte>
+    public sealed class MemoryOwner : IMemoryOwner<byte>
     {
-        private readonly byte[] _raw;
+        readonly byte[] _raw;
 
         public MemoryOwner(int size)
         {
@@ -27,11 +27,11 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
             Memory = _raw;
         }
 
+        public Memory<byte> Memory { get; }
+
         public void Dispose()
         {
             ArrayPool<byte>.Shared.Return(_raw);
         }
-
-        public Memory<byte> Memory { get; }
     }
 }

Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -2,22 +2,21 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using Microsoft.AspNetCore.Builder;
-using Microsoft.AspNetCore.Connections;
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.VisualStudio.TestTools.UnitTesting;
-
 using System;
 using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.AspNetCore.Tests.Mockups;
+using MQTTnet.Client;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
-using MQTTnet.Client;
 using MQTTnet.Protocol;
 using MQTTnet.Tests.Extensions;
 
@@ -26,20 +25,6 @@ namespace MQTTnet.AspNetCore.Tests
     [TestClass]
     public class MqttConnectionContextTest
     {
-        [TestMethod]
-        public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted()
-        {
-            var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
-            var pipe = new DuplexPipeMockup();
-            var connection = new DefaultConnectionContext();
-            connection.Transport = pipe;
-            var ctx = new MqttConnectionContext(serializer, connection);
-
-            pipe.Receive.Writer.Complete();
-
-            await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
-        }
-
         [TestMethod]
         public async Task TestCorruptedConnectPacket()
         {
@@ -49,13 +34,43 @@ namespace MQTTnet.AspNetCore.Tests
             var connection = new DefaultConnectionContext();
             connection.Transport = pipe;
             var ctx = new MqttConnectionContext(serializer, connection);
-            
+
             await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0]));
 
             await Assert.ThrowsExceptionAsync<MqttProtocolViolationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
 
             // the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly
-            await Assert.ThrowsExceptionAsync<InvalidOperationException>(() =>  ctx.ReceivePacketAsync(CancellationToken.None));
+            await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
+        }
+
+        [TestMethod]
+        public async Task TestEndpoint()
+        {
+            var mockup = new ConnectionHandlerMockup();
+
+
+            using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
+                       .UseStartup<Startup>()
+                       .ConfigureServices(
+                           (hostContext, services) =>
+                           {
+                               services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
+                               services.AddSingleton<IMqttServerAdapter>(mockup);
+                           })
+                       .Build())
+
+            using (var client = new MqttFactory().CreateMqttClient())
+            {
+                host.Start();
+                await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build(), CancellationToken.None);
+
+                var ctx = await mockup.Context.Task;
+#if NETCOREAPP3_1
+                var ep = IPEndPoint.Parse(ctx.Endpoint);
+                Assert.IsNotNull(ep);
+#endif
+                Assert.IsNotNull(ctx);
+            }
         }
 
         // COMMENTED OUT DUE TO DEAD LOCK? OR VERY VERY SLOW PERFORMANCE ON LOCAL DEV MACHINE. TEST WAS STILL RUNNING AFTER SEVERAL MINUTES!
@@ -78,7 +93,7 @@ namespace MQTTnet.AspNetCore.Tests
 
         //    await Task.WhenAll(tasks).ConfigureAwait(false);
         //}
-        
+
         [TestMethod]
         public async Task TestLargePacket()
         {
@@ -94,43 +109,25 @@ namespace MQTTnet.AspNetCore.Tests
             Assert.IsTrue(readResult.Buffer.Length > 20000);
         }
 
-        private class Startup 
+        [TestMethod]
+        public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted()
+        {
+            var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
+            var pipe = new DuplexPipeMockup();
+            var connection = new DefaultConnectionContext();
+            connection.Transport = pipe;
+            var ctx = new MqttConnectionContext(serializer, connection);
+
+            pipe.Receive.Writer.Complete();
+
+            await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
+        }
+
+        class Startup
         {
             public void Configure(IApplicationBuilder app)
-            { 
+            {
             }
         }
-
-        [TestMethod]
-        public async Task TestEndpoint()
-        {
-            var mockup = new ConnectionHandlerMockup();
-
-
-            using (var host = new WebHostBuilder()
-                .UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
-                .UseStartup<Startup>()
-                .ConfigureServices((hostContext, services) =>
-                {
-                    services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
-                    services.AddSingleton<IMqttServerAdapter>(mockup);
-                })
-                .Build())
-                
-            using (var client = new MqttFactory().CreateMqttClient())
-            {
-                host.Start();
-                await client.ConnectAsync(new MqttClientOptionsBuilder()
-                    .WithTcpServer("localhost")
-                    .Build(), CancellationToken.None);
-
-                var ctx = await mockup.Context.Task;
-#if NETCOREAPP3_1
-                var ep = IPEndPoint.Parse(ctx.Endpoint);
-                Assert.IsNotNull(ep);
-#endif
-                Assert.IsNotNull(ctx);
-            }               
-        }
     }
-}
+}

Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs

@@ -12,7 +12,7 @@ using MQTTnet.Packets;
 namespace MQTTnet.AspNetCore.Tests
 {
     [TestClass]
-    public class ReaderExtensionsTest
+    public sealed class ReaderExtensionsTest
     {
         [TestMethod]
         public void TestTryDeserialize()

Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs

@@ -21,6 +21,52 @@ namespace MQTTnet.Tests.Client
     [TestClass]
     public sealed class ManagedMqttClient_Tests : BaseTestClass
     {
+        [TestMethod]
+        public async Task Subscribe_When_Offline_But_Existing_Session()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                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");
+
+                MqttApplicationMessage applicationMessage = null;
+                managedClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    applicationMessage = e.ApplicationMessage;
+                    return CompletedTask.Instance;
+                };
+                
+                // 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 LongTestDelay();
+                
+                await server.InjectApplicationMessage(
+                    new InjectedMqttApplicationMessage(
+                        new MqttApplicationMessage
+                        {
+                            Topic = "a"
+                        }));
+
+                await LongTestDelay();
+                
+                Assert.IsNotNull(applicationMessage);
+            }
+        }
+        
         [TestMethod]
         public async Task Connect_To_Invalid_Server()
         {
@@ -29,7 +75,6 @@ namespace MQTTnet.Tests.Client
                 testEnvironment.IgnoreClientLogErrors = true;
 
                 var mqttClient = testEnvironment.CreateClient();
-
                 var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient);
 
                 ConnectingFailedEventArgs connectingFailedEventArgs = null;

Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs

@@ -13,13 +13,12 @@ using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
-using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Tests.Extensions;
 using MQTTnet.Tests.Mockups;
 
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Formatter
 {
     [TestClass]
     public sealed class MqttPacketSerialization_V3_Binary_Tests

Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs

@@ -10,7 +10,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Formatter
 {
     [TestClass]
     public sealed class MqttPacketSerialization_V3_Tests

Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs

@@ -9,7 +9,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Formatter
 {
     [TestClass]
     public sealed class MqttPacketSerialization_V5_Tests

Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs

@@ -0,0 +1,114 @@
+// 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;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Internal;
+
+namespace MQTTnet.Tests.Internal
+{
+    [TestClass]
+    public sealed class AsyncSignal_Tests
+    {
+        [TestMethod]
+        [ExpectedException(typeof(TaskCanceledException))]
+        public async Task Cancel_If_No_Signal()
+        {
+            var asyncSignal = new AsyncSignal();
+
+            using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2)))
+            {
+                await asyncSignal.WaitAsync(timeout.Token);
+
+                Assert.Fail("There is no signal. So we must fail here!");
+            }
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(ObjectDisposedException))]
+        public async Task Dispose_Properly()
+        {
+            var asyncSignal = new AsyncSignal();
+
+            // The timeout will not be reached but another task will kill the async signal.
+            using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(999)))
+            {
+                _ = Task.Run(
+                    async () =>
+                    {
+                        await Task.Delay(2000, CancellationToken.None);
+                        asyncSignal.Dispose();
+                    },
+                    CancellationToken.None);
+
+                await asyncSignal.WaitAsync(timeout.Token);
+
+                Assert.Fail("There is no signal. So we must fail here!");
+            }
+        }
+
+        [TestMethod]
+        public async Task Loop_Signal()
+        {
+            var asyncSignal = new AsyncSignal();
+
+            for (var i = 0; i < 10; i++)
+            {
+                asyncSignal.Set();
+                
+                // WaitAsync should return directly because the signal is available.
+                await asyncSignal.WaitAsync();
+            }
+        }
+
+        [TestMethod]
+        public async Task Signal()
+        {
+            var asyncSignal = new AsyncSignal();
+
+            // The timeout will not be reached but another task will kill the async signal.
+            using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
+            {
+                var stopwatch = Stopwatch.StartNew();
+
+                _ = Task.Run(
+                    async () =>
+                    {
+                        await Task.Delay(1000, CancellationToken.None);
+                        asyncSignal.Set();
+                    },
+                    CancellationToken.None);
+
+                await asyncSignal.WaitAsync(timeout.Token);
+
+                stopwatch.Stop();
+
+                Assert.IsTrue(stopwatch.ElapsedMilliseconds > 900);
+            }
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(InvalidOperationException))]
+        public async Task Fail_For_Two_Waiters()
+        {
+            var asyncSignal = new AsyncSignal();
+
+            // This thread will wait properly because it is the first waiter.
+            _ = Task.Run(
+                async () =>
+                {
+                    await asyncSignal.WaitAsync();
+                },
+                CancellationToken.None);
+
+            await Task.Delay(1000);
+            
+            // Now the current thread must fail because there is already a waiter.
+            await asyncSignal.WaitAsync();
+        }
+    }
+}

Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.Tests.Internal
     public sealed class MqttPacketBus_Tests
     {
         [TestMethod]
-        [ExpectedException(typeof(OperationCanceledException))]
+        [ExpectedException(typeof(TaskCanceledException))]
         public async Task Wait_With_Empty_Bus()
         {
             var bus = new MqttPacketBus();

Source/MQTTnet.Tests/Server/Load_Tests.cs

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

Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs

@@ -130,9 +130,7 @@ namespace MQTTnet.Tests.Server
 
         CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
         {
-            ulong topicHashMask; // not needed
-            bool hasWildcard; // not needed
-            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out topicHashMask, out hasWildcard);
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out _, out _);
             return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId);
         }
     }

Source/MQTTnet.Tests/Server/Publishing_Tests.cs

@@ -0,0 +1,31 @@
+// 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.Protocol;
+
+namespace MQTTnet.Tests.Server
+{
+    [TestClass]
+    public sealed class Publishing_Tests : BaseTestClass
+    {
+        [TestMethod]
+        [ExpectedException(typeof(MqttClientDisconnectedException))]
+        public async Task Disconnect_While_Publishing()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                // The client will be disconnect directly after subscribing!
+                server.InterceptingPublishAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
+
+                var client = await testEnvironment.ConnectClient();
+                await client.PublishStringAsync("test", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+            }
+        }
+    }
+}

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -99,25 +99,16 @@ namespace MQTTnet.Tests.Server
 
                 await LongTestDelay();
 
-                Assert.AreEqual(2, eventArgs.Count);
+                Assert.AreEqual(1, eventArgs.Count);
 
                 var firstEvent = eventArgs[0];
 
                 Assert.IsNotNull(firstEvent);
                 Assert.IsNotNull(firstEvent.PublishPacket);
                 Assert.IsNotNull(firstEvent.AcknowledgePacket);
-                Assert.IsFalse(firstEvent.IsCompleted);
+                Assert.IsTrue(firstEvent.IsCompleted);
 
                 Assert.AreEqual("A", firstEvent.PublishPacket.Topic);
-
-                var secondEvent = eventArgs[1];
-
-                Assert.IsNotNull(secondEvent);
-                Assert.IsNotNull(secondEvent.PublishPacket);
-                Assert.IsNotNull(secondEvent.AcknowledgePacket);
-                Assert.IsTrue(secondEvent.IsCompleted);
-
-                Assert.AreEqual("A", secondEvent.PublishPacket.Topic);
             }
         }
     }

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.Tests/Server/Subscribe_Tests.cs

@@ -46,6 +46,22 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        [ExpectedException(typeof(MqttClientDisconnectedException))]
+        public async Task Disconnect_While_Subscribing()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                // The client will be disconnect directly after subscribing!
+                server.ClientSubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
+
+                var client = await testEnvironment.ConnectClient();
+                await client.SubscribeAsync("#");
+            }
+        }
+        
         [TestMethod]
         public async Task Enqueue_Message_After_Subscription()
         {

Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs

@@ -1,11 +1,11 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Text;
-using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Client;
+using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -13,49 +13,301 @@ using MQTTnet.Tests.Mockups;
 namespace MQTTnet.Tests.Server
 {
     [TestClass]
-    public class Subscription_TopicHash_Tests
+    public sealed class SubscriptionTopicHashTests
     {
-        enum TopicHashSelector {  NoWildcard, SingleWildcard, MultiWildcard };
+        MqttSession _clientSession;
+
+        /// <summary>
+        ///     Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
+        /// </summary>
+        /// <returns></returns>
+        [TestMethod]
+        public void Check_Hash_Bucket_Depth()
+        {
+            const int numPublishers = 5000;
+            const int numTopicsPerPublisher = 10;
+
+            TopicGenerator.Generate(
+                numPublishers,
+                numTopicsPerPublisher,
+                out var topicsByPublisher,
+                out _,
+                out _);
+
+            // There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ...
+            // Hash bucket depths should remain low.
+            var bucketDepths = new Dictionary<ulong, int>();
+            var maxBucketDepth = 0;
+            ulong maxBucketDepthHash = 0;
+
+            var topicsByHash = new Dictionary<ulong, List<string>>();
+
+            foreach (var t in topicsByPublisher)
+            {
+                var topics = t.Value;
+                foreach (var topic in topics)
+                {
+                    MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
+
+                    bucketDepths.TryGetValue(topicHash, out var currentValue);
+                    ++currentValue;
+                    bucketDepths[topicHash] = currentValue;
+
+                    if (currentValue > maxBucketDepth)
+                    {
+                        maxBucketDepth = currentValue;
+                        maxBucketDepthHash = topicHash;
+                    }
+
+                    if (!topicsByHash.TryGetValue(topicHash, out var topicList))
+                    {
+                        topicList = new List<string>();
+                        topicsByHash.Add(topicHash, topicList);
+                    }
+
+                    topicList.Add(topic);
+                }
+            }
+
+            var maxDepthTopics = topicsByHash[maxBucketDepthHash];
+
+            Console.Write("Max bucket depth is " + maxBucketDepth);
+
+            // for the test case the bucket depth should be less than 100
+            Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth");
+        }
 
-        MQTTnet.Server.MqttSession _clientSession;
 
         [TestMethod]
-        public void Match_Hash_Test_SingleWildCard()
+        public void Check_Selected_Topic_Hashes()
         {
-            var l0 = "pub0";
-            var l1 = "topic1";
-            var l2 = "+";
-            var l3 = "prop1";
-            var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
+            CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF);
+            CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF);
+            CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF);
+            CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000);
+            CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000);
+        }
+
+        /// <summary>
+        ///     Test long topic name with last level being #
+        /// </summary>
+        [TestMethod]
+        public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard()
+        {
+            var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#";
+
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
+
+            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
+        }
 
 
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
+        /// <summary>
+        ///     Test long topic name with last level being +
+        /// </summary>
+        [TestMethod]
+        public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard()
+        {
+            var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+";
+
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
+
+            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
+        }
+
+        /// <summary>
+        ///     Test long topic name with last level being #
+        /// </summary>
+        [TestMethod]
+        public void Match_Hash_Test_LongTopic_MultiWildcard()
+        {
+            var sb = new StringBuilder();
+            const int numLevels = 8;
+            var levelNames = new string[numLevels];
+            for (var i = 0; i < numLevels; ++i)
+            {
+                if (i > 0)
+                {
+                    sb.Append("/");
+                }
+
+                string levelName;
+                if (i == numLevels - 1)
+                {
+                    // last one is #
+                    levelName = "#";
+                }
+                else
+                {
+                    levelName = "level" + i;
+                }
+
+                levelNames[i] = levelName;
+                sb.Append(levelName);
+            }
+
+            var topic = sb.ToString();
+
+            // UInt64 is limited to 8 levels
+
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
 
             Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
 
             var hashBytes = GetBytes(topicHash);
-            Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch");
-            Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch");
-            Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch");
-            Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch");
-            Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch");
-            Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch");
-            Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch");
-            Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch");
+            // all bytes should contain checksum
+            var count = 0;
+            foreach (var h in hashBytes)
+            {
+                if (count < 7)
+                {
+                    Assert.AreNotEqual(h, 0, "checksum mismatch");
+                }
+                else
+                {
+                    Assert.AreEqual(h, 0, "checksum mismatch");
+                }
 
-            // The mask should have zeroes where the wildcard and ff at the end
+                ++count;
+            }
+
+            // The mask should have ff except for last level
             var hashMaskBytes = GetBytes(topicHashMask);
-            Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch");
-            Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch");
-            Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch");
-            Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch");
-            Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch");
-            Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch");
-            Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch");
-            Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
+            count = 0;
+            foreach (var h in hashMaskBytes)
+            {
+                if (count < 7)
+                {
+                    Assert.AreEqual(h, 0xff, "mask mismatch");
+                }
+                else
+                {
+                    Assert.AreEqual(h, 0, "last mask mismatch");
+                }
+
+                ++count;
+            }
+        }
+
+        /// <summary>
+        ///     Test long topic name exceeding 8 levels
+        /// </summary>
+        [TestMethod]
+        public void Match_Hash_Test_LongTopic_NoWildCard()
+        {
+            var sb = new StringBuilder();
+            const int numLevels = 9;
+            var levelNames = new string[numLevels];
+            for (var i = 0; i < numLevels; ++i)
+            {
+                if (i > 0)
+                {
+                    sb.Append("/");
+                }
+
+                var levelName = "level" + i;
+                levelNames[i] = levelName;
+                sb.Append(levelName);
+            }
+
+            var topic = sb.ToString();
+
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
+
+            Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present");
+
+
+            var hashBytes = GetBytes(topicHash);
+            // all bytes should contain checksum
+            var count = 0;
+            foreach (var h in hashBytes)
+            {
+                Assert.AreNotEqual(h, 0, "checksum mismatch");
+                ++count;
+            }
+
+            // The mask should have ff 
+            var hashMaskBytes = GetBytes(topicHashMask);
+            count = 0;
+            foreach (var h in hashMaskBytes)
+            {
+                Assert.AreEqual(h, 0xff, "mask mismatch");
+                ++count;
+            }
+        }
+
+        /// <summary>
+        ///     Test long topic name with last level being +
+        /// </summary>
+        [TestMethod]
+        public void Match_Hash_Test_LongTopic_SingleWildCard()
+        {
+            var sb = new StringBuilder();
+            const int numLevels = 8;
+            var levelNames = new string[numLevels];
+            for (var i = 0; i < numLevels; ++i)
+            {
+                if (i > 0)
+                {
+                    sb.Append("/");
+                }
+
+                string levelName;
+                if (i == numLevels - 1)
+                {
+                    // last one is +
+                    levelName = "+";
+                }
+                else
+                {
+                    levelName = "level" + i;
+                }
+
+                levelNames[i] = levelName;
+                sb.Append(levelName);
+            }
+
+            var topic = sb.ToString();
+
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
+
+            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
+
+
+            var hashBytes = GetBytes(topicHash);
+            // all bytes should contain checksum
+            var count = 0;
+            foreach (var h in hashBytes)
+            {
+                if (count < 7)
+                {
+                    Assert.AreNotEqual(h, 0, "checksum mismatch");
+                }
+                else
+                {
+                    // wildcard position
+                    Assert.AreEqual(h, 0, "checksum mismatch");
+                }
+
+                ++count;
+            }
+
+            // The mask should have ff 
+            var hashMaskBytes = GetBytes(topicHashMask);
+            count = 0;
+            foreach (var h in hashMaskBytes)
+            {
+                if (count < 7)
+                {
+                    Assert.AreEqual(h, 0xff, "mask mismatch");
+                }
+                else
+                {
+                    Assert.AreEqual(h, 0, "last mask mismatch");
+                }
+
+                ++count;
+            }
         }
 
         [TestMethod]
@@ -65,13 +317,9 @@ namespace MQTTnet.Tests.Server
             var l1 = "topic1";
             var l2 = "sub1";
             var l3 = "#";
-            var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
+            var topic = $"{l0}/{l1}/{l2}/{l3}";
 
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
 
             Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
 
@@ -104,13 +352,9 @@ namespace MQTTnet.Tests.Server
             var l1 = "topic1";
             var l2 = "sub1";
             var l3 = "prop1";
-            var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
+            var topic = $"{l0}/{l1}/{l2}/{l3}";
 
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
 
             Assert.IsFalse(topicHasWildcard, "Wildcard detected when not wildcard present");
 
@@ -137,225 +381,13 @@ namespace MQTTnet.Tests.Server
             Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
         }
 
-        /// <summary>
-        /// Test long topic name with last level being #
-        /// </summary>
         [TestMethod]
-        public void Match_Hash_Test_LongTopic_MultiWildcard()
+        public async Task Match_Hash_Test_Search_MultiWildcard()
         {
-            var sb = new StringBuilder();
-            const int NumLevels = 8;
-            var levelNames = new string[NumLevels];
-            for (var i = 0; i < NumLevels; ++i)
-            {
-                if (i > 0)
-                    sb.Append("/");
-                string levelName;
-                if (i == NumLevels - 1)
-                {
-                    // last one is #
-                    levelName = "#";
-                }
-                else
-                {
-                   levelName = "level" + i;
-                }
-                levelNames[i] = levelName;
-                sb.Append(levelName);
-            }
-            var topic = sb.ToString();
-            
-            // UInt64 is limited to 8 levels
-
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
-
-            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
-
-            var hashBytes = GetBytes(topicHash);
-            // all bytes should contain checksum
-            int count = 0;
-            foreach (var h in hashBytes)
-            {
-                if (count < 7)
-                {
-                    Assert.AreNotEqual(h, 0, "checksum mismatch");
-                }
-                else
-                {
-                    Assert.AreEqual(h, 0, "checksum mismatch");
-                }
-                ++count;
-            }
-            // The mask should have ff except for last level
-            var hashMaskBytes = GetBytes(topicHashMask);
-            count = 0;
-            foreach (var h in hashMaskBytes)
-            {
-                if (count < 7)
-                {
-                    Assert.AreEqual(h, 0xff, "mask mismatch");
-                }
-                else
-                {
-                    Assert.AreEqual(h, 0, "last mask mismatch");
-                }
-                ++count;
-            }
-        }
-
-        /// <summary>
-        /// Test long topic name with last level being +
-        /// </summary>
-        [TestMethod]
-        public void Match_Hash_Test_LongTopic_SingleWildCard()
-        {
-            var sb = new StringBuilder();
-            const int NumLevels = 8;
-            var levelNames = new string[NumLevels];
-            for (var i = 0; i < NumLevels; ++i)
-            {
-                if (i > 0)
-                    sb.Append("/");
-                string levelName;
-                if (i == NumLevels - 1)
-                {
-                    // last one is +
-                    levelName = "+";
-                }
-                else
-                {
-                    levelName = "level" + i;
-                }
-                levelNames[i] = levelName;
-                sb.Append(levelName);
-            }
-            var topic = sb.ToString();
-
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
-
-            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
-
-
-            var hashBytes = GetBytes(topicHash);
-            // all bytes should contain checksum
-            int count = 0;
-            foreach (var h in hashBytes)
-            {
-                if (count < 7)
-                {
-                    Assert.AreNotEqual(h, 0, "checksum mismatch");
-                }
-                else
-                {
-                    // wildcard position
-                    Assert.AreEqual(h, 0, "checksum mismatch");
-                }
-                ++count;
-            }
-            // The mask should have ff 
-            var hashMaskBytes = GetBytes(topicHashMask);
-            count = 0;
-            foreach (var h in hashMaskBytes)
-            {
-                if (count < 7)
-                {
-                    Assert.AreEqual(h, 0xff, "mask mismatch");
-                }
-                else
-                {
-                    Assert.AreEqual(h, 0, "last mask mismatch");
-                }
-                ++count;
-            }
-        }
-
-        /// <summary>
-        /// Test long topic name exceeding 8 levels
-        /// </summary>
-        [TestMethod]
-        public void Match_Hash_Test_LongTopic_NoWildCard()
-        {
-            var sb = new StringBuilder();
-            const int NumLevels = 9;
-            var levelNames = new string[NumLevels];
-            for(var i=0; i < NumLevels; ++i)
-            {
-                if (i > 0)
-                    sb.Append("/");
-                var levelName = "level" + i;
-                levelNames[i] = levelName;
-                sb.Append(levelName);
-            }
-            var topic = sb.ToString();
-
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
-
-            Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present");
-
-
-            var hashBytes = GetBytes(topicHash);
-            // all bytes should contain checksum
-            int count = 0;
-            foreach(var h in hashBytes)
-            {
-                Assert.AreNotEqual(h, 0, "checksum mismatch");
-                ++count;
-            }
-            // The mask should have ff 
-            var hashMaskBytes = GetBytes(topicHashMask);
-            count = 0;
-            foreach (var h in hashMaskBytes)
-            {
-                Assert.AreEqual(h, 0xff, "mask mismatch");
-                ++count;
-            }
-        }
-
-
-        /// <summary>
-        /// Test long topic name with last level being +
-        /// </summary>
-        [TestMethod]
-        public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard()
-        {
-            var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+";
-
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
-
-            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
-        }
-
-        /// <summary>
-        /// Test long topic name with last level being #
-        /// </summary>
-        [TestMethod]
-        public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard()
-        {
-            var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#";
-
-            UInt64 topicHash;
-            UInt64 topicHashMask;
-            bool topicHasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
-
-            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
+            var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard);
+            var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard");
+            // Should match all topics
+            Assert.AreEqual(topics.Count, matchCount, "Topics not matched");
         }
 
         [TestMethod]
@@ -407,134 +439,125 @@ namespace MQTTnet.Tests.Server
         }
 
         [TestMethod]
-        public async Task Match_Hash_Test_Search_MultiWildcard()
+        public void Match_Hash_Test_SingleWildCard()
         {
-            var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard);
-            var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard");
-            // Should match all topics
-            Assert.AreEqual(topics.Count, matchCount, "Topics not matched");
-        }
+            var l0 = "pub0";
+            var l1 = "topic1";
+            var l2 = "+";
+            var l3 = "prop1";
+            var topic = $"{l0}/{l1}/{l2}/{l3}";
 
-        /// <summary>
-        /// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
-        /// </summary>
-        /// <returns></returns>
-        [TestMethod]
-        public void Check_Hash_Bucket_Depth()
-        {
-            Dictionary<string, List<string>> topicsByPublisher;
-            Dictionary<string, List<string>> singleWildcardTopicsByPublisher;
-            Dictionary<string, List<string>> multiWildcardTopicsByPublisher;
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
 
-            const int NumPublishers = 5000;
-            const int NumTopicsPerPublisher = 10;
+            Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
 
-            TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
+            var hashBytes = GetBytes(topicHash);
+            Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch");
+            Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch");
+            Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch");
+            Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch");
+            Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch");
+            Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch");
+            Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch");
+            Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch");
 
-            // There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ...
-            // Hash bucket depths should remain low.
-            var bucketDepths = new Dictionary<UInt64, int>();
-            int maxBucketDepth = 0;
-            UInt64 maxBucketDepthHash = 0;
-
-            var topicsByHash = new Dictionary<UInt64, List<string>>();
-
-            foreach (var t in topicsByPublisher)
-            {
-                var topics = t.Value;
-                foreach (var topic in topics)
-                {
-                    UInt64 topicHash;
-                    UInt64 hashMask;
-                    bool hasWildcard;
-                    MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard);
-
-                    bucketDepths.TryGetValue(topicHash, out var currentValue);
-                    ++currentValue;
-                    bucketDepths[topicHash] = currentValue;
-
-                    if (currentValue > maxBucketDepth)
-                    {
-                        maxBucketDepth = currentValue;
-                        maxBucketDepthHash = topicHash;
-                    }
-
-                    if (!topicsByHash.TryGetValue(topicHash, out var topicList))
-                    {
-                        topicList = new List<string>();
-                        topicsByHash.Add(topicHash, topicList);
-                    }
-                    topicList.Add(topic);
-                }
-            }
-
-            var maxDepthTopics = topicsByHash[maxBucketDepthHash];
-
-            Console.Write("Max bucket depth is " + maxBucketDepth);
-
-            // for the test case the bucket depth should be less than 100
-            Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth");
-        }
-
-
-        [TestMethod]
-        public void Check_Selected_Topic_Hashes()
-        {
-            CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF);
-            CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF);
-            CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF);
-            CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000);
-            CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000);
+            // The mask should have zeroes where the wildcard and ff at the end
+            var hashMaskBytes = GetBytes(topicHashMask);
+            Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch");
+            Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch");
+            Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch");
+            Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch");
+            Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch");
+            Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch");
+            Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch");
+            Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
         }
 
         void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask)
         {
-            ulong topicHash;
-            ulong hashMask;
-            bool hasWildcard;
-
-            MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard);
+            MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
 
             Console.WriteLine();
             Console.WriteLine("Topic: " + topic);
-            Console.WriteLine(string.Format("Hash: {0:X8}", topicHash));
-            Console.WriteLine(string.Format("Hash Mask: {0:X8}", hashMask));
+            Console.WriteLine($"Hash: {topicHash:X8}");
+            Console.WriteLine($"Hash Mask: {hashMask:X8}");
 
             Assert.AreEqual(expectedHash, topicHash, "Topic hash not as expected. Has the hash function changed?");
             Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected");
         }
 
+        int CheckTopicSubscriptions(MqttSession clientSession, List<string> topicsToFind, string subject)
+        {
+            var matchCount = 0;
+
+            {
+                var resultCount = 0;
+
+                var stopWatch = new Stopwatch();
+                stopWatch.Start();
+                var countUp = 0;
+                var countDown = topicsToFind.Count - 1;
+                for (; countUp < topicsToFind.Count; ++countUp, --countDown)
+                {
+                    var topicToFind = topicsToFind[countDown];
+
+                    MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out _, out _);
+
+                    clientSession.TryCheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient", out var result);
+                    if (result.IsSubscribed)
+                    {
+                        ++matchCount;
+                    }
+
+                    ++resultCount;
+                }
+
+                stopWatch.Stop();
+
+                Console.Write("Match count: " + matchCount + "; ");
+
+                Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds);
+            }
+
+            return matchCount;
+        }
+
+        byte[] GetBytes(ulong value)
+        {
+            var bytes = BitConverter.GetBytes(value);
+            // Ensure that highest byte comes first for comparison left to right
+            if (BitConverter.IsLittleEndian)
+            {
+                return bytes.Reverse().ToArray();
+            }
+
+            return bytes;
+        }
+
 
         async Task<List<string>> PrepareTopicHashSubscriptions(TopicHashSelector selector)
         {
-            Dictionary<string, List<string>> topicsByPublisher;
-            Dictionary<string, List<string>> singleWildcardTopicsByPublisher;
-            Dictionary<string, List<string>> multiWildcardTopicsByPublisher;
+            const int numPublishers = 1;
+            const int numTopicsPerPublisher = 10000;
 
-            const int NumPublishers = 1;
-            const int NumTopicsPerPublisher = 10000;
-
-            TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
+            TopicGenerator.Generate(
+                numPublishers,
+                numTopicsPerPublisher,
+                out var topicsByPublisher,
+                out var singleWildcardTopicsByPublisher,
+                out var multiWildcardTopicsByPublisher);
 
             var topics = topicsByPublisher.FirstOrDefault().Value;
             var singleWildcardTopics = singleWildcardTopicsByPublisher.FirstOrDefault().Value;
             var multiWildcardTopics = multiWildcardTopicsByPublisher.FirstOrDefault().Value;
 
-            const string ClientId = "Client1";
-            var logger = new Mockups.TestLogger();
-            var serverOptions = new MQTTnet.Server.MqttServerOptions();
-            var eventContainer = new MQTTnet.Server.MqttServerEventContainer();
+            const string clientId = "Client1";
+            var logger = new TestLogger();
+            var serverOptions = new MqttServerOptions();
+            var eventContainer = new MqttServerEventContainer();
             var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger);
             var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger);
-            _clientSession = new MQTTnet.Server.MqttSession(
-                        ClientId,
-                        false,
-                        new Dictionary<object, object>(),
-                        serverOptions,
-                        eventContainer,
-                        retainedMessagesManager,
-                        sessionManager
-                        );
+            _clientSession = new MqttSession(clientId, false, new Dictionary<object, object>(), serverOptions, eventContainer, retainedMessagesManager, sessionManager);
 
             List<string> topicsToSubscribe;
 
@@ -550,63 +573,26 @@ namespace MQTTnet.Tests.Server
                     topicsToSubscribe = topics;
                     break;
             }
+
             foreach (var t in topicsToSubscribe)
             {
-                var subPacket = new Packets.MqttSubscribePacket();
-                var filter = new Packets.MqttTopicFilter();
-                filter.Topic = t;
+                var subPacket = new MqttSubscribePacket();
+                var filter = new MqttTopicFilter
+                {
+                    Topic = t
+                };
                 subPacket.TopicFilters.Add(filter);
-                await _clientSession.SubscriptionsManager.Subscribe(subPacket, default(CancellationToken));
+                await _clientSession.Subscribe(subPacket, default);
             }
 
             return topics;
         }
 
-        int CheckTopicSubscriptions(MQTTnet.Server.MqttSession clientSession, List<string> topicsToFind, string subject)
+        enum TopicHashSelector
         {
-            var matchCount = 0;
-
-            {
-                int resultCount = 0;
-
-                var stopWatch = new System.Diagnostics.Stopwatch();
-                stopWatch.Start();
-                var countUp = 0;
-                var countDown = topicsToFind.Count - 1;
-                for (; countUp < topicsToFind.Count; ++countUp, --countDown)
-                {
-                    var topicToFind = topicsToFind[countDown];
-
-                    UInt64 topicHash;
-                    UInt64 hashMask;
-                    bool hasWildcard;
-                    MQTTnet.Server.MqttSubscription.CalculateTopicHash((string)topicToFind, out topicHash, out hashMask, out hasWildcard);
-
-                    var result = clientSession.SubscriptionsManager.CheckSubscriptions((string)topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient");
-                    if (result.IsSubscribed)
-                    {
-                        ++matchCount;
-                    }
-                    ++resultCount;
-                }
-
-                stopWatch.Stop();
-
-                Console.Write("Match count: " + matchCount + "; ");
-
-                Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds);
-            }
-
-            return matchCount;
-        }
-
-        byte[] GetBytes(UInt64 value)
-        {
-            var bytes = BitConverter.GetBytes(value);
-            // Ensure that highest byte comes first for comparison left to right
-            if (BitConverter.IsLittleEndian)
-                return bytes.Reverse().ToArray();
-            return bytes;
+            NoWildcard,
+            SingleWildcard,
+            MultiWildcard
         }
     }
-}
+}

Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs

@@ -0,0 +1,32 @@
+// 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.Protocol;
+
+namespace MQTTnet.Tests.Server
+{
+    [TestClass]
+    public sealed class Unsubscribe_Tests : BaseTestClass
+    {
+        [TestMethod]
+        [ExpectedException(typeof(MqttClientDisconnectedException))]
+        public async Task Disconnect_While_Unsubscribing()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                // The client will be disconnect directly after subscribing!
+                server.ClientUnsubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
+
+                var client = await testEnvironment.ConnectClient();
+                await client.SubscribeAsync("#");
+                await client.UnsubscribeAsync("#");
+            }
+        }
+    }
+}

Source/MQTTnet.Tests/TopicGenerator.cs

@@ -1,8 +1,7 @@
 using System;
 using System.Collections.Generic;
-using System.Text;
 
-namespace MQTTnet.Tests.Server
+namespace MQTTnet.Tests
 {
     public class TopicGenerator
     {

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -13,7 +13,6 @@ using MQTTnet.Channel;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
-using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 
@@ -296,8 +295,26 @@ namespace MQTTnet.Adapter
 
             while (totalBytesRead < buffer.Length)
             {
-                var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
+                // Check two times for cancellation because the call to _ReadAsync_ might block for some time.
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return ReadFixedHeaderResult.Cancelled;
+                }
 
+                int bytesRead;
+                try
+                {
+                    bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    return ReadFixedHeaderResult.Cancelled;
+                }
+                catch (SocketException)
+                {
+                    return ReadFixedHeaderResult.ConnectionClosed;
+                }
+               
                 if (cancellationToken.IsCancellationRequested)
                 {
                     return ReadFixedHeaderResult.Cancelled;

Source/MQTTnet/Client/Exceptions/MqttClientDisconnectedException.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.
+
+using System;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientDisconnectedException : MqttCommunicationException
+    {
+        public MqttClientDisconnectedException(Exception innerException) : base("The MQTT client is disconnected.", innerException)
+        {
+        }
+    }
+}

Source/MQTTnet/Client/Exceptions/MqttClientUnexpectedDisconnectReceivedException.cs

@@ -3,14 +3,15 @@
 // See the LICENSE file in the project root for more information.
 
 using System.Collections.Generic;
+using MQTTnet.Exceptions;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
-namespace MQTTnet.Exceptions
+namespace MQTTnet.Client
 {
-    public class MqttUnexpectedDisconnectReceivedException : MqttCommunicationException
+    public sealed class MqttClientUnexpectedDisconnectReceivedException : MqttCommunicationException
     {
-        public MqttUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) 
+        public MqttClientUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) 
             : base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.")
         {
             ReasonCode = disconnectPacket.ReasonCode;

Source/MQTTnet/Client/MqttClient.cs

@@ -32,25 +32,21 @@ namespace MQTTnet.Client
         readonly object _disconnectLock = new object();
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
         readonly MqttNetSourceLogger _logger;
-        readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
 
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         readonly IMqttNetLogger _rootLogger;
 
         IMqttChannelAdapter _adapter;
-
         CancellationTokenSource _backgroundCancellationTokenSource;
         bool _cleanDisconnectInitiated;
         volatile int _connectionStatus;
-
         MqttClientDisconnectReason _disconnectReason;
         string _disconnectReasonString;
         Task _keepAlivePacketsSenderTask;
-
         DateTime _lastPacketSentTimestamp;
+        MqttPacketDispatcher _packetDispatcher;
         Task _packetReceiverTask;
-
         AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
         Task _publishPacketReceiverTask;
 
@@ -128,7 +124,7 @@ namespace MQTTnet.Client
                 }
 
                 _packetIdentifierProvider.Reset();
-                _packetDispatcher.CancelAll();
+                _packetDispatcher = new MqttPacketDispatcher();
 
                 _backgroundCancellationTokenSource = new CancellationTokenSource();
                 var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
@@ -414,6 +410,10 @@ namespace MQTTnet.Client
                 _publishPacketReceiverQueue = null;
 
                 _adapter?.Dispose();
+                _adapter = null;
+                
+                _packetDispatcher?.Dispose();
+                _packetDispatcher = null;
             }
         }
 
@@ -447,6 +447,8 @@ namespace MQTTnet.Client
 
             try
             {
+                _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception));
+                
                 var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender);
                 var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender);
                 var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
@@ -558,7 +560,7 @@ namespace MQTTnet.Client
             _disconnectReasonString = disconnectPacket.ReasonString;
 
             // Also dispatch disconnect to waiting threads to generate a proper exception.
-            _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
+            _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket));
 
             return DisconnectInternalAsync(_packetReceiverTask, null, null);
         }
@@ -776,7 +778,7 @@ namespace MQTTnet.Client
                 }
                 else
                 {
-                    _logger.Error(exception, "Error while receiving packets.");
+                    _logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name}).");
                 }
 
                 _packetDispatcher.FailAll(exception);
@@ -827,6 +829,11 @@ namespace MQTTnet.Client
                     return;
                 }
 
+                if (exception is AggregateException aggregateException)
+                {
+                    exception = aggregateException.GetBaseException();
+                }
+                
                 if (exception is OperationCanceledException)
                 {
                 }

Source/MQTTnet/Exceptions/MqttCommunicationException.cs

@@ -8,12 +8,8 @@ namespace MQTTnet.Exceptions
 {
     public class MqttCommunicationException : Exception
     {
-        protected MqttCommunicationException()
-        {
-        }
-
         public MqttCommunicationException(Exception innerException)
-            : base(innerException.Message, innerException)
+            : base(innerException?.Message ?? "MQTT communication failed.", innerException)
         {
         }
 

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/Formatter/MqttUnsubAckPacketFactory.cs

@@ -10,16 +10,16 @@ namespace MQTTnet.Formatter
 {
     public sealed class MqttUnsubAckPacketFactory
     {
-        public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubscribeResult mqttUnsubscribeResult)
+        public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult)
         {
             if (unsubscribePacket == null)
             {
                 throw new ArgumentNullException(nameof(unsubscribePacket));
             }
 
-            if (mqttUnsubscribeResult == null)
+            if (unsubscribeResult == null)
             {
-                throw new ArgumentNullException(nameof(mqttUnsubscribeResult));
+                throw new ArgumentNullException(nameof(unsubscribeResult));
             }
 
             var unsubAckPacket = new MqttUnsubAckPacket
@@ -28,7 +28,7 @@ namespace MQTTnet.Formatter
             };
 
             // MQTTv5.0.0 only.
-            unsubAckPacket.ReasonCodes = mqttUnsubscribeResult.ReasonCodes;
+            unsubAckPacket.ReasonCodes = unsubscribeResult.ReasonCodes;
 
             return unsubAckPacket;
         }

Source/MQTTnet/Internal/AsyncEvent.cs

@@ -123,7 +123,7 @@ namespace MQTTnet.Internal
             }
             catch (Exception exception)
             {
-                logger.Warning(exception, $"Error while invoking event ({typeof(TEventArgs)}).");
+                logger.Warning(exception, $"Error while invoking event with arguments of type {typeof(TEventArgs)}.");
             }
         }
     }

Source/MQTTnet/Internal/AsyncQueue.cs

@@ -11,64 +11,39 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncQueue<TItem> : IDisposable
     {
+        readonly AsyncSignal _signal = new AsyncSignal();
         readonly object _syncRoot = new object();
-        SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+
         ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
 
+        bool _isDisposed;
+        
         public int Count => _queue.Count;
 
+        public void Clear()
+        {
+            Interlocked.Exchange(ref _queue, new ConcurrentQueue<TItem>());
+        }
+
+        public void Dispose()
+        {
+            lock (_syncRoot)
+            {
+                _signal?.Dispose();
+
+                _isDisposed = true;
+            }
+        }
+
         public void Enqueue(TItem item)
         {
             lock (_syncRoot)
             {
                 _queue.Enqueue(item);
-                _semaphore?.Release();
+                _signal.Set();
             }
         }
 
-        public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
-        {
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                try
-                {
-                    Task task;
-                    lock (_syncRoot)
-                    {
-                        if (_semaphore == null)
-                        {
-                            return new AsyncQueueDequeueResult<TItem>(false, default);
-                        }
-
-                        task = _semaphore.WaitAsync(cancellationToken);
-                    }
-                    
-                    await task.ConfigureAwait(false);
-
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return new AsyncQueueDequeueResult<TItem>(false, default);
-                    }
-
-                    if (_queue.TryDequeue(out var item))
-                    {
-                        return new AsyncQueueDequeueResult<TItem>(true, item);
-                    }
-                }
-                catch (ArgumentNullException)
-                {
-                    // The semaphore throws this internally sometimes.
-                    return new AsyncQueueDequeueResult<TItem>(false, default);
-                }
-                catch (OperationCanceledException)
-                {
-                    return new AsyncQueueDequeueResult<TItem>(false, default);
-                }
-            }
-
-            return new AsyncQueueDequeueResult<TItem>(false, default);
-        }
-
         public AsyncQueueDequeueResult<TItem> TryDequeue()
         {
             if (_queue.TryDequeue(out var item))
@@ -79,18 +54,42 @@ namespace MQTTnet.Internal
             return new AsyncQueueDequeueResult<TItem>(false, default);
         }
 
-        public void Clear()
+        public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
         {
-            Interlocked.Exchange(ref _queue, new ConcurrentQueue<TItem>());
-        }
-
-        public void Dispose()
-        {
-            lock (_syncRoot)
+            while (!cancellationToken.IsCancellationRequested)
             {
-                _semaphore?.Dispose();
-                _semaphore = null;
+                try
+                {
+                    Task task;
+                    lock (_syncRoot)
+                    {
+                        if (_isDisposed)
+                        {
+                            return new AsyncQueueDequeueResult<TItem>(false, default);
+                        }
+
+                        task = _signal.WaitAsync(cancellationToken);
+                    }
+
+                    await task.ConfigureAwait(false);
+
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return new AsyncQueueDequeueResult<TItem>(false, default);
+                    }
+
+                    if (_queue.TryDequeue(out var item))
+                    {
+                        return new AsyncQueueDequeueResult<TItem>(true, item);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    return new AsyncQueueDequeueResult<TItem>(false, default);
+                }
             }
+
+            return new AsyncQueueDequeueResult<TItem>(false, default);
         }
     }
-}
+}

Source/MQTTnet/Internal/AsyncSignal.cs

@@ -0,0 +1,113 @@
+// 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;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class AsyncSignal : IDisposable
+    {
+        readonly object _syncRoot = new object();
+
+        bool _isDisposed;
+        bool _isSignaled;
+        AsyncSignalWaiter _waiter;
+
+        public void Dispose()
+        {
+            lock (_syncRoot)
+            {
+                _waiter?.Dispose();
+                _waiter = null;
+
+                _isDisposed = true;
+            }
+        }
+
+        public void Set()
+        {
+            lock (_syncRoot)
+            {
+                _isSignaled = true;
+
+                // If there is already a waiting task let it run.
+                _waiter?.Approve();
+                _waiter = null;
+            }
+        }
+
+        public Task WaitAsync(CancellationToken cancellationToken = default)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            lock (_syncRoot)
+            {
+                ThrowIfDisposed();
+
+                if (_isSignaled)
+                {
+                    return CompletedTask.Instance;
+                }
+
+                if (_waiter != null)
+                {
+                    if (!_waiter.Task.IsCompleted)
+                    {
+                        throw new InvalidOperationException("Only one waiting task is permitted per async signal.");
+                    }
+                    
+                    _waiter.Dispose();
+                }
+
+                _waiter = new AsyncSignalWaiter(cancellationToken);
+                return _waiter.Task;
+            }
+        }
+
+        void ThrowIfDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw new ObjectDisposedException(nameof(AsyncSignal));
+            }
+        }
+
+        sealed class AsyncSignalWaiter : IDisposable
+        {
+            readonly AsyncTaskCompletionSource<bool> _promise = new AsyncTaskCompletionSource<bool>();
+
+            // ReSharper disable once FieldCanBeMadeReadOnly.Local
+            CancellationTokenRegistration _cancellationTokenRegistration;
+
+            public AsyncSignalWaiter(CancellationToken cancellationToken)
+            {
+                if (cancellationToken.CanBeCanceled)
+                {
+                    _cancellationTokenRegistration = cancellationToken.Register(Cancel);
+                }
+            }
+
+            public Task Task => _promise.Task;
+
+            public void Approve()
+            {
+                _promise.TrySetResult(true);
+            }
+
+            public void Dispose()
+            {
+                _cancellationTokenRegistration.Dispose();
+
+                _promise.TrySetException(new ObjectDisposedException(nameof(AsyncSignalWaiter)));
+            }
+
+            void Cancel()
+            {
+                _promise.TrySetCanceled();
+            }
+        }
+    }
+}

Source/MQTTnet/Internal/MqttPacketBus.cs

@@ -20,7 +20,7 @@ namespace MQTTnet.Internal
             new LinkedList<MqttPacketBusItem>()
         };
 
-        readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1);
+        readonly AsyncSignal _signal = new AsyncSignal();
         readonly object _syncRoot = new object();
 
         int _activePartition = (int)MqttPacketBusPartition.Health;
@@ -82,7 +82,16 @@ namespace MQTTnet.Internal
 
                 // No partition contains data so that we have to wait and put
                 // the worker back to the thread pool.
-                await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+                try
+                {
+                    await _signal.WaitAsync(cancellationToken).ConfigureAwait(false);
+                }
+                catch (ObjectDisposedException)
+                {
+                    // The cancelled token should "hide" the disposal of the signal.
+                    cancellationToken.ThrowIfCancellationRequested();
+                    throw;
+                }
             }
 
             cancellationToken.ThrowIfCancellationRequested();
@@ -92,7 +101,7 @@ namespace MQTTnet.Internal
 
         public void Dispose()
         {
-            _semaphore?.Dispose();
+            _signal.Dispose();
         }
 
         public void DropFirstItem(MqttPacketBusPartition partition)
@@ -118,11 +127,7 @@ namespace MQTTnet.Internal
             lock (_syncRoot)
             {
                 _partitions[(int)partition].AddLast(item);
-
-                if (_semaphore.CurrentCount == 0)
-                {
-                    _semaphore.Release();
-                }
+                _signal.Set();
             }
         }
 

Source/MQTTnet/MQTTnet.csproj.DotSettings

@@ -2,6 +2,7 @@
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cconnecting/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdiagnostics/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdisconnecting/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cexceptions/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cextendedauthenticationexchange/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Coptions/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cpublishing/@EntryIndexedValue">True</s:Boolean>

Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs

@@ -2,48 +2,103 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using MQTTnet.Packets;
 using System;
 using System.Collections.Generic;
+using MQTTnet.Packets;
 
 namespace MQTTnet.PacketDispatcher
 {
-    public sealed class MqttPacketDispatcher
+    public sealed class MqttPacketDispatcher : IDisposable
     {
-        readonly List<IMqttPacketAwaitable> _awaitables = new List<IMqttPacketAwaitable>();
+        readonly List<IMqttPacketAwaitable> _waiters = new List<IMqttPacketAwaitable>();
 
-        public void FailAll(Exception exception)
+        bool _isDisposed;
+
+        public MqttPacketAwaitable<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
         {
-            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            var awaitable = new MqttPacketAwaitable<TResponsePacket>(packetIdentifier, this);
 
-            lock (_awaitables)
+            lock (_waiters)
             {
-                foreach (var awaitable in _awaitables)
-                {
-                    awaitable.Fail(exception);
-                }
-
-                _awaitables.Clear();
+                _waiters.Add(awaitable);
             }
+
+            return awaitable;
         }
 
         public void CancelAll()
         {
-            lock (_awaitables)
+            lock (_waiters)
             {
-                foreach (var awaitable in _awaitables)
+                foreach (var awaitable in _waiters)
                 {
                     awaitable.Cancel();
                 }
 
-                _awaitables.Clear();
+                _waiters.Clear();
             }
         }
-        
+
+        public void Dispose()
+        {
+            Dispose(new ObjectDisposedException(nameof(MqttPacketDispatcher)));
+        }
+
+        public void Dispose(Exception exception)
+        {
+            if (exception == null)
+            {
+                throw new ArgumentNullException(nameof(exception));
+            }
+
+            lock (_waiters)
+            {
+                FailAll(exception);
+                
+                // Make sure that no task can start waiting after this instance is already disposed.
+                // This will prevent unexpected freezes.
+                _isDisposed = true;
+            }
+        }
+
+        public void FailAll(Exception exception)
+        {
+            if (exception == null)
+            {
+                throw new ArgumentNullException(nameof(exception));
+            }
+
+            lock (_waiters)
+            {
+                foreach (var awaitable in _waiters)
+                {
+                    awaitable.Fail(exception);
+                }
+
+                _waiters.Clear();
+            }
+        }
+
+        public void RemoveAwaitable(IMqttPacketAwaitable awaitable)
+        {
+            if (awaitable == null)
+            {
+                throw new ArgumentNullException(nameof(awaitable));
+            }
+
+            lock (_waiters)
+            {
+                _waiters.Remove(awaitable);
+            }
+        }
+
         public bool TryDispatch(MqttPacket packet)
         {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
-            
+            if (packet == null)
+            {
+                throw new ArgumentNullException(nameof(packet));
+            }
+
             ushort identifier = 0;
             if (packet is MqttPacketWithIdentifier packetWithIdentifier)
             {
@@ -51,13 +106,15 @@ namespace MQTTnet.PacketDispatcher
             }
 
             var packetType = packet.GetType();
-            var awaitables = new List<IMqttPacketAwaitable>();
-            
-            lock (_awaitables)
+            var waiters = new List<IMqttPacketAwaitable>();
+
+            lock (_waiters)
             {
-                for (var i = _awaitables.Count - 1; i >= 0; i--)
+                ThrowIfDisposed();
+                
+                for (var i = _waiters.Count - 1; i >= 0; i--)
                 {
-                    var entry = _awaitables[i];
+                    var entry = _waiters[i];
 
                     // Note: The PingRespPacket will also arrive here and has NO identifier but there
                     // is code which waits for it. So the code must be able to deal with filters which
@@ -66,39 +123,25 @@ namespace MQTTnet.PacketDispatcher
                     {
                         continue;
                     }
-                    
-                    awaitables.Add(entry);
-                    _awaitables.RemoveAt(i);
+
+                    waiters.Add(entry);
+                    _waiters.RemoveAt(i);
                 }
             }
-            
-            foreach (var matchingEntry in awaitables)
+
+            foreach (var matchingEntry in waiters)
             {
                 matchingEntry.Complete(packet);
             }
 
-            return awaitables.Count > 0;
-        }
-        
-        public MqttPacketAwaitable<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
-        {
-            var awaitable = new MqttPacketAwaitable<TResponsePacket>(packetIdentifier, this);
-
-            lock (_awaitables)
-            {
-                _awaitables.Add(awaitable);
-            }
-            
-            return awaitable;
+            return waiters.Count > 0;
         }
 
-        public void RemoveAwaitable(IMqttPacketAwaitable awaitable)
+        void ThrowIfDisposed()
         {
-            if (awaitable == null) throw new ArgumentNullException(nameof(awaitable));
-            
-            lock (_awaitables)
+            if (_isDisposed)
             {
-                _awaitables.Remove(awaitable);
+                throw new ObjectDisposedException(nameof(MqttPacketDispatcher));
             }
         }
     }

Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs

@@ -10,15 +10,23 @@ namespace MQTTnet.Server
 {
     public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
     {
+        public ClientAcknowledgedPublishPacketEventArgs(string clientId, IDictionary sessionItems, MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+            PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
+            AcknowledgePacket = acknowledgePacket ?? throw new ArgumentNullException(nameof(acknowledgePacket));
+        }
+        
         /// <summary>
         ///     Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet.
         /// </summary>
-        public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; }
+        public MqttPacketWithIdentifier AcknowledgePacket { get; }
 
         /// <summary>
         ///     Gets the ID of the client which acknowledged a PUBLISH packet.
         /// </summary>
-        public string ClientId { get; internal set; }
+        public string ClientId { get; }
 
         /// <summary>
         ///     Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2.
@@ -28,11 +36,11 @@ namespace MQTTnet.Server
         /// <summary>
         ///     Gets the PUBLISH packet which was acknowledged.
         /// </summary>
-        public MqttPublishPacket PublishPacket { get; internal set; }
+        public MqttPublishPacket PublishPacket { get; }
 
         /// <summary>
         ///     Gets the session items which contain custom user data per session.
         /// </summary>
-        public IDictionary SessionItems { get; internal set; }
+        public IDictionary SessionItems { get; }
     }
 }

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -11,7 +11,6 @@ using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
-using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
@@ -34,6 +33,7 @@ namespace MQTTnet.Server
         readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
 
         CancellationTokenSource _cancellationToken;
+        bool _disconnectPacketSent;
 
         public MqttClient(
             MqttConnectPacket connectPacket,
@@ -147,19 +147,22 @@ namespace MQTTnet.Server
         {
             IsRunning = false;
 
-            // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
-            if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
+            if (!_disconnectPacketSent)
             {
-                // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
-                // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
-                // a configuration option is requested in the future.
-                if (reason != MqttDisconnectReasonCode.NormalDisconnection)
+                // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
+                if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
                 {
-                    // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
-                    // token because the entire connection is closed (disposed) as soon as the cancellation
-                    // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
-                    // at the client!
-                    await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+                    // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
+                    // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
+                    // a configuration option is requested in the future.
+                    if (reason != MqttDisconnectReasonCode.NormalDisconnection)
+                    {
+                        // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
+                        // token because the entire connection is closed (disposed) as soon as the cancellation
+                        // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+                        // at the client!
+                        await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+                    }
                 }
             }
 
@@ -170,14 +173,7 @@ namespace MQTTnet.Server
         {
             if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
             {
-                var eventArgs = new ClientAcknowledgedPublishPacketEventArgs
-                {
-                    PublishPacket = publishPacket,
-                    AcknowledgePacket = acknowledgePacket,
-                    ClientId = Id,
-                    SessionItems = Session.Items
-                };
-
+                var eventArgs = new ClientAcknowledgedPublishPacketEventArgs(Id, Session.Items, publishPacket, acknowledgePacket);
                 return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
             }
 
@@ -272,17 +268,14 @@ namespace MQTTnet.Server
             }
         }
 
-        async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
+        Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
         {
-            var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier);
-
-            if (acknowledgedPublishPacket != null)
-            {
-                await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false);
-            }
-
+            // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
+            // properly when the client has sent the PUBCOMP packet.
             var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
             Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
+
+            return CompletedTask.Instance;
         }
 
         void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
@@ -293,7 +286,7 @@ namespace MQTTnet.Server
 
         async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
-            var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
+            var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
 
             var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
 
@@ -307,18 +300,17 @@ 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);
+            var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
 
             var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
 
@@ -496,7 +488,7 @@ namespace MQTTnet.Server
 
             try
             {
-                while (!cancellationToken.IsCancellationRequested)
+                while (!cancellationToken.IsCancellationRequested && !IsTakenOver && IsRunning)
                 {
                     packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
 
@@ -578,6 +570,9 @@ namespace MQTTnet.Server
         {
             try
             {
+                // This also indicates that it was tried at least!
+                _disconnectPacketSent = true;
+                
                 var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode);
 
                 using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -150,11 +150,17 @@ namespace MQTTnet.Server
 
                 foreach (var session in subscriberSessions)
                 {
-                    var checkSubscriptionsResult = session.SubscriptionsManager.CheckSubscriptions(
+                    if(!session.TryCheckSubscriptions(
                         applicationMessage.Topic,
                         topicHash,
                         applicationMessage.QualityOfServiceLevel,
-                        senderId);
+                        senderId,
+                        out var checkSubscriptionsResult))
+                    {
+                        // Checking the subscriptions has failed for the session. The session
+                        // will be ignored.
+                        continue;
+                    }
 
                     if (!checkSubscriptionsResult.IsSubscribed)
                     {
@@ -301,7 +307,7 @@ namespace MQTTnet.Server
                         channelAdapter.Endpoint,
                         client.Session.Items);
 
-                    await _eventContainer.ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                    await _eventContainer.ClientConnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
                 }
 
                 await client.RunAsync().ConfigureAwait(false);
@@ -414,13 +420,13 @@ namespace MQTTnet.Server
 
             var clientSession = GetClientSession(clientId);
 
-            var subscribeResult = await clientSession.SubscriptionsManager.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
+            var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
 
             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));
                 }
             }
@@ -441,7 +447,7 @@ namespace MQTTnet.Server
             var fakeUnsubscribePacket = new MqttUnsubscribePacket();
             fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters);
 
-            return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
+            return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
         }
 
         MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
@@ -484,10 +490,15 @@ namespace MQTTnet.Server
 
             using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false))
             {
-                MqttSession session;
+                MqttSession oldSession;
+                MqttClient oldClient;
+                
                 lock (_sessionsManagementLock)
                 {
-                    if (!_sessions.TryGetValue(connectPacket.ClientId, out session))
+                    MqttSession session;
+                    
+                    // Create a new session (if required).
+                    if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession))
                     {
                         session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
                     }
@@ -496,11 +507,15 @@ namespace MQTTnet.Server
                         if (connectPacket.CleanSession)
                         {
                             _logger.Verbose("Deleting existing session of client '{0}' due to clean start.", connectPacket.ClientId);
+                            _subscriberSessions.Remove(oldSession);
                             session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
                         }
                         else
                         {
                             _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
+                            session = oldSession;
+                            oldSession = null;
+                            
                             // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
                             session.IsPersistent = sessionShouldPersist;
                             connAckPacket.IsSessionPresent = true;
@@ -509,41 +524,45 @@ namespace MQTTnet.Server
                     }
 
                     _sessions[connectPacket.ClientId] = session;
+                    
+                    // Create a new client (always required).
+                    _clients.TryGetValue(connectPacket.ClientId, out oldClient);
+                    if (oldClient != null)
+                    {
+                        // This will stop the current client from sending and receiving but remains the connection active
+                        // for a later DISCONNECT packet.
+                        oldClient.IsTakenOver = true;
+                    }
+                    
+                    client = CreateClient(connectPacket, channelAdapter, session);
+
+                    _clients[connectPacket.ClientId] = client;
                 }
 
                 if (!connAckPacket.IsSessionPresent)
                 {
                     // TODO: This event is not yet final. It can already be used but restoring sessions from storage will be added later!
                     var preparingSessionEventArgs = new PreparingSessionEventArgs();
-                    await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false);
+                    await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false);
                 }
-
-                MqttClient existingClient;
-
-                lock (_clients)
+                
+                if (oldClient != null)
                 {
-                    _clients.TryGetValue(connectPacket.ClientId, out existingClient);
-                    client = CreateClient(connectPacket, channelAdapter, session);
+                    await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
 
-                    _clients[connectPacket.ClientId] = client;
-                }
-
-                if (existingClient != null)
-                {
-                    existingClient.IsTakenOver = true;
-                    await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
-
-                    if (_eventContainer.ClientConnectedEvent.HasHandlers)
+                    if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
                     {
                         var eventArgs = new ClientDisconnectedEventArgs(
-                            existingClient.Id,
+                            oldClient.Id,
                             MqttClientDisconnectType.Takeover,
-                            existingClient.Endpoint,
-                            existingClient.Session.Items);
+                            oldClient.Endpoint,
+                            oldClient.Session.Items);
                         
-                        await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                        await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
                     }
                 }
+
+                oldSession?.Dispose();
             }
 
             return client;

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -7,6 +7,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
@@ -30,7 +31,7 @@ namespace MQTTnet.Server
         readonly Dictionary<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>();
 
         // Use subscription lock to maintain consistency across subscriptions and topic hash dictionaries
-        readonly SemaphoreSlim _subscriptionsLock = new SemaphoreSlim(1);
+        readonly AsyncLock _subscriptionsLock = new AsyncLock();
         readonly Dictionary<ulong, TopicHashMaskSubscriptions> _wildcardSubscriptionsByTopicHash = new Dictionary<ulong, TopicHashMaskSubscriptions>();
 
         public MqttClientSubscriptionsManager(
@@ -45,13 +46,12 @@ namespace MQTTnet.Server
             _subscriptionChangedNotification = subscriptionChangedNotification;
         }
 
-        public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
+        public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId)
         {
             var possibleSubscriptions = new List<MqttSubscription>();
 
             // Check for possible subscriptions. They might have collisions but this is fine.
-            _subscriptionsLock.Wait();
-            try
+            using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
             {
                 if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions))
                 {
@@ -63,18 +63,14 @@ namespace MQTTnet.Server
                     var wildcardSubscriptions = wcs.Value;
                     var subscriptionHash = wcs.Key;
                     var subscriptionHashMask = wildcardSubscriptions.HashMask;
-                    
+
                     if ((topicHash & subscriptionHashMask) == subscriptionHash)
                     {
                         possibleSubscriptions.AddRange(wildcardSubscriptions.Subscriptions.ToList());
                     }
                 }
             }
-            finally
-            {
-                _subscriptionsLock.Release();
-            }
-            
+
             // The pre check has evaluated that nothing is subscribed.
             // If there were some possible candidates they get checked below
             // again to avoid collisions.
@@ -83,7 +79,7 @@ namespace MQTTnet.Server
                 return CheckSubscriptionsResult.NotSubscribed;
             }
 
-            var senderIsReceiver = string.Equals(senderClientId, _session.Id);
+            var senderIsReceiver = string.Equals(senderId, _session.Id);
             var maxQoSLevel = -1; // Not subscribed.
 
             HashSet<uint> subscriptionIdentifiers = null;
@@ -136,7 +132,7 @@ namespace MQTTnet.Server
                 SubscriptionIdentifiers = subscriptionIdentifiers?.ToList() ?? EmptySubscriptionIdentifiers,
 
                 // Start with the same QoS as the publisher.
-                QualityOfServiceLevel = applicationMessageQoSLevel
+                QualityOfServiceLevel = qualityOfServiceLevel
             };
 
             // Now downgrade if required.
@@ -148,7 +144,7 @@ namespace MQTTnet.Server
             // 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.
-            if (maxQoSLevel < (int)applicationMessageQoSLevel)
+            if (maxQoSLevel < (int)qualityOfServiceLevel)
             {
                 result.QualityOfServiceLevel = (MqttQualityOfServiceLevel)maxQoSLevel;
             }
@@ -169,10 +165,7 @@ namespace MQTTnet.Server
             }
 
             var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false);
-            var result = new SubscribeResult
-            {
-                ReasonCodes = new List<MqttSubscribeReasonCode>(subscribePacket.TopicFilters.Count)
-            };
+            var result = new SubscribeResult(subscribePacket.TopicFilters.Count);
 
             var addedSubscriptions = new List<string>();
             var finalTopicFilters = new List<MqttTopicFilter>();
@@ -212,7 +205,7 @@ namespace MQTTnet.Server
             // This call will add the new subscription to the internal storage.
             // So the event _ClientSubscribedTopicEvent_ must be called afterwards.
             _subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
-            
+
             if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
             {
                 foreach (var finalTopicFilter in finalTopicFilters)
@@ -221,23 +214,22 @@ namespace MQTTnet.Server
                     await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
             }
-            
+
             return result;
         }
 
-        public async Task<MqttUnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        public async Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
             if (unsubscribePacket == null)
             {
                 throw new ArgumentNullException(nameof(unsubscribePacket));
             }
 
-            var result = new MqttUnsubscribeResult();
+            var result = new UnsubscribeResult();
 
             var removedSubscriptions = new List<string>();
 
-            await _subscriptionsLock.WaitAsync(cancellationToken).ConfigureAwait(false);
-            try
+            using (await _subscriptionsLock.EnterAsync(cancellationToken).ConfigureAwait(false))
             {
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
@@ -268,7 +260,7 @@ namespace MQTTnet.Server
                         if (existingSubscription != null)
                         {
                             var topicHash = existingSubscription.TopicHash;
-                            
+
                             if (existingSubscription.TopicHasWildcard)
                             {
                                 if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
@@ -297,11 +289,8 @@ namespace MQTTnet.Server
                     }
                 }
             }
-            finally
-            {
-                _subscriptionsLock.Release();
-                _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
-            }
+
+            _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
 
             if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers)
             {
@@ -348,8 +337,7 @@ namespace MQTTnet.Server
 
             // Add to subscriptions and maintain topic hash dictionaries
 
-            _subscriptionsLock.Wait();
-            try
+            using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
             {
                 MqttSubscription.CalculateTopicHash(topicFilter.Topic, out var topicHash, out var topicHashMask, out var hasWildcard);
 
@@ -399,10 +387,6 @@ namespace MQTTnet.Server
                     subscriptions.Add(subscription);
                 }
             }
-            finally
-            {
-                _subscriptionsLock.Release();
-            }
 
             return new CreateSubscriptionResult
             {
@@ -412,14 +396,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];
-                if (retainedApplicationMessage == null)
+                var retainedMessage = retainedMessages[index];
+                if (retainedMessage == null)
                 {
                     continue;
                 }
@@ -436,16 +420,21 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch)
+                if (MqttTopicFilterComparer.Compare(retainedMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch)
                 {
                     continue;
                 }
 
-                var retainedMessageMatch = new MqttRetainedMessageMatch
+                var retainedMessageMatch = new MqttRetainedMessageMatch(retainedMessage, createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel);
+                if (retainedMessageMatch.SubscriptionQualityOfServiceLevel > retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel)
                 {
-                    ApplicationMessage = retainedApplicationMessage,
-                    SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel
-                };
+                    // UPGRADING the QoS is not allowed! 
+                    // From MQTT spec: Subscribing to a Topic Filter at QoS 2 is equivalent to saying
+                    // "I would like to receive Messages matching this filter at the QoS with which they were published".
+                    // This means a publisher is responsible for determining the maximum QoS a Message can be delivered at,
+                    // but a subscriber is able to require that the Server downgrades the QoS to one more suitable for its usage.
+                    retainedMessageMatch.SubscriptionQualityOfServiceLevel = retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel;
+                }
 
                 if (subscribeResult.RetainedMessages == null)
                 {
@@ -456,7 +445,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;
             }
         }
 
@@ -491,23 +480,20 @@ namespace MQTTnet.Server
 
         async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
         {
-            var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter);
-
-            if (mqttSubscription == null)
+            var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter)
             {
-                clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted;
-            }
-            else
-            {
-                clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.Success;
-            }
+                Response =
+                {
+                    ReasonCode = mqttSubscription == null ? MqttUnsubscribeReasonCode.NoSubscriptionExisted : MqttUnsubscribeReasonCode.Success
+                }
+            };
 
             await _eventContainer.InterceptingUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false);
 
             return clientUnsubscribingTopicEventArgs;
         }
 
-        public sealed class CreateSubscriptionResult
+        sealed class CreateSubscriptionResult
         {
             public bool IsNewSubscription { get; set; }
 

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -23,6 +23,7 @@ namespace MQTTnet.Server
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         readonly MqttServerOptions _serverOptions;
+        readonly MqttClientSubscriptionsManager _subscriptionsManager;
 
         readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
 
@@ -45,7 +46,7 @@ namespace MQTTnet.Server
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
 
-            SubscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
+            _subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
         }
 
         public DateTime CreatedTimestamp { get; } = DateTime.UtcNow;
@@ -67,18 +68,8 @@ namespace MQTTnet.Server
 
         public long PendingDataPacketsCount => _packetBus.PartitionItemsCount(MqttPacketBusPartition.Data);
 
-        public MqttClientSubscriptionsManager SubscriptionsManager { get; }
-
         public bool WillMessageSent { get; set; }
 
-        public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
-        {
-            // 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;
-        }
-        
         public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
         {
             _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
@@ -107,8 +98,8 @@ namespace MQTTnet.Server
 
         public void Dispose()
         {
-            _packetBus?.Dispose();
-            SubscriptionsManager.Dispose();
+            _packetBus.Dispose();
+            _subscriptionsManager.Dispose();
         }
 
         public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
@@ -150,6 +141,14 @@ namespace MQTTnet.Server
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
         }
 
+        public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
+        {
+            // 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;
+        }
+
         public void Recover()
         {
             // TODO: Keep the bus and only insert pending items again.
@@ -167,5 +166,30 @@ namespace MQTTnet.Server
         {
             _subscribedTopics?.Remove(topic);
         }
+
+        public Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+        {
+            return _subscriptionsManager.Subscribe(subscribePacket, cancellationToken);
+        }
+
+        public bool TryCheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId, out CheckSubscriptionsResult result)
+        {
+            result = null;
+
+            try
+            {
+                result = _subscriptionsManager.CheckSubscriptions(topic, topicHash, qualityOfServiceLevel, senderId);
+                return true;
+            }
+            catch
+            {
+                return false;
+            }
+        }
+
+        public Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        {
+            return _subscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken);
+        }
     }
 }

Source/MQTTnet/Server/Internal/MqttSubscription.cs

@@ -179,7 +179,7 @@ namespace MQTTnet.Server
             RetainAsPublished = retainAsPublished;
             GrantedQualityOfServiceLevel = qualityOfServiceLevel;
             Identifier = identifier;
-            
+
             CalculateTopicHash(Topic, out var hash, out var hashMask, out var hasWildcard);
             TopicHash = hash;
             TopicHashMask = hashMask;

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/Internal/UnsubscribeResult.cs

@@ -7,7 +7,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttUnsubscribeResult
+    public sealed class UnsubscribeResult
     {
         public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
         

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