Improve PingAsync timeout handling.

.github/workflows/ReleaseNotes.md

@@ -1,4 +1,5 @@
 * [Core] MQTT Packets being sent over web socket transport are now setting the web socket frame boundaries correctly (#1499).
 * [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.

Samples/Client/Client_Connection_Samples.cs

@@ -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.");
                             }
                         }

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/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,26 +86,29 @@ 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);
             remove => _inspectPacketEvent.RemoveHandler(value);
         }
 
+        [Obsolete(
+            "This property will be removed in the future. Checking for a working connection should be done via calling _PinAsync_. Also other successful traffic can be used to indicate connection status.")]
         public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
         public MqttClientOptions Options { get; private set; }
 
         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 +194,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 +224,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 +271,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 +344,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 +402,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 +483,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,124 +683,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)
-                    {
-                        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)
-            {
-                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.");
             }
         }
 
@@ -694,158 +788,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)
@@ -878,37 +939,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)