Add new samples.

fix cross access to async events.

.github/workflows/ReleaseNotes.md

@@ -0,0 +1,7 @@
+* [Core] MQTT Packets being sent over web socket transport are now setting the web socket frame boundaries correctly (#1499).
+* [Core] Add support for attaching and detaching events from different threads.
+* [Core] Fixed a deadlock in _AsyncLock_ implementation (#1520).
+* [Client] Keep alive mechanism now uses the configured timeout value from the options (thanks to @Stannieman, #1495).
+* [Client] The _PingAsync_ will fallback to the timeout specified in the client options when the cancellation token cannot be cancelled.
+* [Server] A DISCONNECT packet is no longer sent to MQTT clients < 5.0.0 (thanks to @logicaloud, #1506).
+* [Server] Improved "take over" process handling.

.github/workflows/ci.yml

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

MQTTnet.sln.DotSettings

@@ -210,6 +210,7 @@
         </Or>
       </Entry.Match>
         <Entry.SortBy>
+            <Access Order="Public Internal Protected Private" />
             <Name />
         </Entry.SortBy>
     </Entry>
@@ -240,5 +241,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_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<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/=Tnet/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

README.md

@@ -12,7 +12,7 @@
 
 # MQTTnet
 
-MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
+MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework version and CPU architecture.
 
 ## Features
 
@@ -50,39 +50,6 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * A custom message interceptor can be added which allows transforming or extending every received application message
 * Validate subscriptions and deny subscribing of certain topics depending on requesting clients
 
-## Supported frameworks
-
-| Framwork           | Version     |
-| ------------------ | ----------- |
-|.NET                | 5.0+        |
-|.NET Framework      | 4.5.2+      |
-|.NET Standard       | 1.3+        |
-|.NET Core           | 1.1+        |
-|.NET Core App       | 1.1+        |
-| Mono               | 5.2+        |
-| UWP                | 10.0.10240+ |
-| Xamarin.Android    | 7.5+        |
-| Xamarin.iOS        | 10.14+      |
-| Blazor WebAssembly | 3.2.0+      |
-
-## References
-
-This library is used in the following projects:
-
-* Azure Functions MQTT Bindings, <https://github.com/keesschollaart81/CaseOnline.Azure.WebJobs.Extensions.Mqtt>)
-* HA4IoT (Open Source Home Automation system for .NET, <https://github.com/chkr1011/HA4IoT>)
-* MQTT Client Rx (Wrapper for Reactive Extensions, <https://github.com/1iveowl/MQTTClient.rx>)
-* MQTT Client Rx (Managed Client Wrapper for Reactive Extensions, <https://github.com/mmuecke/RxMQTTnet>)
-* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8))
-* MQTTnet App (Cross platform client application for MQTT debugging, inspection etc., <https://github.com/chkr1011/MQTTnetApp>)
-* Wirehome.Core (Open Source Home Automation system for .NET Core, <https://github.com/chkr1011/Wirehome.Core>)
-* SparkplugNet (Sparkplug library for .Net, <https://github.com/SeppPenner/SparkplugNet>)
-* Silverback (Framework to build event-driven applications - support for MQTT, Kafka & RabbitMQ) <https://github.com/BEagle1984/silverback>
-
-Further projects using this project can be found under https://github.com/dotnet/MQTTnet/network/dependents.
-
-If you use this library and want to see your project here please create a pull request.
-
 ## Code of Conduct
 
 This project has adopted the code of conduct defined by the Contributor Covenant to clarify expected behavior in our community.

Samples/Client/Client_Connection_Samples.cs

@@ -10,7 +10,7 @@ using System.Security.Authentication;
 using MQTTnet.Client;
 using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
-using MQTTnet.Samples.Shared;
+using MQTTnet.Samples.Helpers;
 
 namespace MQTTnet.Samples.Client;
 
@@ -321,14 +321,12 @@ public static class Client_Connection_Samples
                     {
                         try
                         {
-                            // This code will also do the very first connect! So no call to _ConnectAsync_ is required
+                            // This code will also do the very first connect! So no call to _ConnectAsync_ is required in the first place.
-                            // in the first place.
+                            if (!await mqttClient.TryPingAsync())
-                            if (!mqttClient.IsConnected)
                             {
                                 await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
                                 // Subscribe to topics when session is clean etc.
-
                                 Console.WriteLine("The MQTT client is connected.");
                             }
                         }

Samples/Client/Client_Publish_Samples.cs

@@ -40,6 +40,53 @@ public static class Client_Publish_Samples
 
             await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
 
+            await mqttClient.DisconnectAsync();
+            
+            Console.WriteLine("MQTT application message is published.");
+        }
+    }
+    
+    public static async Task Publish_Multiple_Application_Messages()
+    {
+        /*
+         * This sample pushes multiple simple application message including a topic and a payload.
+         *
+         * See sample _Publish_Application_Message_ for more details.
+         */
+
+        var mqttFactory = new MqttFactory();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("broker.hivemq.com")
+                .Build();
+
+            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            var applicationMessage = new MqttApplicationMessageBuilder()
+                .WithTopic("samples/temperature/living_room")
+                .WithPayload("19.5")
+                .Build();
+
+            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
+            
+            applicationMessage = new MqttApplicationMessageBuilder()
+                .WithTopic("samples/temperature/living_room")
+                .WithPayload("20.0")
+                .Build();
+            
+            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
+
+            applicationMessage = new MqttApplicationMessageBuilder()
+                .WithTopic("samples/temperature/living_room")
+                .WithPayload("21.0")
+                .Build();
+            
+            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
+
+            await mqttClient.DisconnectAsync();
+            
             Console.WriteLine("MQTT application message is published.");
         }
     }

Samples/Client/Client_Subscribe_Samples.cs

@@ -7,7 +7,8 @@
 // ReSharper disable InconsistentNaming
 
 using MQTTnet.Client;
-using MQTTnet.Samples.Shared;
+using MQTTnet.Packets;
+using MQTTnet.Samples.Helpers;
 
 namespace MQTTnet.Samples.Client;
 
@@ -81,4 +82,48 @@ public static class Client_Subscribe_Samples
             response.DumpToConsole();
         }
     }
+    
+    public static async Task Send_Responses()
+    {
+        /*
+         * This sample subscribes to a topic and sends a response to the broker. This requires at least QoS level 1 to work!
+         */
+
+        var mqttFactory = new MqttFactory();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            mqttClient.ApplicationMessageReceivedAsync += delegate(MqttApplicationMessageReceivedEventArgs args)
+            {
+                // Do some work with the message...
+
+                // Now respond to the broker with a reason code other than success.
+                args.ReasonCode = MqttApplicationMessageReceivedReasonCode.ImplementationSpecificError;
+                args.ResponseReasonString = "That did not work!";
+                
+                // User properties require MQTT v5!
+                args.ResponseUserProperties.Add(new MqttUserProperty("My", "Data"));
+                
+                // Now the broker will resend the message again.
+                return Task.CompletedTask;
+            };
+            
+            var mqttClientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("broker.hivemq.com")
+                .Build();
+
+            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
+                .WithTopicFilter(f => { f.WithTopic("mqttnet/samples/topic/1"); })
+                .Build();
+
+            var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+
+            Console.WriteLine("MQTT client subscribed to topic.");
+
+            // The response contains additional data sent by the server after subscribing.
+            response.DumpToConsole();
+        }
+    }
 }

Samples/Diagnostics/Logger_Samples.cs

@@ -8,7 +8,6 @@
 
 using System.Text;
 using MQTTnet.Diagnostics;
-using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Diagnostics;
 
@@ -23,7 +22,9 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(new MyLogger());
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
+            .WithTcpServer("broker.hivemq.com")
+            .Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -31,31 +32,13 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
+                .Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
     }
 
-    public static async Task Log_To_Console()
-    {
-        /*
-       * This sample covers the creation of a custom logger which is used to log to the console.
-       */
-
-        var mqttFactory = new MqttFactory(new ConsoleLogger());
-
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
-
-        using (var mqttClient = mqttFactory.CreateMqttClient())
-        {
-            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
-
-            Console.WriteLine("MQTT client is connected. Press any key to exit.");
-            Console.ReadKey(true);
-        }
-    }
-
     public static async Task Use_Event_Logger()
     {
         /*
@@ -81,7 +64,9 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(mqttEventLogger);
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
+            .WithTcpServer("broker.hivemq.com")
+            .Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -89,7 +74,8 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
+                .Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
@@ -97,7 +83,7 @@ public static class Logger_Samples
 
     sealed class MyLogger : IMqttNetLogger
     {
-        public bool IsEnabled { get; } = true;
+        public bool IsEnabled { get; set; } = true;
 
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {

Samples/Helpers/ObjectExtensions.cs

@@ -4,7 +4,7 @@
 
 using System.Text.Json;
 
-namespace MQTTnet.Samples.Shared;
+namespace MQTTnet.Samples.Helpers;
 
 internal static class ObjectExtensions
 {

Samples/Program.cs

@@ -8,7 +8,6 @@ Console.WriteLine("Welcome to MQTTnet samples!");
 Console.WriteLine();
 
 var sampleClasses = Assembly.GetExecutingAssembly().GetExportedTypes().OrderBy(c => c.Name).ToList();
-sampleClasses.RemoveAll(c => c.Namespace.Contains(".Shared"));
 
 var index = 0;
 foreach (var sampleClass in sampleClasses)

Samples/Server/Server_Simple_Samples.cs

@@ -6,8 +6,8 @@
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
 
+using MQTTnet.Diagnostics;
 using MQTTnet.Protocol;
-using MQTTnet.Samples.Shared;
 using MQTTnet.Server;
 
 namespace MQTTnet.Samples.Server;
@@ -170,4 +170,50 @@ public static class Server_Simple_Samples
         await server.StartAsync();
         return server;
     }
+
+    class ConsoleLogger : IMqttNetLogger
+    {
+        readonly object _consoleSyncRoot = new();
+
+        public bool IsEnabled => true;
+
+        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
+        {
+            var foregroundColor = ConsoleColor.White;
+            switch (logLevel)
+            {
+                case MqttNetLogLevel.Verbose:
+                    foregroundColor = ConsoleColor.White;
+                    break;
+
+                case MqttNetLogLevel.Info:
+                    foregroundColor = ConsoleColor.Green;
+                    break;
+
+                case MqttNetLogLevel.Warning:
+                    foregroundColor = ConsoleColor.DarkYellow;
+                    break;
+
+                case MqttNetLogLevel.Error:
+                    foregroundColor = ConsoleColor.Red;
+                    break;
+            }
+
+            if (parameters?.Length > 0)
+            {
+                message = string.Format(message, parameters);
+            }
+
+            lock (_consoleSyncRoot)
+            {
+                Console.ForegroundColor = foregroundColor;
+                Console.WriteLine(message);
+
+                if (exception != null)
+                {
+                    Console.WriteLine(exception);
+                }
+            }
+        }
+    }
 }

Samples/Shared/ConsoleLogger.cs

@@ -1,53 +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 MQTTnet.Diagnostics;
-
-namespace MQTTnet.Samples.Shared;
-
-public sealed class ConsoleLogger : IMqttNetLogger
-{
-    readonly object _consoleSyncRoot = new();
-
-    public bool IsEnabled => true;
-
-    public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
-    {
-        var foregroundColor = ConsoleColor.White;
-        switch (logLevel)
-        {
-            case MqttNetLogLevel.Verbose:
-                foregroundColor = ConsoleColor.White;
-                break;
-
-            case MqttNetLogLevel.Info:
-                foregroundColor = ConsoleColor.Green;
-                break;
-
-            case MqttNetLogLevel.Warning:
-                foregroundColor = ConsoleColor.DarkYellow;
-                break;
-
-            case MqttNetLogLevel.Error:
-                foregroundColor = ConsoleColor.Red;
-                break;
-        }
-
-        if (parameters?.Length > 0)
-        {
-            message = string.Format(message, parameters);
-        }
-
-        lock (_consoleSyncRoot)
-        {
-            Console.ForegroundColor = foregroundColor;
-            Console.WriteLine(message);
-
-            if (exception != null)
-            {
-                Console.WriteLine(exception);
-            }
-        }
-    }
-}

Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs

@@ -0,0 +1,66 @@
+// 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;
+using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using MQTTnet.Internal;
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net60)]
+    [MemoryDiagnoser]
+    public class AsyncLockBenchmark : BaseBenchmark
+    {
+        [Benchmark]
+        public async Task Synchronize_100_Tasks()
+        {
+            const int tasksCount = 100;
+
+            var tasks = new Task[tasksCount];
+            var asyncLock = new AsyncLock();
+            var globalI = 0;
+
+            for (var i = 0; i < tasksCount; i++)
+            {
+                tasks[i] = Task.Run(
+                    async () =>
+                    {
+                        using (await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
+                        {
+                            var localI = globalI;
+                            await Task.Delay(5).ConfigureAwait(false); // Increase the chance for wrong data.
+                            localI++;
+                            globalI = localI;
+                        }
+                    });
+            }
+
+            await Task.WhenAll(tasks).ConfigureAwait(false);
+
+            if (globalI != tasksCount)
+            {
+                throw new Exception($"Code is broken ({globalI})!");
+            }
+        }
+        
+        [Benchmark]
+        public async Task Wait_100_000_Times()
+        {
+            var asyncLock = new AsyncLock();
+
+            using (var cancellationToken = new CancellationTokenSource())
+            {
+                for (var i = 0; i < 100000; i++)
+                {
+                    using (await asyncLock.WaitAsync(cancellationToken.Token).ConfigureAwait(false))
+                    {
+                    }
+                }
+            }
+        }
+    }
+}

Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -12,14 +12,10 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="BenchmarkDotNet" Version="0.13.1" />
+        <PackageReference Include="BenchmarkDotNet" Version="0.13.2" />
         <PackageReference Include="Microsoft.AspNetCore" Version="2.2.0" />
     </ItemGroup>
     
-    <ItemGroup Condition="'$(TargetFramework)' != 'net5.0'">
-        <PackageReference Include="System.IO.Pipelines" Version="4.7.2" />
-    </ItemGroup>
-
     <ItemGroup Condition="'$(TargetFramework)' == 'net5.0'">
         <PackageReference Include="System.IO.Pipelines" Version="5.0.0" />
     </ItemGroup>

Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -2,6 +2,7 @@
 // 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 BenchmarkDotNet.Attributes;
 using MQTTnet.Channel;
 using MQTTnet.Implementations;
@@ -70,8 +71,8 @@ namespace MQTTnet.Benchmarks
 
             while (read < expected)
             {
-                var readresult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
+                var readResult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
-                read += readresult;
+                read += readResult;
             }
         }
 
@@ -79,9 +80,11 @@ namespace MQTTnet.Benchmarks
         {
             await Task.Yield();
 
+            var buffer = new ArraySegment<byte>(new byte[size]);
+            
             for (var i = 0; i < iterations; i++)
             {
-                await _serverChannel.WriteAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
+                await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
             }
         }
     }

Source/MQTTnet.Benchmarks/Program.cs

@@ -29,6 +29,7 @@ namespace MQTTnet.Benchmarks
             Console.WriteLine("c = SubscribeBenchmark");
             Console.WriteLine("d = UnsubscribeBenchmark");
             Console.WriteLine("e = MessageDeliveryBenchmark");
+            Console.WriteLine("f = AsyncLockBenchmark");
 
             var pressedKey = Console.ReadKey(true);
             switch (pressedKey.KeyChar)
@@ -75,6 +76,9 @@ namespace MQTTnet.Benchmarks
                 case 'e':
                     BenchmarkRunner.Run<MessageDeliveryBenchmark>();
                     break;
+                case 'f':
+                    BenchmarkRunner.Run<AsyncLockBenchmark>();
+                    break;
             }
 
             Console.ReadLine();

Source/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -104,7 +104,7 @@ namespace MQTTnet.Benchmarks
                 return Task.FromResult(count);
             }
 
-            public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+            public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
             {
                 throw new NotSupportedException();
             }

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -154,9 +154,9 @@ namespace MQTTnet.Extensions.WebSocket4Net
             return Task.FromResult(readBytes);
         }
 
-        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
         {
-            _webSocket.Send(buffer, offset, count);
+            _webSocket.Send(buffer.Array, buffer.Offset, buffer.Count);
             return Task.FromResult(0);
         }
 

Source/MQTTnet.TestApp/AsyncLockTest.cs

@@ -0,0 +1,28 @@
+// 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;
+using System.Threading.Tasks;
+using MQTTnet.Internal;
+
+namespace MQTTnet.TestApp
+{
+    public sealed class AsyncLockTest
+    {
+        public async Task Run()
+        {
+            var asyncLock = new AsyncLock();
+
+            using (var cancellationToken = new CancellationTokenSource())
+            {
+                for (var i = 0; i < 100000; i++)
+                {
+                    using (await asyncLock.WaitAsync(cancellationToken.Token).ConfigureAwait(false))
+                    {
+                    }
+                } 
+            }
+        }
+    }
+}

Source/MQTTnet.TestApp/Program.cs

@@ -28,6 +28,7 @@ namespace MQTTnet.TestApp
             Console.WriteLine("c = Start QoS 0 benchmark");
             Console.WriteLine("d = Start server with logging");
             Console.WriteLine("e = Start Message Throughput Test");
+            Console.WriteLine("f = Start AsyncLock Test");
 
             var pressedKey = Console.ReadKey(true);
             if (pressedKey.KeyChar == '1')
@@ -88,6 +89,10 @@ namespace MQTTnet.TestApp
             {
                 Task.Run(new MessageThroughputTest().Run);
             }
+            else if (pressedKey.KeyChar == 'f')
+            {
+                Task.Run(new AsyncLockTest().Run);
+            }
 
             Thread.Sleep(Timeout.Infinite);
         }

Source/MQTTnet.Tests/Client/MqttClient_Tests.cs

@@ -19,10 +19,12 @@ using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Tests.Mockups;
 
+// ReSharper disable InconsistentNaming
+
 namespace MQTTnet.Tests.Client
 {
     [TestClass]
-    public sealed class Client_Tests : BaseTestClass
+    public sealed class MqttClient_Tests : BaseTestClass
     {
         [DataTestMethod]
         [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
@@ -145,6 +147,7 @@ namespace MQTTnet.Tests.Client
                 }
                 catch
                 {
+                    // Ignore errors.
                 }
 
                 await Task.Delay(500);
@@ -239,10 +242,10 @@ namespace MQTTnet.Tests.Client
 
                 for (var i = 0; i < 98; i++)
                 {
-                    Assert.IsFalse(clients[i].IsConnected, $"clients[{i}] is not connected");
+                    Assert.IsFalse(clients[i].TryPingAsync().GetAwaiter().GetResult(), $"clients[{i}] is not connected");
                 }
 
-                Assert.IsTrue(clients[99].IsConnected);
+                Assert.IsTrue(clients[99].TryPingAsync().GetAwaiter().GetResult());
 
                 Assert.AreEqual(1, clientStatus.Count);
                 Assert.AreEqual(1, sessionStatus.Count);
@@ -332,7 +335,7 @@ namespace MQTTnet.Tests.Client
 
                 var client = await testEnvironment.ConnectClient();
 
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult());
             }
         }
 
@@ -343,11 +346,11 @@ namespace MQTTnet.Tests.Client
             {
                 await testEnvironment.StartServer();
                 var client = await testEnvironment.ConnectClient();
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult());
 
                 await client.DisconnectAsync();
 
-                Assert.IsFalse(client.IsConnected);
+                Assert.IsFalse(client.TryPingAsync().GetAwaiter().GetResult());
             }
         }
 
@@ -591,17 +594,17 @@ namespace MQTTnet.Tests.Client
                 var client = await testEnvironment.ConnectClient();
 
                 await Task.Delay(500);
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(await client.TryPingAsync());
 
                 await server.StopAsync();
                 await Task.Delay(500);
-                Assert.IsFalse(client.IsConnected);
+                Assert.IsFalse(await client.TryPingAsync());
 
                 await server.StartAsync();
                 await Task.Delay(500);
 
                 await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(await client.TryPingAsync());
             }
         }
 
@@ -637,6 +640,7 @@ namespace MQTTnet.Tests.Client
                 }
                 catch
                 {
+                    // Ignore errors.
                 }
 
                 SpinWait.SpinUntil(() => tries >= maxTries, 10000);
@@ -656,11 +660,11 @@ namespace MQTTnet.Tests.Client
                 var client = await testEnvironment.ConnectClient();
 
                 await Task.Delay(500);
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult());
 
                 await server.StopAsync();
                 await Task.Delay(500);
-                Assert.IsFalse(client.IsConnected);
+                Assert.IsFalse(client.TryPingAsync().GetAwaiter().GetResult());
 
                 for (var i = 0; i < 5; i++)
                 {
@@ -671,6 +675,7 @@ namespace MQTTnet.Tests.Client
                     }
                     catch
                     {
+                        // Ignore errors.
                     }
                 }
 
@@ -678,7 +683,7 @@ namespace MQTTnet.Tests.Client
                 await Task.Delay(500);
 
                 await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-                Assert.IsTrue(client.IsConnected);
+                await client.PingAsync();
             }
         }
 
@@ -828,10 +833,10 @@ namespace MQTTnet.Tests.Client
         {
             using (var testEnvironment = new TestEnvironment(TestContext))
             {
-                var server = await testEnvironment.StartServer();
+                await testEnvironment.StartServer();
                 var client = await testEnvironment.ConnectClient();
 
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult());
                 client.DisconnectedAsync += e =>
                 {
                     Assert.IsTrue(e.ClientWasConnected);
@@ -851,7 +856,7 @@ namespace MQTTnet.Tests.Client
                 var server = await testEnvironment.StartServer();
                 var client = await testEnvironment.ConnectClient();
 
-                Assert.IsTrue(client.IsConnected);
+                Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult());
                 client.DisconnectedAsync += e =>
                 {
                     Assert.IsTrue(e.ClientWasConnected);
@@ -941,7 +946,7 @@ namespace MQTTnet.Tests.Client
                 await Task.Delay(500);
 
                 Assert.IsTrue(messageReceived);
-                Assert.IsTrue(client1.IsConnected);
+                Assert.IsTrue(client1.TryPingAsync().GetAwaiter().GetResult());
                 Assert.IsFalse(disconnectedFired);
             }
         }

Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs

@@ -41,6 +41,33 @@ namespace MQTTnet.Tests.Internal
             Assert.AreEqual(0, _testEventAsyncCount);
         }
 
+        [TestMethod]
+        public void Has_Handlers()
+        {
+            var testClass = new TestClass();
+            testClass.TestEventAsync += OnTestEventAsync;
+
+            Assert.AreEqual(true, testClass.HasTestHandlers);
+        }
+
+        [TestMethod]
+        public void No_Handlers()
+        {
+            var testClass = new TestClass();
+
+            Assert.AreEqual(false, testClass.HasTestHandlers);
+        }
+
+        [TestMethod]
+        public void Remove_Handlers()
+        {
+            var testClass = new TestClass();
+            testClass.TestEventAsync += OnTestEventAsync;
+            testClass.TestEventAsync -= OnTestEventAsync;
+
+            Assert.AreEqual(false, testClass.HasTestHandlers);
+        }
+
         Task OnTestEventAsync(EventArgs arg)
         {
             Interlocked.Increment(ref _testEventAsyncCount);
@@ -57,6 +84,8 @@ namespace MQTTnet.Tests.Internal
                 remove => _testEvent.RemoveHandler(value);
             }
 
+            public bool HasTestHandlers => _testEvent.HasHandlers;
+
             public Task FireEventAsync(EventArgs eventArgs)
             {
                 return _testEvent.InvokeAsync(eventArgs);

Source/MQTTnet.Tests/Internal/AsyncLock_Tests.cs

@@ -3,6 +3,7 @@
 // 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;
@@ -13,6 +14,80 @@ namespace MQTTnet.Tests.Internal
     [TestClass]
     public class AsyncLock_Tests
     {
+        [TestMethod]
+        public void Lock_10_Parallel_Tasks()
+        {
+            const int ThreadsCount = 10;
+
+            var threads = new Task[ThreadsCount];
+            var @lock = new AsyncLock();
+            var globalI = 0;
+            for (var i = 0; i < ThreadsCount; i++)
+            {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                threads[i] = Task.Run(
+                    async () =>
+                    {
+                        using (await @lock.WaitAsync(CancellationToken.None))
+                        {
+                            var localI = globalI;
+                            await Task.Delay(10); // Increase the chance for wrong data.
+                            localI++;
+                            globalI = localI;
+                        }
+                    });
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            }
+
+            Task.WaitAll(threads);
+            Assert.AreEqual(ThreadsCount, globalI);
+        }
+
+        [TestMethod]
+        public void Lock_10_Parallel_Tasks_With_Dispose_Doesnt_Lockup()
+        {
+            const int ThreadsCount = 10;
+
+            var threads = new Task[ThreadsCount];
+            var @lock = new AsyncLock();
+            var globalI = 0;
+            for (var i = 0; i < ThreadsCount; i++)
+            {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                threads[i] = Task.Run(
+                        async () =>
+                        {
+                            using (await @lock.WaitAsync(CancellationToken.None))
+                            {
+                                var localI = globalI;
+                                await Task.Delay(10); // Increase the chance for wrong data.
+                                localI++;
+                                globalI = localI;
+                            }
+                        })
+                    .ContinueWith(
+                        x =>
+                        {
+                            if (globalI == 5)
+                            {
+                                @lock.Dispose();
+                                @lock = new AsyncLock();
+                            }
+
+                            if (x.Exception != null)
+                            {
+                                Debug.WriteLine(x.Exception.GetBaseException().GetType().Name);
+                            }
+                        });
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            }
+
+            Task.WaitAll(threads);
+
+            // Expect only 6 because the others are failing due to disposal (if (globalI == 5)).
+            Assert.AreEqual(6, globalI);
+        }
+
         [TestMethod]
         public async Task Lock_Serial_Calls()
         {
@@ -45,58 +120,32 @@ namespace MQTTnet.Tests.Internal
             }
         }
 
-        //[TestMethod]
-        //public async Task Test_Cancellation_With_Later_Access()
-        //{
-        //    var @lock = new AsyncLock();
-
-        //    var releaser = await @lock.WaitAsync().ConfigureAwait(false);
-
-        //    try
-        //    {
-        //        using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
-        //        {
-        //            await @lock.WaitAsync(cts.Token).ConfigureAwait(false);
-        //        }
-        //    }
-        //    catch (OperationCanceledException)
-        //    {
-        //    }
-            
-        //    releaser.Dispose();
-
-        //    using (await @lock.WaitAsync().ConfigureAwait(false))
-        //    {
-        //        // When the method finished, the thread got access.
-        //    }
-        //}
-
         [TestMethod]
-        public void Lock_10_Parallel_Tasks()
+        public async Task Test_Cancellation_With_Later_Access()
         {
-            const int ThreadsCount = 10;
+            var asyncLock = new AsyncLock();
 
-            var threads = new Task[ThreadsCount];
+            var releaser = await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false);
-            var @lock = new AsyncLock();
+
-            var globalI = 0;
+            try
-            for (var i = 0; i < ThreadsCount; i++)
             {
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
-                threads[i] = Task.Run(async () =>
                 {
-                    using (var releaser = await @lock.WaitAsync(CancellationToken.None))
+                    await asyncLock.WaitAsync(timeout.Token).ConfigureAwait(false);
-                    {
+                }
-                        var localI = globalI;
+                
-                        await Task.Delay(10); // Increase the chance for wrong data.
+                Assert.Fail("Exception should be thrown!");
-                        localI++;
+            }
-                        globalI = localI;
+            catch (OperationCanceledException)
-                    }
+            {
-                });
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
             }
 
-            Task.WaitAll(threads);
+            releaser.Dispose();
-            Assert.AreEqual(ThreadsCount, globalI);
+
+            using (await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
+            {
+                // When the method finished, the thread got access.
+            }
         }
     }
 }

Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs

@@ -175,6 +175,49 @@ namespace MQTTnet.Tests.MQTTv5
             }
         }
 
+        [TestMethod]
+        public async Task Disconnect_with_Reason()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var disconnectReason = MQTTnet.Client.MqttClientDisconnectReason.UnspecifiedError;
+
+                string testClientId = null;
+
+                await testEnvironment.StartServer();
+
+                testEnvironment.Server.ClientConnectedAsync += e =>
+                {
+                    testClientId = e.ClientId;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client = await testEnvironment.ConnectClient(new MQTTnet.Client.MqttClientOptionsBuilder().WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500));
+
+                client.DisconnectedAsync += e =>
+                {
+                    disconnectReason = e.Reason;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                await LongTestDelay();
+
+                // Test client should be connected now
+
+                Assert.IsTrue(testClientId != null);
+
+                // Have the server disconnect the client with AdministrativeAction reason
+
+                await testEnvironment.Server.DisconnectClientAsync(testClientId, Protocol.MqttDisconnectReasonCode.AdministrativeAction);
+
+                await LongTestDelay();
+
+                // The reason should be returned to the client in the DISCONNECT packet
+
+                Assert.AreEqual(MQTTnet.Client.MqttClientDisconnectReason.AdministrativeAction, disconnectReason);
+            }
+        }
+
         MqttClientOptions CreateClientOptions(TestEnvironment testEnvironment, string clientId, bool cleanSession, uint sessionExpiryInterval)
         {
             return testEnvironment.Factory.CreateClientOptionsBuilder()

Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs

@@ -2,6 +2,7 @@
 // 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.IO;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
@@ -45,9 +46,9 @@ namespace MQTTnet.Tests.Mockups
             return _stream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
-        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
         {
-            return _stream.WriteAsync(buffer, offset, count, cancellationToken);
+            return _stream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken);
         }
 
         public void Dispose()

Source/MQTTnet.Tests/Server/General.cs

@@ -936,6 +936,61 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Disconnect_Client_with_Reason()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var disconnectPacketReceived = false;
+
+                string testClientId = null;
+
+                await testEnvironment.StartServer();
+
+                testEnvironment.Server.ClientConnectedAsync += e =>
+                {
+                    testClientId = e.ClientId;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client = testEnvironment.CreateClient();
+
+                client.InspectPackage += e =>
+                {
+                    if (e.Buffer.Length > 0)
+                    {
+                        if (e.Buffer[0] == (byte)Protocol.MqttControlPacketType.Disconnect << 4)
+                        {
+                            disconnectPacketReceived = true;
+                        }
+                    }
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                await client.ConnectAsync(
+                    new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort)
+                    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
+                    .Build()
+                    );
+
+                await LongTestDelay();
+
+                // Test client should be connected now
+
+                Assert.IsTrue(testClientId != null);
+
+                // Have the server disconnect the client with SessionTakenOver reason; no DISCONNECT should be sent to MQTT 3.1.1 clients
+
+                await testEnvironment.Server.DisconnectClientAsync(testClientId, Protocol.MqttDisconnectReasonCode.SessionTakenOver);
+
+                await LongTestDelay();
+
+                // MQTT 3.1.1 client should not receive a disconnect packet
+
+                Assert.IsFalse(disconnectPacketReceived, "Disconnect packet received when none is expected");
+            }
+        }
+
         void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs)
         {
             if (_connected.ContainsKey(eventArgs.ClientId))

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -124,7 +124,9 @@ namespace MQTTnet.Tests.Server
         }
 
         [TestMethod]
-        public async Task Handle_Parallel_Connection_Attempts()
+        [DataRow(MqttProtocolVersion.V311)]
+        [DataRow(MqttProtocolVersion.V500)]
+        public async Task Handle_Parallel_Connection_Attempts(MqttProtocolVersion protocolVersion)
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
@@ -132,7 +134,7 @@ namespace MQTTnet.Tests.Server
 
                 await testEnvironment.StartServer();
 
-                var options = new MqttClientOptionsBuilder().WithClientId("1").WithKeepAlivePeriod(TimeSpan.FromSeconds(5));
+                var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion).WithKeepAlivePeriod(TimeSpan.FromSeconds(5));
 
                 var hasReceive = false;
 
@@ -146,7 +148,10 @@ namespace MQTTnet.Tests.Server
 
                 // Try to connect 50 clients at the same time.
                 var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
-                var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
+
+                await LongTestDelay();
+                
+                var connectedClients = clients.Where(c => c != null).Where(c => c.TryPingAsync().GetAwaiter().GetResult()).ToList();
 
                 Assert.AreEqual(1, connectedClients.Count);
 
@@ -162,25 +167,6 @@ namespace MQTTnet.Tests.Server
             }
         }
         
-        [TestMethod]
-        public async Task Manage_Session_MaxParallel()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                testEnvironment.IgnoreClientLogErrors = true;
-                var serverOptions = new MqttServerOptionsBuilder();
-                await testEnvironment.StartServer(serverOptions);
-
-                var options = new MqttClientOptionsBuilder().WithClientId("1");
-
-                var clients = await Task.WhenAll(Enumerable.Range(0, 10).Select(i => TryConnect(testEnvironment, options)));
-
-                var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
-
-                Assert.AreEqual(1, connectedClients.Count);
-            }
-        }
-
         [DataTestMethod]
         [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
         [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
@@ -360,7 +346,7 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
+        static async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
         {
             try
             {
@@ -379,13 +365,13 @@ namespace MQTTnet.Tests.Server
 
                 return sendClient;
             }
-            catch (Exception)
+            catch
             {
                 return null;
             }
         }
 
-        async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
+        static async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
         {
             try
             {

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -188,11 +188,14 @@ namespace MQTTnet.Adapter
 
                     _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
 
-                    await _channel.WriteAsync(packetBuffer.Packet.Array, packetBuffer.Packet.Offset, packetBuffer.Packet.Count, cancellationToken).ConfigureAwait(false);
-
                     if (packetBuffer.Payload.Count > 0)
                     {
-                        await _channel.WriteAsync(packetBuffer.Payload.Array, packetBuffer.Payload.Offset, packetBuffer.Payload.Count, cancellationToken).ConfigureAwait(false);
+                        await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false);
+                        await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false);
                     }
 
                     Interlocked.Add(ref _bytesReceived, packetBuffer.Length);

Source/MQTTnet/Channel/IMqttChannel.cs

@@ -12,13 +12,17 @@ namespace MQTTnet.Channel
     public interface IMqttChannel : IDisposable
     {
         string Endpoint { get; }
+        
         bool IsSecureConnection { get; }
+        
         X509Certificate2 ClientCertificate { get; }
 
         Task ConnectAsync(CancellationToken cancellationToken);
+        
         Task DisconnectAsync(CancellationToken cancellationToken);
 
         Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken);
-        Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken);
+        
+        Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken);
     }
 }

Source/MQTTnet/Client/IMqttClient.cs

@@ -7,14 +7,13 @@ namespace MQTTnet.Client
 {
     public interface IMqttClient : IDisposable
     {
+        event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
         event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
 
         event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync;
 
         event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
 
-        event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
-        
         event Func<InspectMqttPacketEventArgs, Task> InspectPackage;
 
         bool IsConnected { get; }
@@ -27,12 +26,12 @@ namespace MQTTnet.Client
 
         Task PingAsync(CancellationToken cancellationToken = default);
 
+        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
+
         Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default);
 
         Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default);
 
         Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default);
-        
-        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -2,59 +2,58 @@
 // 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;
 using MQTTnet.Adapter;
+using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-using MQTTnet.Formatter;
-using MQTTnet.Implementations;
 
 namespace MQTTnet.Client
 {
     public sealed class MqttClient : Disposable, IMqttClient
     {
-        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
+        readonly IMqttClientAdapterFactory _adapterFactory;
+        readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
+        readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
 
         readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
         readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory();
         readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory();
-        readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
-        
-        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
-        readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
-        readonly object _disconnectLock = new object();
 
         readonly AsyncEvent<MqttClientConnectedEventArgs> _connectedEvent = new AsyncEvent<MqttClientConnectedEventArgs>();
         readonly AsyncEvent<MqttClientConnectingEventArgs> _connectingEvent = new AsyncEvent<MqttClientConnectingEventArgs>();
         readonly AsyncEvent<MqttClientDisconnectedEventArgs> _disconnectedEvent = new AsyncEvent<MqttClientDisconnectedEventArgs>();
-        readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
+        readonly object _disconnectLock = new object();
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
-
-        readonly IMqttClientAdapterFactory _adapterFactory;
-        readonly IMqttNetLogger _rootLogger;
         readonly MqttNetSourceLogger _logger;
+        readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
+        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
 
-        CancellationTokenSource _backgroundCancellationTokenSource;
+        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
-        Task _packetReceiverTask;
+        readonly IMqttNetLogger _rootLogger;
-        Task _keepAlivePacketsSenderTask;
-        Task _publishPacketReceiverTask;
-
-        AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
 
         IMqttChannelAdapter _adapter;
+
+        CancellationTokenSource _backgroundCancellationTokenSource;
         bool _cleanDisconnectInitiated;
         volatile int _connectionStatus;
 
         MqttClientDisconnectReason _disconnectReason;
+        string _disconnectReasonString;
+        Task _keepAlivePacketsSenderTask;
 
         DateTime _lastPacketSentTimestamp;
-        string _disconnectReasonString;
+        Task _packetReceiverTask;
+
+        AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
+        Task _publishPacketReceiverTask;
 
         public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
         {
@@ -63,6 +62,12 @@ namespace MQTTnet.Client
             _logger = logger.WithSource(nameof(MqttClient));
         }
 
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => _applicationMessageReceivedEvent.AddHandler(value);
+            remove => _applicationMessageReceivedEvent.RemoveHandler(value);
+        }
+
         public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
         {
             add => _connectedEvent.AddHandler(value);
@@ -81,12 +86,6 @@ namespace MQTTnet.Client
             remove => _disconnectedEvent.RemoveHandler(value);
         }
 
-        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
-        {
-            add => _applicationMessageReceivedEvent.AddHandler(value);
-            remove => _applicationMessageReceivedEvent.RemoveHandler(value);
-        }
-        
         public event Func<InspectMqttPacketEventArgs, Task> InspectPackage
         {
             add => _inspectPacketEvent.AddHandler(value);
@@ -99,8 +98,15 @@ namespace MQTTnet.Client
 
         public async Task<MqttClientConnectResult> ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default)
         {
-            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (options == null)
-            if (options.ChannelOptions == null) throw new ArgumentException("ChannelOptions are not set.");
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            if (options.ChannelOptions == null)
+            {
+                throw new ArgumentException("ChannelOptions are not set.");
+            }
 
             ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
 
@@ -186,7 +192,10 @@ namespace MQTTnet.Client
 
         public async Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default)
         {
-            if (options is null) throw new ArgumentNullException(nameof(options));
+            if (options is null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
 
             ThrowIfDisposed();
 
@@ -213,63 +222,19 @@ namespace MQTTnet.Client
             }
         }
 
-        public Task PingAsync(CancellationToken cancellationToken = default)
+        public async Task PingAsync(CancellationToken cancellationToken = default)
         {
-            return SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken);
+            if (!cancellationToken.CanBeCanceled)
-        }
-
-        public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default)
-        {
-            if (data == null) throw new ArgumentNullException(nameof(data));
-
-            ThrowIfDisposed();
-            ThrowIfNotConnected();
-
-            var authPacket = new MqttAuthPacket
             {
-                // This must always be equal to the value from the CONNECT packet. So we use it here to ensure that.
+                using (var timeout = new CancellationTokenSource(Options.Timeout))
-                AuthenticationMethod = Options.AuthenticationMethod,
+                {
-                AuthenticationData = data.AuthenticationData,
+                    await SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, timeout.Token).ConfigureAwait(false);
-                ReasonString = data.ReasonString,
+                }
-                UserProperties = data.UserProperties
+            }
-            };
+            else
-
+            {
-            return SendAsync(authPacket, cancellationToken);
+                await SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false);
-        }
-
-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            foreach (var topicFilter in options.TopicFilters)
-            {
-                MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
             }
-
-            ThrowIfDisposed();
-            ThrowIfNotConnected();
-
-            var subscribePacket = _packetFactories.Subscribe.Create(options);
-            subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-
-            var subAckPacket = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket, cancellationToken).ConfigureAwait(false);
-
-            return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket);
-        }
-
-        public async Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            ThrowIfDisposed();
-            ThrowIfNotConnected();
-            
-            var unsubscribePacket = _packetFactories.Unsubscribe.Create(options);
-            unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-
-            var unsubAckPacket = await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, cancellationToken).ConfigureAwait(false);
-
-            return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket);
         }
 
         public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default)
@@ -304,22 +269,67 @@ namespace MQTTnet.Client
             }
         }
 
-        void Cleanup()
+        public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default)
         {
-            try
+            if (data == null)
             {
-                _backgroundCancellationTokenSource?.Cancel(false);
+                throw new ArgumentNullException(nameof(data));
             }
-            finally
+
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
+            var authPacket = new MqttAuthPacket
             {
-                _backgroundCancellationTokenSource?.Dispose();
+                // This must always be equal to the value from the CONNECT packet. So we use it here to ensure that.
-                _backgroundCancellationTokenSource = null;
+                AuthenticationMethod = Options.AuthenticationMethod,
+                AuthenticationData = data.AuthenticationData,
+                ReasonString = data.ReasonString,
+                UserProperties = data.UserProperties
+            };
 
-                _publishPacketReceiverQueue?.Dispose();
+            return SendAsync(authPacket, cancellationToken);
-                _publishPacketReceiverQueue = null;
+        }
 
-                _adapter?.Dispose();
+        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default)
+        {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
             }
+
+            foreach (var topicFilter in options.TopicFilters)
+            {
+                MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
+            }
+
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
+            var subscribePacket = _packetFactories.Subscribe.Create(options);
+            subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
+            var subAckPacket = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket, cancellationToken).ConfigureAwait(false);
+
+            return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket);
+        }
+
+        public async Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default)
+        {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
+            var unsubscribePacket = _packetFactories.Unsubscribe.Create(options);
+            unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
+            var unsubAckPacket = await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, cancellationToken).ConfigureAwait(false);
+
+            return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket);
         }
 
         protected override void Dispose(bool disposing)
@@ -332,6 +342,36 @@ namespace MQTTnet.Client
             base.Dispose(disposing);
         }
 
+        Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken)
+        {
+            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                // no response required
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubAckPacket = _packetFactories.PubAck.Create(eventArgs);
+                    return SendAsync(pubAckPacket, cancellationToken);
+                }
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubRecPacket = _packetFactories.PubRec.Create(eventArgs);
+                    return SendAsync(pubRecPacket, cancellationToken);
+                }
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Received a not supported QoS level.");
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
         async Task<MqttClientConnectResult> AuthenticateAsync(MqttClientOptions options, CancellationToken cancellationToken)
         {
             MqttClientConnectResult result;
@@ -360,32 +400,27 @@ namespace MQTTnet.Client
             return result;
         }
 
-        void ThrowIfNotConnected()
+        void Cleanup()
         {
-            if (!IsConnected)
+            try
             {
-                throw new MqttCommunicationException("The client is not connected.");
+                _backgroundCancellationTokenSource?.Cancel(false);
+            }
+            finally
+            {
+                _backgroundCancellationTokenSource?.Dispose();
+                _backgroundCancellationTokenSource = null;
+
+                _publishPacketReceiverQueue?.Dispose();
+                _publishPacketReceiverQueue = null;
+
+                _adapter?.Dispose();
             }
         }
 
-        void ThrowIfConnected(string message)
+        MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand)
         {
-            if (IsConnected)
+            return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
-            {
-                throw new MqttProtocolViolationException(message);
-            }
-        }
-
-        Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
-        {
-            var clientWasConnected = IsConnected;
-            
-            if (!DisconnectIsPendingOrFinished())
-            {
-                return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected);
-            }
-            
-            return PlatformAbstractionLayer.CompletedTask;
         }
 
         async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected)
@@ -446,28 +481,166 @@ namespace MQTTnet.Client
             }
         }
 
-        void TryInitiateDisconnect()
+        Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
         {
-            lock (_disconnectLock)
+            var clientWasConnected = IsConnected;
+
+            if (!DisconnectIsPendingOrFinished())
+            {
+                return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected);
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        bool DisconnectIsPendingOrFinished()
+        {
+            var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
+
+            do
+            {
+                switch (connectionStatus)
+                {
+                    case MqttClientConnectionStatus.Disconnected:
+                    case MqttClientConnectionStatus.Disconnecting:
+                        return true;
+                    case MqttClientConnectionStatus.Connected:
+                    case MqttClientConnectionStatus.Connecting:
+                        // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards.
+                        // So the first caller will get a "false" and all subsequent ones will get "true".
+                        var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus);
+                        if (curStatus == connectionStatus)
+                        {
+                            return false;
+                        }
+
+                        connectionStatus = curStatus;
+                        break;
+                }
+            } while (true);
+        }
+
+        void EnqueueReceivedPublishPacket(MqttPublishPacket publishPacket)
+        {
+            try
+            {
+                _publishPacketReceiverQueue.Enqueue(publishPacket);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while queueing application message.");
+            }
+        }
+
+        async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
+        {
+            var applicationMessage = _applicationMessageFactory.Create(publishPacket);
+            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
+
+            await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
+            return eventArgs;
+        }
+
+        Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
+        {
+            var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
+            if (extendedAuthenticationExchangeHandler != null)
+            {
+                return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this));
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
+        {
+            _disconnectReason = (MqttClientDisconnectReason)disconnectPacket.ReasonCode;
+            _disconnectReasonString = disconnectPacket.ReasonString;
+
+            // Also dispatch disconnect to waiting threads to generate a proper exception.
+            _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
+
+            return DisconnectInternalAsync(_packetReceiverTask, null, null);
+        }
+
+        async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    _backgroundCancellationTokenSource?.Cancel(false);
+                    var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+                    if (!publishPacketDequeueResult.IsSuccess)
+                    {
+                        return;
+                    }
+
+                    var publishPacket = publishPacketDequeueResult.Item;
+                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+
+                    if (eventArgs.AutoAcknowledge)
+                    {
+                        await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
                 }
                 catch (Exception exception)
                 {
-                    _logger.Warning(exception, "Error while initiating disconnect.");
+                    _logger.Error(exception, "Error while handling application message.");
                 }
             }
         }
 
-        Task SendAsync(MqttPacket packet, CancellationToken cancellationToken)
+        Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
         {
-            cancellationToken.ThrowIfCancellationRequested();
+            if (!_packetDispatcher.TryDispatch(pubRecPacket))
+            {
+                // The packet is unknown. Probably due to a restart of the client.
+                // So wen send this to the server to trigger a full resend of the message.
+                var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
+                return SendAsync(pubRelPacket, cancellationToken);
+            }
 
-            _lastPacketSentTimestamp = DateTime.UtcNow;
+            return PlatformAbstractionLayer.CompletedTask;
+        }
 
-            return _adapter.SendPacketAsync(packet, cancellationToken);
+        Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
+        {
+            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+            return SendAsync(pubCompPacket, cancellationToken);
+        }
+
+        async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
+            var pubAckPacket = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+
+            return _clientPublishResultFactory.Create(pubAckPacket);
+        }
+
+        async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
+            await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
+
+            return _clientPublishResultFactory.Create(null);
+        }
+
+        async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
+            var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+
+            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+
+            var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
+
+            return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
         }
 
         async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket
@@ -508,120 +681,43 @@ namespace MQTTnet.Client
             }
         }
 
-        async Task TrySendKeepAliveMessagesAsync(CancellationToken cancellationToken)
+        Task SendAsync(MqttPacket packet, CancellationToken cancellationToken)
         {
-            try
+            cancellationToken.ThrowIfCancellationRequested();
+
+            _lastPacketSentTimestamp = DateTime.UtcNow;
+
+            return _adapter.SendPacketAsync(packet, cancellationToken);
+        }
+
+        void ThrowIfConnected(string message)
+        {
+            if (IsConnected)
             {
-                _logger.Verbose("Start sending keep alive packets.");
+                throw new MqttProtocolViolationException(message);
-
-                var keepAlivePeriod = Options.KeepAlivePeriod;
-     
-                while (!cancellationToken.IsCancellationRequested)
-                {
-                    // Values described here: [MQTT-3.1.2-24].
-                    var timeWithoutPacketSent = DateTime.UtcNow - _lastPacketSentTimestamp;
-
-                    if (timeWithoutPacketSent > keepAlivePeriod)
-                    {
-                        await PingAsync(cancellationToken).ConfigureAwait(false);
-                    }
-
-                    // Wait a fixed time in all cases. Calculation of the remaining time is complicated
-                    // due to some edge cases and was buggy in the past. Now we wait several ms because the
-                    // min keep alive value is one second so that the server will wait 1.5 seconds for a PING
-                    // packet.
-                    await Task.Delay(250, cancellationToken).ConfigureAwait(false);
-                }
-            }
-            catch (Exception exception)
-            {
-                if (_cleanDisconnectInitiated)
-                {
-                    return;
-                }
-
-                if (exception is OperationCanceledException)
-                {
-                    return;
-                }
-                else if (exception is MqttCommunicationException)
-                {
-                    _logger.Warning(exception, "Communication error while sending/receiving keep alive packets.");
-                }
-                else
-                {
-                    _logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
-                }
-
-                await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
-            }
-            finally
-            {
-                _logger.Verbose("Stopped sending keep alive packets.");
             }
         }
 
-        async Task TryReceivePacketsAsync(CancellationToken cancellationToken)
+        void ThrowIfNotConnected()
         {
-            try
+            if (!IsConnected)
             {
-                _logger.Verbose("Start receiving packets.");
+                throw new MqttCommunicationException("The client is not connected.");
-
-                while (!cancellationToken.IsCancellationRequested)
-                {
-                    MqttPacket packet;
-                    var packetTask = _adapter.ReceivePacketAsync(cancellationToken);
-
-                    if (packetTask.IsCompleted)
-                    {
-                        packet = packetTask.Result;
-                    }
-                    else
-                    {
-                        packet = await packetTask.ConfigureAwait(false);
-                    }
-
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return;
-                    }
-
-                    if (packet == null)
-                    {
-                        await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
-
-                        return;
-                    }
-
-                    await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
-                }
             }
-            catch (Exception exception)
+        }
+
+        void TryInitiateDisconnect()
+        {
+            lock (_disconnectLock)
             {
-                if (_cleanDisconnectInitiated)
+                try
                 {
-                    return;
+                    _backgroundCancellationTokenSource?.Cancel(false);
                 }
-
+                catch (Exception exception)
-                if (exception is OperationCanceledException)
                 {
+                    _logger.Warning(exception, "Error while initiating disconnect.");
                 }
-                else if (exception is MqttCommunicationException)
-                {
-                    _logger.Warning(exception, "Communication error while receiving packets.");
-                }
-                else
-                {
-                    _logger.Error(exception, "Error while receiving packets.");
-                }
-
-                _packetDispatcher.FailAll(exception);
-
-                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-            }
-            finally
-            {
-                _logger.Verbose("Stopped receiving packets.");
             }
         }
 
@@ -690,158 +786,125 @@ namespace MQTTnet.Client
             }
         }
 
-        void EnqueueReceivedPublishPacket(MqttPublishPacket publishPacket)
+        async Task TryReceivePacketsAsync(CancellationToken cancellationToken)
         {
             try
             {
-                _publishPacketReceiverQueue.Enqueue(publishPacket);
+                _logger.Verbose("Start receiving packets.");
-            }
-            catch (Exception exception)
-            {
-                _logger.Error(exception, "Error while queueing application message.");
-            }
-        }
 
-        async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken)
+                while (!cancellationToken.IsCancellationRequested)
-        {
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                try
                 {
-                    var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+                    MqttPacket packet;
-                    if (!publishPacketDequeueResult.IsSuccess)
+                    var packetTask = _adapter.ReceivePacketAsync(cancellationToken);
+
+                    if (packetTask.IsCompleted)
+                    {
+                        packet = packetTask.Result;
+                    }
+                    else
+                    {
+                        packet = await packetTask.ConfigureAwait(false);
+                    }
+
+                    if (cancellationToken.IsCancellationRequested)
                     {
                         return;
                     }
 
-                    var publishPacket = publishPacketDequeueResult.Item;
+                    if (packet == null)
-                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-
-                    if (eventArgs.AutoAcknowledge)
                     {
-                        await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
+                        await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
+
+                        return;
                     }
+
+                    await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
                 }
-                catch (OperationCanceledException)
+            }
+            catch (Exception exception)
+            {
+                if (_cleanDisconnectInitiated)
+                {
+                    return;
+                }
+
+                if (exception is OperationCanceledException)
                 {
                 }
-                catch (Exception exception)
+                else if (exception is MqttCommunicationException)
                 {
-                    _logger.Error(exception, "Error while handling application message.");
+                    _logger.Warning(exception, "Communication error while receiving packets.");
+                }
+                else
+                {
+                    _logger.Error(exception, "Error while receiving packets.");
+                }
+
+                _packetDispatcher.FailAll(exception);
+
+                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
+            }
+            finally
+            {
+                _logger.Verbose("Stopped receiving packets.");
+            }
+        }
+
+        async Task TrySendKeepAliveMessagesAsync(CancellationToken cancellationToken)
+        {
+            try
+            {
+                _logger.Verbose("Start sending keep alive packets.");
+
+                var keepAlivePeriod = Options.KeepAlivePeriod;
+
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    // Values described here: [MQTT-3.1.2-24].
+                    var timeWithoutPacketSent = DateTime.UtcNow - _lastPacketSentTimestamp;
+
+                    if (timeWithoutPacketSent > keepAlivePeriod)
+                    {
+                        using (var timeoutCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
+                        {
+                            timeoutCancellationTokenSource.CancelAfter(Options.Timeout);
+                            await PingAsync(timeoutCancellationTokenSource.Token).ConfigureAwait(false);
+                        }
+                    }
+
+                    // Wait a fixed time in all cases. Calculation of the remaining time is complicated
+                    // due to some edge cases and was buggy in the past. Now we wait several ms because the
+                    // min keep alive value is one second so that the server will wait 1.5 seconds for a PING
+                    // packet.
+                    await Task.Delay(250, cancellationToken).ConfigureAwait(false);
                 }
             }
-        }
+            catch (Exception exception)
-
-        Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken)
-        {
-            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
-                // no response required
+                if (_cleanDisconnectInitiated)
-            }
-            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-            {
-                if (!eventArgs.ProcessingFailed)
                 {
-                    var pubAckPacket = _packetFactories.PubAck.Create(eventArgs);
+                    return;
-                    return SendAsync(pubAckPacket, cancellationToken);
                 }
-            }
+
-            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+                if (exception is OperationCanceledException)
-            {
-                if (!eventArgs.ProcessingFailed)
                 {
-                    var pubRecPacket = _packetFactories.PubRec.Create(eventArgs);
+                    return;
-                    return SendAsync(pubRecPacket, cancellationToken);
                 }
+                else if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning(exception, "Communication error while sending/receiving keep alive packets.");
+                }
+                else
+                {
+                    _logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
+                }
+
+                await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
             }
-            else
+            finally
             {
-                throw new MqttProtocolViolationException("Received a not supported QoS level.");
+                _logger.Verbose("Stopped sending keep alive packets.");
             }
-
-            return PlatformAbstractionLayer.CompletedTask;
-        }
-
-        Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
-        {
-            if (!_packetDispatcher.TryDispatch(pubRecPacket))
-            {
-                // The packet is unknown. Probably due to a restart of the client.
-                // So wen send this to the server to trigger a full resend of the message.
-                var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
-                return SendAsync(pubRelPacket, cancellationToken);
-            }
-
-            return PlatformAbstractionLayer.CompletedTask;
-        }
-
-        Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
-        {
-            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            return SendAsync(pubCompPacket, cancellationToken);
-        }
-
-        Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
-        {
-            _disconnectReason = (MqttClientDisconnectReason) disconnectPacket.ReasonCode;
-            _disconnectReasonString = disconnectPacket.ReasonString;
-            
-            // Also dispatch disconnect to waiting threads to generate a proper exception.
-            _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
-
-            return DisconnectInternalAsync(_packetReceiverTask, null, null);
-        }
-
-        Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
-        {
-            var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
-            if (extendedAuthenticationExchangeHandler != null)
-            {
-                return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this));
-            }
-
-            return PlatformAbstractionLayer.CompletedTask;
-        }
-
-        async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
-        {
-            // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-            await SendAsync(publishPacket, cancellationToken)
-                .ConfigureAwait(false);
-
-            return _clientPublishResultFactory.Create(null);
-        }
-
-        async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
-        {
-            publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-            
-            var pubAckPacket = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
-
-            return _clientPublishResultFactory.Create(pubAckPacket);
-        }
-
-        async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
-        {
-            publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-
-            var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
-
-            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-
-            var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
-
-            return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
-        }
-
-        async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
-        {
-            var applicationMessage = _applicationMessageFactory.Create(publishPacket);
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
-            
-            await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-
-            return eventArgs;
         }
 
         async Task WaitForTaskAsync(Task task, Task sender)
@@ -874,37 +937,5 @@ namespace MQTTnet.Client
             {
             }
         }
-
-        bool DisconnectIsPendingOrFinished()
-        {
-            var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
-            
-            do
-            {
-                switch (connectionStatus)
-                {
-                    case MqttClientConnectionStatus.Disconnected:
-                    case MqttClientConnectionStatus.Disconnecting:
-                        return true;
-                    case MqttClientConnectionStatus.Connected:
-                    case MqttClientConnectionStatus.Connecting:
-                        // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards.
-                        // So the first caller will get a "false" and all subsequent ones will get "true".
-                        var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus);
-                        if (curStatus == connectionStatus)
-                        {
-                            return false;
-                        }
-                        
-                        connectionStatus = curStatus;
-                        break;
-                }
-            } while (true);
-        }
-
-        MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand)
-        {
-            return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
-        }
     }
 }

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -14,6 +14,26 @@ namespace MQTTnet.Client
 {
     public static class MqttClientExtensions
     {
+        public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
+        {
+            if (client == null)
+            {
+                throw new ArgumentNullException(nameof(client));
+            }
+
+            try
+            {
+                await client.PingAsync(cancellationToken).ConfigureAwait(false);
+                return true;
+            }
+            catch
+            {
+                // Ignore errors.
+            }
+
+            return false;
+        }
+        
         public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null)
         {
             if (client == null)

Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs

@@ -24,55 +24,58 @@ namespace MQTTnet.Client
         {
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
-            PublishPacket = publishPacket;
+            PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
-            PacketIdentifier = publishPacket.PacketIdentifier;
             _acknowledgeHandler = acknowledgeHandler;
         }
 
-        internal MqttPublishPacket PublishPacket { get; set; }
+        public MqttApplicationMessage ApplicationMessage { get; }
 
         /// <summary>
-        /// Gets the client identifier.
+        ///     Gets or sets whether the library should send MQTT ACK packets automatically if required.
-        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
+        /// </summary>
+        public bool AutoAcknowledge { get; set; } = true;
+
+        /// <summary>
+        ///     Gets the client identifier.
+        ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
         public string ClientId { get; }
 
-        public MqttApplicationMessage ApplicationMessage { get; }
-
-        public bool ProcessingFailed { get; set; }
-
         /// <summary>
-        /// Gets or sets the reason code which will be sent to the server.
+        ///     Gets or sets whether this message was handled.
-        /// </summary>
+        ///     This value can be used in user code for custom control flow.
-        public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success;
-
-        /// <summary>
-        /// Gets or sets the user properties which will be sent to the server in the ACK packet etc.
-        /// </summary>
-        public List<MqttUserProperty> ResponseUserProperties { get; } = new List<MqttUserProperty>();
-
-        /// <summary>
-        /// Gets or sets the reason string which will be sent to the server in the ACK packet.
-        /// </summary>
-        public string ResponseReasonString { get; set; }
-        
-        /// <summary>
-        /// Gets or sets whether this message was handled.
-        /// This value can be used in user code for custom control flow.
         /// </summary>
         public bool IsHandled { get; set; }
 
         /// <summary>
-        /// Gets or sets whether the library should send MQTT ACK packets automatically if required.
+        ///     Gets the identifier of the MQTT packet
         /// </summary>
-        public bool AutoAcknowledge { get; set; } = true;
+        public ushort PacketIdentifier => PublishPacket.PacketIdentifier;
+
+        /// <summary>
+        ///     Indicates if the processing of this PUBLISH packet has failed.
+        ///     If the processing has failed the client will not send an ACK packet etc.
+        /// </summary>
+        public bool ProcessingFailed { get; set; }
+
+        /// <summary>
+        ///     Gets or sets the reason code which will be sent to the server.
+        /// </summary>
+        public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success;
+
+        /// <summary>
+        ///     Gets or sets the reason string which will be sent to the server in the ACK packet.
+        /// </summary>
+        public string ResponseReasonString { get; set; }
+
+        /// <summary>
+        ///     Gets or sets the user properties which will be sent to the server in the ACK packet etc.
+        /// </summary>
+        public List<MqttUserProperty> ResponseUserProperties { get; } = new List<MqttUserProperty>();
 
         public object Tag { get; set; }
 
-        /// <summary>
+        internal MqttPublishPacket PublishPacket { get; set; }
-        /// Gets the indentifier of MQTT packet
-        /// </summary>
-        public ushort PacketIdentifier { get; }
 
         public Task AcknowledgeAsync(CancellationToken cancellationToken)
         {

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -108,12 +108,12 @@ namespace MQTTnet.Implementations
             return _readStream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
-        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
         {
             // In the write method only the internal buffer will be filled. So here is no
             // async/await required. The real network transmit is done when calling the
             // Flush method.
-            _writeStream.Write(buffer, offset, count);
+            _writeStream.Write(buffer.Array, buffer.Offset, buffer.Count);
             return _writeStream.FlushAsync(cancellationToken);
         }
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -225,7 +225,7 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public async Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
 
@@ -239,12 +239,12 @@ namespace MQTTnet.Implementations
                 }
 
 #if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER
-                await stream.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
+                await stream.WriteAsync(buffer.AsMemory(), cancellationToken).ConfigureAwait(false);
 #else
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
                 using (cancellationToken.Register(_disposeAction))
                 {
-                    await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                    await stream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken).ConfigureAwait(false);
                 }
 #endif
             }

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -134,7 +134,7 @@ namespace MQTTnet.Implementations
                         continue;
                     }
 
-                    Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).RunInBackground(_logger);
+                    _ = Task.Factory.StartNew(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException)
                 {

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -2,15 +2,15 @@
 // 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.Channel;
-using MQTTnet.Internal;
 using System;
 using System.Net;
 using System.Net.WebSockets;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Channel;
 using MQTTnet.Client;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Implementations
 {
@@ -35,12 +35,12 @@ namespace MQTTnet.Implementations
             ClientCertificate = clientCertificate;
         }
 
+        public X509Certificate2 ClientCertificate { get; }
+
         public string Endpoint { get; }
 
         public bool IsSecureConnection { get; private set; }
 
-        public X509Certificate2 ClientCertificate { get; }
-
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
             var uri = _options.Uri;
@@ -89,14 +89,26 @@ namespace MQTTnet.Implementations
             Cleanup();
         }
 
+        public void Dispose()
+        {
+            Cleanup();
+        }
+
         public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
             var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, count), cancellationToken).ConfigureAwait(false);
             return response.Count;
         }
 
-        public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public async Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+            
+#if NET5_0_OR_GREATER
+            // MQTT Control Packets MUST be sent in WebSocket binary data frames. If any other type of data frame is received the recipient MUST close the Network Connection [MQTT-6.0.0-1].
+            // A single WebSocket data frame can contain multiple or partial MQTT Control Packets. The receiver MUST NOT assume that MQTT Control Packets are aligned on WebSocket frame boundaries [MQTT-6.0.0-2].
+            await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false);
+#else
             // The lock is required because the client will throw an exception if _SendAsync_ is 
             // called from multiple threads at the same time. But this issue only happens with several
             // framework versions.
@@ -107,86 +119,9 @@ namespace MQTTnet.Implementations
 
             using (await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false))
             {
-                await _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken).ConfigureAwait(false);
+                await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false);
             }
-        }
-
-        public void Dispose()
-        {
-            Cleanup();
-        }
-
-        void SetupClientWebSocket(ClientWebSocket clientWebSocket)
-        {
-            if (_options.ProxyOptions != null)
-            {
-                clientWebSocket.Options.Proxy = CreateProxy();
-            }
-
-            if (_options.RequestHeaders != null)
-            {
-                foreach (var requestHeader in _options.RequestHeaders)
-                {
-                    clientWebSocket.Options.SetRequestHeader(requestHeader.Key, requestHeader.Value);
-                }
-            }
-
-            if (_options.SubProtocols != null)
-            {
-                foreach (var subProtocol in _options.SubProtocols)
-                {
-                    clientWebSocket.Options.AddSubProtocol(subProtocol);
-                }
-            }
-
-            if (_options.CookieContainer != null)
-            {
-                clientWebSocket.Options.Cookies = _options.CookieContainer;
-            }
-
-            if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null)
-            {
-                clientWebSocket.Options.ClientCertificates = new X509CertificateCollection();
-                foreach (var certificate in _options.TlsOptions.Certificates)
-                {
-#if WINDOWS_UWP
-                    clientWebSocket.Options.ClientCertificates.Add(new X509Certificate(certificate));
-#else
-                    clientWebSocket.Options.ClientCertificates.Add(certificate);
 #endif
-
-                }
-            }
-
-            var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler;
-            if (certificateValidationHandler != null)
-            {
-#if NETSTANDARD1_3
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard1.3'.");
-#elif NETSTANDARD2_0
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard2.0'.");
-#elif WINDOWS_UWP
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'uap10.0'.");
-#elif NET452
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
-#elif NET461
-                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'.");
-#else
-                clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
-                {
-                    // TODO: Find a way to add client options to same callback. Problem is that they have a different type.
-                    var context = new MqttClientCertificateValidationEventArgs
-                    {
-                        Certificate = certificate,
-                        Chain = chain,
-                        SslPolicyErrors = sslPolicyErrors,
-                        ClientOptions = _options
-                    };
-
-                    return certificateValidationHandler(context);
-                };
-#endif
-            }
         }
 
         void Cleanup()
@@ -230,5 +165,77 @@ namespace MQTTnet.Implementations
             return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList);
 #endif
         }
+
+        void SetupClientWebSocket(ClientWebSocket clientWebSocket)
+        {
+            if (_options.ProxyOptions != null)
+            {
+                clientWebSocket.Options.Proxy = CreateProxy();
+            }
+
+            if (_options.RequestHeaders != null)
+            {
+                foreach (var requestHeader in _options.RequestHeaders)
+                {
+                    clientWebSocket.Options.SetRequestHeader(requestHeader.Key, requestHeader.Value);
+                }
+            }
+
+            if (_options.SubProtocols != null)
+            {
+                foreach (var subProtocol in _options.SubProtocols)
+                {
+                    clientWebSocket.Options.AddSubProtocol(subProtocol);
+                }
+            }
+
+            if (_options.CookieContainer != null)
+            {
+                clientWebSocket.Options.Cookies = _options.CookieContainer;
+            }
+
+            if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null)
+            {
+                clientWebSocket.Options.ClientCertificates = new X509CertificateCollection();
+                foreach (var certificate in _options.TlsOptions.Certificates)
+                {
+#if WINDOWS_UWP
+                    clientWebSocket.Options.ClientCertificates.Add(new X509Certificate(certificate));
+#else
+                    clientWebSocket.Options.ClientCertificates.Add(certificate);
+#endif
+                }
+            }
+
+            var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler;
+            if (certificateValidationHandler != null)
+            {
+#if NETSTANDARD1_3
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard1.3'.");
+#elif NETSTANDARD2_0
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard2.0'.");
+#elif WINDOWS_UWP
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'uap10.0'.");
+#elif NET452
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
+#elif NET461
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'.");
+#else
+                clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
+                {
+                    // TODO: Find a way to add client options to same callback. Problem is that they have a different type.
+                    var context = new MqttClientCertificateValidationEventArgs
+                    {
+                        Certificate = certificate,
+                        Chain = chain,
+                        SslPolicyErrors = sslPolicyErrors,
+                        ClientOptions = _options
+                    };
+
+                    return certificateValidationHandler(context);
+                };
+#endif
+            }
+        }
     }
 }

Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs

@@ -11,11 +11,11 @@ namespace MQTTnet.Implementations
     public static class PlatformAbstractionLayer
     {
 #if NET452
-        public static Task CompletedTask => Task.FromResult(0);
+        public static Task CompletedTask { get; } = Task.FromResult(0);
 
         public static byte[] EmptyByteArray { get; } = new byte[0];
 #else
-        public static Task CompletedTask => Task.CompletedTask;
+        public static Task CompletedTask { get; } = Task.CompletedTask;
 
         public static byte[] EmptyByteArray { get; } = Array.Empty<byte>();
 #endif

Source/MQTTnet/Internal/AsyncEvent.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
 
@@ -13,7 +14,16 @@ namespace MQTTnet.Internal
     {
         readonly List<AsyncEventInvocator<TEventArgs>> _handlers = new List<AsyncEventInvocator<TEventArgs>>();
 
-        public bool HasHandlers => _handlers.Count > 0;
+        ICollection<AsyncEventInvocator<TEventArgs>> _handlersForInvoke;
+
+        public AsyncEvent()
+        {
+            _handlersForInvoke = _handlers;
+        }
+
+        // Track the existence of handlers in a separate field so that checking it all the time will not
+        // require locking the actual list (_handlers).
+        public bool HasHandlers { get; private set; }
 
         public void AddHandler(Func<TEventArgs, Task> handler)
         {
@@ -22,7 +32,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
+            lock (_handlers)
+            {
+                _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
+
+                HasHandlers = true;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public void AddHandler(Action<TEventArgs> handler)
@@ -32,12 +48,27 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
+            lock (_handlers)
+            {
+                _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
+
+                HasHandlers = true;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public async Task InvokeAsync(TEventArgs eventArgs)
         {
-            foreach (var handler in _handlers)
+            if (!HasHandlers)
+            {
+                return;
+            }
+
+            // Adding or removing handlers will produce a new list instance all the time.
+            // So locking here is not required since only the reference to an immutable list
+            // of handlers is used.
+            var handlers = _handlersForInvoke;
+            foreach (var handler in handlers)
             {
                 await handler.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
@@ -50,7 +81,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.RemoveAll(h => h.WrapsHandler(handler));
+            lock (_handlers)
+            {
+                _handlers.RemoveAll(h => h.WrapsHandler(handler));
+
+                HasHandlers = _handlers.Count > 0;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public void RemoveHandler(Action<TEventArgs> handler)
@@ -60,7 +97,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.RemoveAll(h => h.WrapsHandler(handler));
+            lock (_handlers)
+            {
+                _handlers.RemoveAll(h => h.WrapsHandler(handler));
+
+                HasHandlers = _handlers.Count > 0;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public async Task TryInvokeAsync(TEventArgs eventArgs, MqttNetSourceLogger logger)

Source/MQTTnet/Internal/AsyncLock.cs

@@ -3,6 +3,8 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -10,76 +12,173 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncLock : IDisposable
     {
-        readonly object _syncRoot = new object();
+        /*
-        readonly Task<IDisposable> _releaser;
+         * This async supporting lock does not support reentrancy!
+         */
 
-        SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
+        readonly List<Releaser> _queuedTasks = new List<Releaser>(64);
+        readonly Task<IDisposable> _releaserTaskWithDirectApproval;
+
+        readonly Releaser _releaserWithDirectApproval;
+
+        readonly object _syncRoot = new object();
+
+        bool _isDisposed;
         
         public AsyncLock()
         {
-            _releaser = Task.FromResult((IDisposable)new Releaser(this));
+            _releaserWithDirectApproval = new Releaser(this, null, CancellationToken.None);
-        }
+            _releaserTaskWithDirectApproval = Task.FromResult((IDisposable)_releaserWithDirectApproval);
-
-        public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
-        {
-            Task task;
-
-            // This lock is required to avoid ObjectDisposedExceptions.
-            // These are fired when this lock gets disposed (and thus the semaphore)
-            // and a worker thread tries to call this method at the same time.
-            // Another way would be catching all ObjectDisposedExceptions but this situation happens
-            // quite often when clients are disconnecting.
-            lock (_syncRoot)
-            {
-                task = _semaphore?.WaitAsync(cancellationToken);
-            }
-
-            if (task == null)
-            {
-                throw new ObjectDisposedException("The AsyncLock is disposed.");
-            }
-
-            if (task.Status == TaskStatus.RanToCompletion)
-            {
-                return _releaser;
-            }
-
-            // Wait for the _WaitAsync_ method and return the releaser afterwards.
-            return task.ContinueWith(
-                (_, state) => (IDisposable)state, 
-                _releaser.Result, 
-                cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
         }
 
         public void Dispose()
         {
             lock (_syncRoot)
             {
-                _semaphore?.Dispose();
+                foreach (var waitingTask in _queuedTasks)
-                _semaphore = null;
+                {
+                    waitingTask.Fail(new ObjectDisposedException(nameof(AsyncLock)));
+                }
+
+                _queuedTasks.Clear();
+                
+                _isDisposed = true;
             }
         }
 
-        internal void Release()
+        public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
+        {
+            var hasDirectApproval = false;
+            Releaser releaser;
+
+            lock (_syncRoot)
+            {
+                if (_isDisposed)
+                {
+                    throw new ObjectDisposedException(nameof(AsyncLock));
+                }
+                
+                if (_queuedTasks.Count == 0)
+                {
+                    // There is no other waiting task apart from the current one.
+                    // So we can approve the current task directly.
+                    releaser = _releaserWithDirectApproval;
+                    hasDirectApproval = true;
+                    Debug.WriteLine("AsyncLock: Task -1 directly approved.");
+                }
+                else
+                {
+                    releaser = new Releaser(this, new TaskCompletionSource<IDisposable>(), cancellationToken);
+                }
+
+                _queuedTasks.Add(releaser);
+            }
+
+            if (!hasDirectApproval)
+            {
+                return releaser.Task;
+            }
+
+            return _releaserTaskWithDirectApproval;
+        }
+
+        void Release(Releaser releaser)
         {
             lock (_syncRoot)
             {
-                _semaphore?.Release();
+                if (_isDisposed)
+                {
+                    // There is no much left to do!
+                    return;
+                }
+
+                var activeTask = _queuedTasks[0];
+                if (!ReferenceEquals(activeTask, releaser))
+                {
+                    throw new InvalidOperationException("The active task must be the current releaser.");
+                }
+
+                _queuedTasks.RemoveAt(0);
+                
+                while (_queuedTasks.Count > 0)
+                {
+                    var nextTask = _queuedTasks[0];
+                    if (!nextTask.IsPending)
+                    {
+                        // Dequeue all canceled or failed tasks.
+                        _queuedTasks.RemoveAt(0);
+                        continue;
+                    }
+
+                    nextTask.Approve();
+                    return;
+                }
+
+                Debug.WriteLine("AsyncLock: No Task pending.");
             }
         }
 
         sealed class Releaser : IDisposable
         {
-            readonly AsyncLock _lock;
+            readonly AsyncLock _asyncLock;
+            readonly CancellationToken _cancellationToken;
+            readonly int _id;
+            readonly TaskCompletionSource<IDisposable> _promise;
 
-            internal Releaser(AsyncLock @lock)
+            // ReSharper disable once FieldCanBeMadeReadOnly.Local
+            CancellationTokenRegistration _cancellationTokenRegistration;
+
+            internal Releaser(AsyncLock asyncLock, TaskCompletionSource<IDisposable> promise, CancellationToken cancellationToken)
             {
-                _lock = @lock;
+                _asyncLock = asyncLock ?? throw new ArgumentNullException(nameof(asyncLock));
+                _promise = promise;
+                _cancellationToken = cancellationToken;
+
+                if (cancellationToken.CanBeCanceled)
+                {
+                    _cancellationTokenRegistration = cancellationToken.Register(Cancel);
+                }
+
+                _id = promise?.Task.Id ?? -1;
+
+                Debug.WriteLine($"AsyncLock: Task {_id} queued.");
+            }
+
+            public bool IsPending => _promise != null && !_promise.Task.IsCanceled && !_promise.Task.IsFaulted && !_promise.Task.IsCompleted;
+
+            public Task<IDisposable> Task => _promise?.Task;
+
+            public void Approve()
+            {
+                _promise?.TrySetResult(this);
+
+                Debug.WriteLine($"AsyncLock: Task {_id} approved.");
             }
 
             public void Dispose()
             {
-                _lock.Release();
+                if (_cancellationToken.CanBeCanceled)
+                {
+                    _cancellationTokenRegistration.Dispose();
+                }
+
+                Debug.WriteLine($"AsyncLock: Task {_id} completed.");
+
+                _asyncLock.Release(this);
+            }
+
+            public void Fail(Exception exception)
+            {
+                _promise?.TrySetException(exception);
+
+                Debug.WriteLine($"AsyncLock: Task {_id} failed ({exception.GetType().Name}).");
+            }
+
+            void Cancel()
+            {
+                _promise?.TrySetCanceled();
+
+                Debug.WriteLine($"AsyncLock: Task {_id} canceled.");
             }
         }
     }

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -98,7 +98,7 @@ namespace MQTTnet.Server
 
                 try
                 {
-                    Task.Run(() => SendPacketsLoop(cancellationToken), cancellationToken).RunInBackground(_logger);
+                    _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
 
                     IsRunning = true;
 
@@ -147,13 +147,20 @@ namespace MQTTnet.Server
         {
             IsRunning = false;
 
-            if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
+            // 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
+                // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
-                // token because the entire connection is closed (disposed) as soon as the cancellation
+                // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
-                // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+                // a configuration option is requested in the future.
-                // at the client!
+                if (reason != MqttDisconnectReasonCode.NormalDisconnection)
-                await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+                {
+                    // 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);
+                }
             }
 
             StopInternal();
@@ -385,6 +392,19 @@ namespace MQTTnet.Server
                         return;
                     }
                     
+                    // Check for cancellation again because receive packet might block some time.
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
+                    // The TCP connection of this client may be still open but the client has already been taken over by
+                    // a new TCP connection. So we must exit here to make sure to no longer process any message.
+                    if (IsTakenOver || !IsRunning)
+                    {
+                        return;
+                    }
+                    
                     var processPacket = true;
 
                     if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
@@ -486,6 +506,11 @@ namespace MQTTnet.Server
                         return;
                     }
 
+                    if (IsTakenOver || !IsRunning)
+                    {
+                        return;
+                    }
+
                     try
                     {
                         await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -24,6 +24,7 @@ namespace MQTTnet.Server
         readonly Dictionary<string, MqttClient> _clients = new Dictionary<string, MqttClient>(4096);
 
         readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
+
         readonly MqttServerEventContainer _eventContainer;
         readonly MqttNetSourceLogger _logger;
         readonly MqttServerOptions _options;
@@ -75,6 +76,8 @@ namespace MQTTnet.Server
 
         public async Task DeleteSessionAsync(string clientId)
         {
+            _logger.Verbose("Deleting session for client '{0}'.", clientId);
+
             MqttClient connection;
 
             lock (_clients)
@@ -193,7 +196,7 @@ namespace MQTTnet.Server
 
         public void Dispose()
         {
-            _createConnectionSyncRoot?.Dispose();
+            _createConnectionSyncRoot.Dispose();
 
             lock (_sessionsManagementLock)
             {
@@ -438,13 +441,18 @@ namespace MQTTnet.Server
             return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
         }
 
+        MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
+        {
+            return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger);
+        }
+
         async Task<MqttClient> CreateClientConnection(
             MqttConnectPacket connectPacket,
             MqttConnAckPacket connAckPacket,
             IMqttChannelAdapter channelAdapter,
             ValidatingConnectionEventArgs validatingConnectionEventArgs)
         {
-            MqttClient connection;
+            MqttClient client;
 
             bool sessionShouldPersist;
 
@@ -484,7 +492,7 @@ namespace MQTTnet.Server
                     {
                         if (connectPacket.CleanSession)
                         {
-                            _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
+                            _logger.Verbose("Deleting existing session of client '{0}' due to clean start.", connectPacket.ClientId);
                             session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
                         }
                         else
@@ -507,40 +515,39 @@ namespace MQTTnet.Server
                     await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false);
                 }
 
-                MqttClient existing;
+                MqttClient existingClient;
 
                 lock (_clients)
                 {
-                    _clients.TryGetValue(connectPacket.ClientId, out existing);
+                    _clients.TryGetValue(connectPacket.ClientId, out existingClient);
-                    connection = CreateConnection(connectPacket, channelAdapter, session);
+                    client = CreateClient(connectPacket, channelAdapter, session);
 
-                    _clients[connectPacket.ClientId] = connection;
+                    _clients[connectPacket.ClientId] = client;
                 }
 
-                if (existing != null)
+                if (existingClient != null)
                 {
-                    existing.IsTakenOver = true;
+                    existingClient.IsTakenOver = true;
-                    await existing.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
+                    await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
 
                     if (_eventContainer.ClientConnectedEvent.HasHandlers)
                     {
-                        var eventArgs = new ClientDisconnectedEventArgs(existing.Id, MqttClientDisconnectType.Takeover, existing.Endpoint, existing.Session.Items);
+                        var eventArgs = new ClientDisconnectedEventArgs(
+                            existingClient.Id,
+                            MqttClientDisconnectType.Takeover,
+                            existingClient.Endpoint,
+                            existingClient.Session.Items);
                         await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
                 }
             }
 
-            return connection;
+            return client;
-        }
-
-        MqttClient CreateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
-        {
-            return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger);
         }
 
         MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent)
         {
-            _logger.Verbose("Created a new session for client '{0}'.", clientId);
+            _logger.Verbose("Created new session for client '{0}'.", clientId);
 
             return new MqttSession(clientId, isPersistent, sessionItems, _options, _eventContainer, _retainedMessagesManager, this);
         }