Add samples.

.github/workflows/ReleaseNotes.md

@@ -1,7 +0,0 @@
-* [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.1.${{github.run_number}}"
+  VERSION: "4.1.0.${{github.run_number}}"
 
 jobs:
   build:

MQTTnet.sln.DotSettings

@@ -210,7 +210,6 @@
         </Or>
       </Entry.Match>
         <Entry.SortBy>
-            <Access Order="Public Internal Protected Private" />
             <Name />
         </Entry.SortBy>
     </Entry>
@@ -241,6 +240,5 @@ 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. It is compatible with mostly any supported .NET Framework version and CPU architecture.
+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.
 
 ## Features
 
@@ -50,6 +50,39 @@ 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.Helpers;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Client;
 
@@ -321,12 +321,14 @@ public static class Client_Connection_Samples
                     {
                         try
                         {
-                            // This code will also do the very first connect! So no call to _ConnectAsync_ is required in the first place.
+                            // This code will also do the very first connect! So no call to _ConnectAsync_ is required
-                            if (!await mqttClient.TryPingAsync())
+                            // in the first place.
+                            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,53 +40,6 @@ 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,8 +7,7 @@
 // ReSharper disable InconsistentNaming
 
 using MQTTnet.Client;
-using MQTTnet.Packets;
+using MQTTnet.Samples.Shared;
-using MQTTnet.Samples.Helpers;
 
 namespace MQTTnet.Samples.Client;
 
@@ -82,48 +81,4 @@ 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,6 +8,7 @@
 
 using System.Text;
 using MQTTnet.Diagnostics;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Diagnostics;
 
@@ -22,9 +23,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(new MyLogger());
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -32,13 +31,31 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
-                .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()
     {
         /*
@@ -64,9 +81,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(mqttEventLogger);
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -74,8 +89,7 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
-                .Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
@@ -83,7 +97,7 @@ public static class Logger_Samples
 
     sealed class MyLogger : IMqttNetLogger
     {
-        public bool IsEnabled { get; set; } = true;
+        public bool IsEnabled { get; } = true;
 
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {

Samples/Program.cs

@@ -8,6 +8,7 @@ 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,50 +170,4 @@ 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

@@ -0,0 +1,53 @@
+// 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);
+            }
+        }
+    }
+}

Samples/Shared/ObjectExtensions.cs

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

Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs

@@ -1,66 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System;
-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,10 +12,14 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="BenchmarkDotNet" Version="0.13.2" />
+        <PackageReference Include="BenchmarkDotNet" Version="0.13.1" />
         <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,7 +2,6 @@
 // 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;
@@ -71,8 +70,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;
             }
         }
 
@@ -80,11 +79,9 @@ namespace MQTTnet.Benchmarks
         {
             await Task.Yield();
 
-            var buffer = new ArraySegment<byte>(new byte[size]);
-            
             for (var i = 0; i < iterations; i++)
             {
-                await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false);
+                await _serverChannel.WriteAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false);
             }
         }
     }

Source/MQTTnet.Benchmarks/Program.cs

@@ -29,7 +29,6 @@ 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)
@@ -76,9 +75,6 @@ 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(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
+            public Task WriteAsync(byte[] buffer, int offset, int count, 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(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
+        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            _webSocket.Send(buffer.Array, buffer.Offset, buffer.Count);
+            _webSocket.Send(buffer, offset, count);
             return Task.FromResult(0);
         }
 

Source/MQTTnet.TestApp/AsyncLockTest.cs

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

Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs

@@ -41,33 +41,6 @@ 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);
@@ -84,8 +57,6 @@ 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,7 +3,6 @@
 // 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;
@@ -14,80 +13,6 @@ 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()
         {
@@ -120,32 +45,58 @@ 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 async Task Test_Cancellation_With_Later_Access()
+        public void Lock_10_Parallel_Tasks()
         {
-            var asyncLock = new AsyncLock();
+            const int ThreadsCount = 10;
 
-            var releaser = await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false);
+            var threads = new Task[ThreadsCount];
-
+            var @lock = new AsyncLock();
-            try
+            var globalI = 0;
+            for (var i = 0; i < ThreadsCount; i++)
             {
-                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
+#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 () =>
                 {
-                    await asyncLock.WaitAsync(timeout.Token).ConfigureAwait(false);
+                    using (var releaser = await @lock.WaitAsync(CancellationToken.None))
-                }
+                    {
-                
+                        var localI = globalI;
-                Assert.Fail("Exception should be thrown!");
+                        await Task.Delay(10); // Increase the chance for wrong data.
-            }
+                        localI++;
-            catch (OperationCanceledException)
+                        globalI = localI;
-            {
+                    }
+                });
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
             }
 
-            releaser.Dispose();
+            Task.WaitAll(threads);
-
+            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,49 +175,6 @@ 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,7 +2,6 @@
 // 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;
@@ -46,9 +45,9 @@ namespace MQTTnet.Tests.Mockups
             return _stream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
-        public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
+        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            return _stream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken);
+            return _stream.WriteAsync(buffer, offset, count, cancellationToken);
         }
 
         public void Dispose()

Source/MQTTnet.Tests/Server/General.cs

@@ -935,62 +935,7 @@ namespace MQTTnet.Tests.Server
                 Assert.IsTrue(client.IsConnected);
             }
         }
-
+        
-        [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,9 +124,7 @@ namespace MQTTnet.Tests.Server
         }
 
         [TestMethod]
-        [DataRow(MqttProtocolVersion.V311)]
+        public async Task Handle_Parallel_Connection_Attempts()
-        [DataRow(MqttProtocolVersion.V500)]
-        public async Task Handle_Parallel_Connection_Attempts(MqttProtocolVersion protocolVersion)
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
@@ -134,7 +132,7 @@ namespace MQTTnet.Tests.Server
 
                 await testEnvironment.StartServer();
 
-                var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion).WithKeepAlivePeriod(TimeSpan.FromSeconds(5));
+                var options = new MqttClientOptionsBuilder().WithClientId("1").WithKeepAlivePeriod(TimeSpan.FromSeconds(5));
 
                 var hasReceive = false;
 
@@ -148,10 +146,7 @@ namespace MQTTnet.Tests.Server
 
                 // Try to connect 50 clients at the same time.
                 var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
-
+                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);
 
@@ -166,7 +161,26 @@ namespace MQTTnet.Tests.Server
                 Assert.AreEqual(true, hasReceive);
             }
         }
-        
+
+        [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)]
@@ -346,7 +360,7 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        static async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
+        async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
         {
             try
             {
@@ -365,13 +379,13 @@ namespace MQTTnet.Tests.Server
 
                 return sendClient;
             }
-            catch
+            catch (Exception)
             {
                 return null;
             }
         }
 
-        static async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
+        async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
         {
             try
             {

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -188,14 +188,11 @@ 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.Packet, false, cancellationToken).ConfigureAwait(false);
+                        await _channel.WriteAsync(packetBuffer.Payload.Array, packetBuffer.Payload.Offset, packetBuffer.Payload.Count, 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,17 +12,13 @@ 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,31 +7,32 @@ 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; }
-
+        
         MqttClientOptions Options { get; }
-
+        
         Task<MqttClientConnectResult> ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default);
-
+        
         Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default);
-
+        
         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/MqttClientExtensions.cs

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

Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs

@@ -13,83 +13,80 @@ namespace MQTTnet.Client
     public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
     {
         readonly Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> _acknowledgeHandler;
-
+        
         int _isAcknowledged;
-
+        
         public MqttApplicationMessageReceivedEventArgs(
-            string clientId,
+            string clientId, 
             MqttApplicationMessage applicationMessage,
             MqttPublishPacket publishPacket,
             Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> acknowledgeHandler)
         {
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
-            PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
+            PublishPacket = publishPacket;
+            PacketIdentifier = publishPacket.PacketIdentifier;
             _acknowledgeHandler = acknowledgeHandler;
         }
 
-        public MqttApplicationMessage ApplicationMessage { get; }
+        internal MqttPublishPacket PublishPacket { get; set; }
-
+        
         /// <summary>
-        ///     Gets or sets whether the library should send MQTT ACK packets automatically if required.
+        /// Gets the client identifier.
-        /// </summary>
+        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
-        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; }
 
-        /// <summary>
+        public MqttApplicationMessage ApplicationMessage { get; }
-        ///     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 the identifier of the MQTT packet
-        /// </summary>
-        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.
+        /// 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.
+        /// Gets or sets the user properties which will be sent to the server in the ACK packet etc.
-        /// </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>();
 
+        /// <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.
+        /// </summary>
+        public bool AutoAcknowledge { get; set; } = true;
+        
         public object Tag { get; set; }
 
-        internal MqttPublishPacket PublishPacket { get; set; }
+        /// <summary>
+        /// Gets the indentifier of MQTT packet
+        /// </summary>
+        public ushort PacketIdentifier { get; }
 
-        public Task AcknowledgeAsync(CancellationToken cancellationToken)
+        public Task AcknowledgeAsync(CancellationToken cancellationToken) 
         {
             if (_acknowledgeHandler == null)
             {
                 throw new NotSupportedException("Deferred acknowledgement of application message is not yet supported in MQTTnet server.");
             }
-
+            
             if (Interlocked.CompareExchange(ref _isAcknowledged, 1, 0) == 0)
             {
                 return _acknowledgeHandler(this, cancellationToken);
             }
-
+            
             throw new InvalidOperationException("The application message is already acknowledged.");
         }
     }
-}
+}

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -108,12 +108,12 @@ namespace MQTTnet.Implementations
             return _readStream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
-        public Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
+        public Task WriteAsync(byte[] buffer, int offset, int count, 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.Array, buffer.Offset, buffer.Count);
+            _writeStream.Write(buffer, offset, count);
             return _writeStream.FlushAsync(cancellationToken);
         }
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

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

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -134,7 +134,7 @@ namespace MQTTnet.Implementations
                         continue;
                     }
 
-                    _ = Task.Factory.StartNew(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
+                    Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).RunInBackground(_logger);
                 }
                 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,26 +89,14 @@ 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(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
+        public async Task WriteAsync(byte[] buffer, int offset, int count, 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.
@@ -119,9 +107,86 @@ namespace MQTTnet.Implementations
 
             using (await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false))
             {
-                await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false);
+                await _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, 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()
@@ -165,77 +230,5 @@ 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 { get; } = Task.FromResult(0);
+        public static Task CompletedTask => Task.FromResult(0);
 
         public static byte[] EmptyByteArray { get; } = new byte[0];
 #else
-        public static Task CompletedTask { get; } = Task.CompletedTask;
+        public static Task CompletedTask => Task.CompletedTask;
 
         public static byte[] EmptyByteArray { get; } = Array.Empty<byte>();
 #endif

Source/MQTTnet/Internal/AsyncEvent.cs

@@ -4,7 +4,6 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
 
@@ -14,16 +13,7 @@ namespace MQTTnet.Internal
     {
         readonly List<AsyncEventInvocator<TEventArgs>> _handlers = new List<AsyncEventInvocator<TEventArgs>>();
 
-        ICollection<AsyncEventInvocator<TEventArgs>> _handlersForInvoke;
+        public bool HasHandlers => _handlers.Count > 0;
-
-        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)
         {
@@ -32,13 +22,7 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            lock (_handlers)
+            _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
-            {
-                _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
-
-                HasHandlers = true;
-                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
-            }
         }
 
         public void AddHandler(Action<TEventArgs> handler)
@@ -48,32 +32,17 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            lock (_handlers)
+            _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
-            {
-                _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
-
-                HasHandlers = true;
-                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
-            }
         }
 
         public async Task InvokeAsync(TEventArgs eventArgs)
         {
-            if (!HasHandlers)
+            foreach (var handler in _handlers)
-            {
-                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);
             }
         }
-
+        
         public void RemoveHandler(Func<TEventArgs, Task> handler)
         {
             if (handler == null)
@@ -81,13 +50,7 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            lock (_handlers)
+            _handlers.RemoveAll(h => h.WrapsHandler(handler));
-            {
-                _handlers.RemoveAll(h => h.WrapsHandler(handler));
-
-                HasHandlers = _handlers.Count > 0;
-                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
-            }
         }
 
         public void RemoveHandler(Action<TEventArgs> handler)
@@ -97,13 +60,7 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            lock (_handlers)
+            _handlers.RemoveAll(h => h.WrapsHandler(handler));
-            {
-                _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,8 +3,6 @@
 // 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;
 
@@ -12,174 +10,77 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncLock : IDisposable
     {
-        /*
-         * This async supporting lock does not support reentrancy!
-         */
-
-        readonly List<Releaser> _queuedTasks = new List<Releaser>(64);
-        readonly Task<IDisposable> _releaserTaskWithDirectApproval;
-
-        readonly Releaser _releaserWithDirectApproval;
-
         readonly object _syncRoot = new object();
+        readonly Task<IDisposable> _releaser;
+
+        SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
-        bool _isDisposed;
-        
         public AsyncLock()
         {
-            _releaserWithDirectApproval = new Releaser(this, null, CancellationToken.None);
+            _releaser = Task.FromResult((IDisposable)new Releaser(this));
-            _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)
             {
-                foreach (var waitingTask in _queuedTasks)
+                _semaphore?.Dispose();
-                {
+                _semaphore = null;
-                    waitingTask.Fail(new ObjectDisposedException(nameof(AsyncLock)));
-                }
-
-                _queuedTasks.Clear();
-                
-                _isDisposed = true;
             }
         }
 
-        public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
+        internal void Release()
-        {
-            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)
             {
-                if (_isDisposed)
+                _semaphore?.Release();
-                {
-                    // 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 _asyncLock;
+            readonly AsyncLock _lock;
-            readonly CancellationToken _cancellationToken;
-            readonly int _id;
-            readonly TaskCompletionSource<IDisposable> _promise;
 
-            // ReSharper disable once FieldCanBeMadeReadOnly.Local
+            internal Releaser(AsyncLock @lock)
-            CancellationTokenRegistration _cancellationTokenRegistration;
-
-            internal Releaser(AsyncLock asyncLock, TaskCompletionSource<IDisposable> promise, CancellationToken cancellationToken)
             {
-                _asyncLock = asyncLock ?? throw new ArgumentNullException(nameof(asyncLock));
+                _lock = @lock;
-                _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()
             {
-                if (_cancellationToken.CanBeCanceled)
+                _lock.Release();
-                {
-                    _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.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
+                    Task.Run(() => SendPacketsLoop(cancellationToken), cancellationToken).RunInBackground(_logger);
 
                     IsRunning = true;
 
@@ -147,20 +147,13 @@ namespace MQTTnet.Server
         {
             IsRunning = false;
 
-            // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
+            if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
-            if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
             {
-                // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
+                // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
-                // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
+                // token because the entire connection is closed (disposed) as soon as the cancellation
-                // a configuration option is requested in the future.
+                // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
-                if (reason != MqttDisconnectReasonCode.NormalDisconnection)
+                // at the client!
-                {
+                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();
@@ -391,20 +384,7 @@ 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)
@@ -506,11 +486,6 @@ namespace MQTTnet.Server
                         return;
                     }
 
-                    if (IsTakenOver || !IsRunning)
-                    {
-                        return;
-                    }
-
                     try
                     {
                         await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

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