Move code into dedicated components

Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs

@@ -30,7 +30,7 @@ namespace MQTTnet.Tests.Diagnostics
 
                     var packets = new List<string>();
                     
-                    mqttClient.InspectPackage += eventArgs =>
+                    mqttClient.InspectPackageAsync += eventArgs =>
                     {
                         packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer));
                         return CompletedTask.Instance;

Source/MQTTnet.Tests/Server/General.cs

@@ -954,7 +954,7 @@ namespace MQTTnet.Tests.Server
 
                 var client = testEnvironment.CreateClient();
 
-                client.InspectPackage += e =>
+                client.InspectPackageAsync += e =>
                 {
                     if (e.Buffer.Length > 0)
                     {

Source/MQTTnet/Client/ExtendedAuthenticationExchange/IMqttExtendedAuthenticationExchangeHandler.cs

@@ -8,6 +8,6 @@ namespace MQTTnet.Client
 {
     public interface IMqttExtendedAuthenticationExchangeHandler
     {
-        Task HandleRequestAsync(MqttExtendedAuthenticationExchangeContext context);
+        Task HandleRequestAsync(MqttExtendedAuthenticationExchangeEventArgs eventArgs);
     }
 }

Source/MQTTnet/Client/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeContext.cs

@@ -1,62 +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.Collections.Generic;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.Client
-{
-    public class MqttExtendedAuthenticationExchangeContext
-    {
-        public MqttExtendedAuthenticationExchangeContext(MqttAuthPacket authPacket, MqttClient client)
-        {
-            if (authPacket == null) throw new ArgumentNullException(nameof(authPacket));
-
-            ReasonCode = authPacket.ReasonCode;
-            ReasonString = authPacket.ReasonString;
-            AuthenticationMethod = authPacket.AuthenticationMethod;
-            AuthenticationData = authPacket.AuthenticationData;
-            UserProperties = authPacket.UserProperties;
-
-            Client = client ?? throw new ArgumentNullException(nameof(client));
-        }
-
-        /// <summary>
-        /// Gets the reason code.
-        /// Hint: MQTT 5 feature only.
-        /// </summary>
-        public MqttAuthenticateReasonCode ReasonCode { get; }
-
-        /// <summary>
-        /// Gets the reason string.
-        /// Hint: MQTT 5 feature only.
-        /// </summary>
-        public string ReasonString { get; }
-
-        /// <summary>
-        /// Gets the authentication method.
-        /// Hint: MQTT 5 feature only.
-        /// </summary>
-        public string AuthenticationMethod { get; }
-
-        /// <summary>
-        /// Gets the authentication data.
-        /// Hint: MQTT 5 feature only.
-        /// </summary>
-        public byte[] AuthenticationData { get; }
-
-        /// <summary>
-        /// Gets the user properties.
-        /// In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT packet.
-        /// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add metadata to MQTT messages and pass information between publisher, broker, and subscriber.
-        /// The feature is very similar to the HTTP header concept.
-        /// Hint: MQTT 5 feature only.
-        /// </summary>
-        public List<MqttUserProperty> UserProperties { get; }
-
-        public MqttClient Client { get; }
-    }
-}

Source/MQTTnet/Client/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeData.cs

@@ -39,4 +39,4 @@ namespace MQTTnet.Client
         /// </summary>
         public List<MqttUserProperty> UserProperties { get; }
     }
-}
+}

Source/MQTTnet/Client/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeEventArgs.cs

@@ -0,0 +1,65 @@
+// 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.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttExtendedAuthenticationExchangeEventArgs : EventArgs
+    {
+        static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>();
+
+        public MqttExtendedAuthenticationExchangeEventArgs(MqttAuthPacket authPacket)
+        {
+            if (authPacket == null)
+            {
+                throw new ArgumentNullException(nameof(authPacket));
+            }
+
+            ReasonCode = authPacket.ReasonCode;
+            ReasonString = authPacket.ReasonString;
+            AuthenticationMethod = authPacket.AuthenticationMethod;
+            AuthenticationData = authPacket.AuthenticationData;
+            UserProperties = authPacket.UserProperties ?? EmptyUserProperties;
+        }
+
+        /// <summary>
+        ///     Gets the authentication data.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public byte[] AuthenticationData { get; }
+
+        /// <summary>
+        ///     Gets the authentication method.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public string AuthenticationMethod { get; }
+
+        /// <summary>
+        ///     Gets the reason code.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public MqttAuthenticateReasonCode ReasonCode { get; }
+
+        /// <summary>
+        ///     Gets the reason string.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public string ReasonString { get; }
+
+        /// <summary>
+        ///     Gets or sets the user properties.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
+
+        /// <summary>
+        /// Gets the response which will be sent to the server.
+        /// </summary>
+        public MqttExtendedAuthenticationExchangeResponse Response { get; } = new MqttExtendedAuthenticationExchangeResponse();
+    }
+}

Source/MQTTnet/Client/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeResponse.cs

@@ -0,0 +1,24 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttExtendedAuthenticationExchangeResponse
+    {
+        /// <summary>
+        ///     Gets or sets the authentication data.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public byte[] AuthenticationData { get; set; }
+
+        /// <summary>
+        ///     Gets or sets the user properties which will be sent to the server.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
+    }
+}

Source/MQTTnet/Client/IMqttClient.cs

@@ -8,14 +8,14 @@ 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<InspectMqttPacketEventArgs, Task> InspectPackage;
+        event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
 
         bool IsConnected { get; }
 
@@ -27,10 +27,10 @@ namespace MQTTnet.Client
 
         Task PingAsync(CancellationToken cancellationToken = default);
 
-        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
-
         Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default);
 
+        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
+
         Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default);
 
         Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default);

Source/MQTTnet/Client/Internal/MqttClientAuthenticationHandler.cs

@@ -0,0 +1,109 @@
+// 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 MQTTnet.Adapter;
+using MQTTnet.Diagnostics;
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using MQTTnet.Internal;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Client.Internal
+{
+    public sealed class MqttClientAuthenticationHandler
+    {
+        readonly MqttClient _client;
+        readonly MqttNetSourceLogger _logger;
+        readonly MqttClientConnectResultFactory _clientConnectResultFactory = new MqttClientConnectResultFactory();
+
+        public MqttClientAuthenticationHandler(MqttClient client, IMqttNetLogger logger)
+        {
+            if (logger == null)
+            {
+                throw new ArgumentNullException(nameof(logger));
+            }
+
+            _client = client ?? throw new ArgumentNullException(nameof(client));
+            _logger = logger.WithSource(nameof(MqttClientAuthenticationHandler));
+        }
+
+        public AsyncEvent<MqttExtendedAuthenticationExchangeEventArgs> ExtendedAuthenticationExchangeEvent { get; } = new AsyncEvent<MqttExtendedAuthenticationExchangeEventArgs>();
+
+        public async Task<MqttClientConnectResult> Authenticate(IMqttChannelAdapter channelAdapter, MqttClientOptions options, CancellationToken cancellationToken)
+        {
+            // From the RFC:
+            // The CONNACK packet is the packet sent by the Server in response to a CONNECT packet received from a Client.
+            // The Server MUST send a CONNACK with a 0x00 (Success) Reason Code before sending any Packet other than AUTH [MQTT-3.2.0-1].
+            // The Server MUST NOT send more than one CONNACK in a Network Connection [MQTT-3.2.0-2].
+
+            MqttClientConnectResult result;
+
+            try
+            {
+                // Send the CONNECT packet..
+                var connectPacket = MqttPacketFactories.Connect.Create(options);
+                await _client.SendAsync(connectPacket, cancellationToken).ConfigureAwait(false);
+
+                // Wait for the CONNACK packet...
+                MqttConnAckPacket connAckPacket = null;
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var receivedPacket = await channelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
+
+                    if (receivedPacket is MqttAuthPacket authPacket)
+                    {
+                        // MQTT v3.1.1 cannot send an AUTH packet.
+                        // If the Server requires additional information to complete the authentication, it can send an AUTH packet to the Client.
+                        // This packet MUST contain a Reason Code of 0x18 (Continue authentication)
+                        if (authPacket.ReasonCode != MqttAuthenticateReasonCode.ContinueAuthentication)
+                        {
+                            throw new MqttProtocolViolationException("Wrong reason code received [MQTT-4.12.0-2].");
+                        }
+
+                        var response = await OnExtendedAuthentication(authPacket).ConfigureAwait(false);
+                        authPacket = MqttPacketFactories.Auth.Create(authPacket, response);
+                        await _client.SendAsync(authPacket, cancellationToken).ConfigureAwait(false);
+                    }
+                    else if (receivedPacket is MqttConnAckPacket connAckPacketBuffer)
+                    {
+                        connAckPacket = connAckPacketBuffer;
+                        break;
+                    }
+                }
+
+                result = _clientConnectResultFactory.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
+            }
+            catch (Exception exception)
+            {
+                throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
+            }
+
+            // TODO: Remove because it is wrong.
+            if (result.ResultCode != MqttClientConnectResultCode.Success)
+            {
+                throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
+            }
+
+            _logger.Verbose("Authenticated MQTT connection with server established.");
+
+            return result;
+        }
+
+        async Task<MqttExtendedAuthenticationExchangeResponse> OnExtendedAuthentication(MqttAuthPacket authPacket)
+        {
+            if (!ExtendedAuthenticationExchangeEvent.HasHandlers)
+            {
+                throw new InvalidOperationException("Cannot handle extended authentication without attached event handler.");
+            }
+
+            var eventArgs = new MqttExtendedAuthenticationExchangeEventArgs(authPacket);
+            await ExtendedAuthenticationExchangeEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+            return eventArgs.Response;
+        }
+    }
+}

Source/MQTTnet/Client/Internal/MqttClientKeepAliveHandler.cs

@@ -0,0 +1,126 @@
+// 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 MQTTnet.Diagnostics;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Client.Internal
+{
+    public sealed class MqttClientKeepAliveHandler
+    {
+        readonly MqttClient _client;
+        readonly MqttNetSourceLogger _logger;
+        CancellationTokenSource _cancellationToken;
+        TimeSpan _keepAliveInterval;
+        Task _keepAlivePacketsSenderTask;
+
+        DateTime _lastPacketSentTimestamp;
+
+        public MqttClientKeepAliveHandler(MqttClient client, IMqttNetLogger logger)
+        {
+            _client = client ?? throw new ArgumentNullException(nameof(client));
+            _logger = logger.WithSource(nameof(MqttClientKeepAliveHandler));
+        }
+
+        public void Disable()
+        {
+            try
+            {
+                _cancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _cancellationToken?.Dispose();
+                _cancellationToken = null;
+            }
+        }
+
+        public void Enable(MqttClientConnectResult connectResult)
+        {
+            if (connectResult == null)
+            {
+                throw new ArgumentNullException(nameof(connectResult));
+            }
+
+            _keepAliveInterval = _client.Options.KeepAlivePeriod;
+
+            if (connectResult.ServerKeepAlive > 0)
+            {
+                _logger.Info("Using keep alive value {0} sent from the server", connectResult.ServerKeepAlive);
+                _keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
+            }
+
+            if (_keepAliveInterval == TimeSpan.Zero)
+            {
+                return;
+            }
+
+            _cancellationToken = new CancellationTokenSource();
+            _keepAlivePacketsSenderTask = Task.Run(() => SendKeepAliveMessagesLoop(_cancellationToken.Token), _cancellationToken.Token);
+        }
+
+        public void TrackSentPacket()
+        {
+            // Only use UTC to avoid performance hit by loading current time zone all the time.
+            _lastPacketSentTimestamp = DateTime.UtcNow;
+        }
+
+        async Task SendKeepAliveMessagesLoop(CancellationToken cancellationToken)
+        {
+            try
+            {
+                _logger.Verbose("Start sending keep alive packets.");
+
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    // Values described here: [MQTT-3.1.2-24].
+                    var timeWithoutPacketSent = DateTime.UtcNow - _lastPacketSentTimestamp;
+
+                    if (timeWithoutPacketSent > _keepAliveInterval)
+                    {
+                        using (var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
+                        {
+                            timeout.CancelAfter(_client.Options.Timeout);
+                            await _client.PingAsync(timeout.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 _client.DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
+            }
+            finally
+            {
+                _logger.Verbose("Stopped sending keep alive packets");
+            }
+        }
+    }
+}

Source/MQTTnet/Client/MqttClient.cs

@@ -4,10 +4,10 @@
 
 using System;
 using System.Collections.Generic;
-using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
+using MQTTnet.Client.Internal;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
@@ -23,6 +23,8 @@ namespace MQTTnet.Client
         readonly IMqttClientAdapterFactory _adapterFactory;
         readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
 
+        readonly MqttClientAuthenticationHandler _authenticationHandler;
+
         readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
         readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory();
         readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory();
@@ -30,8 +32,10 @@ namespace MQTTnet.Client
         readonly AsyncEvent<MqttClientConnectedEventArgs> _connectedEvent = new AsyncEvent<MqttClientConnectedEventArgs>();
         readonly AsyncEvent<MqttClientConnectingEventArgs> _connectingEvent = new AsyncEvent<MqttClientConnectingEventArgs>();
         readonly AsyncEvent<MqttClientDisconnectedEventArgs> _disconnectedEvent = new AsyncEvent<MqttClientDisconnectedEventArgs>();
+
         readonly object _disconnectLock = new object();
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
+        readonly MqttClientKeepAliveHandler _keepAliveHandler;
         readonly MqttNetSourceLogger _logger;
 
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
@@ -40,6 +44,8 @@ namespace MQTTnet.Client
         IMqttChannelAdapter _adapter;
 
         bool _cleanDisconnectInitiated;
+
+        CancellationTokenSource _clientAlive;
         volatile int _connectionStatus;
 
         // The value for this field can be set from two different enums.
@@ -47,11 +53,6 @@ namespace MQTTnet.Client
         int _disconnectReason;
         string _disconnectReasonString;
         List<MqttUserProperty> _disconnectUserProperties;
-
-        Task _keepAlivePacketsSenderTask;
-        DateTime _lastPacketSentTimestamp;
-
-        CancellationTokenSource _mqttClientAlive;
         MqttPacketDispatcher _packetDispatcher;
         Task _packetReceiverTask;
         AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
@@ -62,6 +63,9 @@ namespace MQTTnet.Client
             _adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
             _rootLogger = logger ?? throw new ArgumentNullException(nameof(logger));
             _logger = logger.WithSource(nameof(MqttClient));
+
+            _authenticationHandler = new MqttClientAuthenticationHandler(this, logger);
+            _keepAliveHandler = new MqttClientKeepAliveHandler(this, logger);
         }
 
         public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
@@ -88,7 +92,13 @@ namespace MQTTnet.Client
             remove => _disconnectedEvent.RemoveHandler(value);
         }
 
-        public event Func<InspectMqttPacketEventArgs, Task> InspectPackage
+        public event Func<MqttExtendedAuthenticationExchangeEventArgs, Task> ExtendedAuthenticationExchangeAsync
+        {
+            add => _authenticationHandler.ExtendedAuthenticationExchangeEvent.AddHandler(value);
+            remove => _authenticationHandler.ExtendedAuthenticationExchangeEvent.RemoveHandler(value);
+        }
+
+        public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
         {
             add => _inspectPacketEvent.AddHandler(value);
             remove => _inspectPacketEvent.RemoveHandler(value);
@@ -124,9 +134,7 @@ namespace MQTTnet.Client
 
                 _packetIdentifierProvider.Reset();
                 _packetDispatcher = new MqttPacketDispatcher();
-
-                _mqttClientAlive = new CancellationTokenSource();
-                var mqttClientAliveToken = _mqttClientAlive.Token;
+                _clientAlive = new CancellationTokenSource();
 
                 var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_inspectPacketEvent, _rootLogger), _rootLogger);
                 _adapter = adapter;
@@ -145,29 +153,15 @@ namespace MQTTnet.Client
                     }
                 }
 
-                _lastPacketSentTimestamp = DateTime.UtcNow;
-
-                var keepAliveInterval = Options.KeepAlivePeriod;
-                if (connectResult.ServerKeepAlive > 0)
+                if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
                 {
-                    _logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server.");
-                    keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
-                }
-
-                if (keepAliveInterval != TimeSpan.Zero)
-                {
-                    _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(mqttClientAliveToken), mqttClientAliveToken);
+                    await DisconnectInternalAsync(null, null, connectResult).ConfigureAwait(false);
+                    return connectResult;
                 }
 
                 CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
 
-                _logger.Info("Connected.");
-
-                if (_connectedEvent.HasHandlers)
-                {
-                    var eventArgs = new MqttClientConnectedEventArgs(connectResult);
-                    await _connectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-                }
+                await OnConnected(connectResult).ConfigureAwait(false);
 
                 return connectResult;
             }
@@ -388,6 +382,27 @@ namespace MQTTnet.Client
             return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket);
         }
 
+        internal Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
+        {
+            var clientWasConnected = IsConnected;
+
+            if (!DisconnectIsPendingOrFinished())
+            {
+                return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected);
+            }
+
+            return CompletedTask.Instance;
+        }
+
+        internal Task SendAsync(MqttPacket packet, CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            _keepAliveHandler.TrackSentPacket();
+
+            return _adapter.SendPacketAsync(packet, cancellationToken);
+        }
+
         protected override void Dispose(bool disposing)
         {
             if (disposing)
@@ -428,65 +443,16 @@ namespace MQTTnet.Client
             return CompletedTask.Instance;
         }
 
-        async Task<MqttClientConnectResult> AuthenticateAsync(MqttClientOptions options, CancellationToken cancellationToken)
-        {
-            // From the RFC:
-            // The CONNACK packet is the packet sent by the Server in response to a CONNECT packet received from a Client.
-            // The Server MUST send a CONNACK with a 0x00 (Success) Reason Code before sending any Packet other than AUTH [MQTT-3.2.0-1].
-            // The Server MUST NOT send more than one CONNACK in a Network Connection [MQTT-3.2.0-2].
-            
-            MqttClientConnectResult result;
-
-            try
-            {
-                var connectPacket = MqttPacketFactories.Connect.Create(options);
-                await SendAsync(connectPacket, cancellationToken).ConfigureAwait(false);
-                
-                MqttConnAckPacket connAckPacket = null;
-                while (!cancellationToken.IsCancellationRequested)
-                {
-                    var receivedPacket = await _adapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
-
-                    if (receivedPacket is MqttAuthPacket authPacket)
-                    {
-                        // MQTT v3.1.1 cannot send an AUTH packet.
-                        Options.ExtendedAuthenticationExchangeHandler
-                    }
-                    else if (receivedPacket is MqttConnAckPacket connAckPacketBuffer)
-                    {
-                        connAckPacket = connAckPacketBuffer;
-                        break;
-                    }
-                }
-                
-                var clientConnectResultFactory = new MqttClientConnectResultFactory();
-                result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
-            }
-            catch (Exception exception)
-            {
-                throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
-            }
-
-            if (result.ResultCode != MqttClientConnectResultCode.Success)
-            {
-                throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
-            }
-
-            _logger.Verbose("Authenticated MQTT connection with server established.");
-
-            return result;
-        }
-
         void Cleanup()
         {
             try
             {
-                _mqttClientAlive?.Cancel(false);
+                _clientAlive?.Cancel(false);
             }
             finally
             {
-                _mqttClientAlive?.Dispose();
-                _mqttClientAlive = null;
+                _clientAlive?.Dispose();
+                _clientAlive = null;
 
                 _publishPacketReceiverQueue?.Dispose();
                 _publishPacketReceiverQueue = null;
@@ -506,23 +472,27 @@ namespace MQTTnet.Client
 
         async Task<MqttClientConnectResult> ConnectInternal(CancellationToken cancellationToken)
         {
-            var backgroundCancellationToken = _mqttClientAlive.Token;
+            var clientAliveToken = _clientAlive.Token;
 
-            using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
+            using (var effectiveCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(clientAliveToken, cancellationToken))
             {
-                _logger.Verbose("Trying to connect with server '{0}'.", Options.ChannelOptions);
+                _logger.Verbose("Trying to connect with server '{0}'", Options.ChannelOptions);
+
                 await _adapter.ConnectAsync(effectiveCancellationToken.Token).ConfigureAwait(false);
-                _logger.Verbose("Connection with server established.");
-                
-                var connectResult = await AuthenticateAsync(Options, effectiveCancellationToken.Token).ConfigureAwait(false);
+
+                _logger.Verbose("Connection with server established");
+
+                var connectResult = await _authenticationHandler.Authenticate(_adapter, Options, effectiveCancellationToken.Token).ConfigureAwait(false);
 
                 if (connectResult.ResultCode == MqttClientConnectResultCode.Success)
                 {
                     _publishPacketReceiverQueue?.Dispose();
                     _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
-                    
-                    _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
-                    _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
+
+                    _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(clientAliveToken), clientAliveToken);
+                    _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(clientAliveToken), clientAliveToken);
+
+                    _keepAliveHandler.Enable(connectResult);
                 }
 
                 return connectResult;
@@ -558,9 +528,10 @@ namespace MQTTnet.Client
 
                 var receiverTask = _packetReceiverTask.WaitAsync(sender, _logger);
                 var publishPacketReceiverTask = _publishPacketReceiverTask.WaitAsync(sender, _logger);
-                var keepAliveTask = _keepAlivePacketsSenderTask.WaitAsync(sender, _logger);
 
-                await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false);
+                _keepAliveHandler.Disable();
+
+                await Task.WhenAll(receiverTask, publishPacketReceiverTask).ConfigureAwait(false);
             }
             catch (Exception innerException)
             {
@@ -572,34 +543,10 @@ namespace MQTTnet.Client
                 _cleanDisconnectInitiated = false;
                 CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnected, MqttClientConnectionStatus.Disconnecting);
 
-                _logger.Info("Disconnected.");
-
-                var eventArgs = new MqttClientDisconnectedEventArgs(
-                    clientWasConnected,
-                    connectResult,
-                    (MqttClientDisconnectReason)_disconnectReason,
-                    _disconnectReasonString,
-                    _disconnectUserProperties,
-                    exception);
-
-                // This handler must be executed in a new thread because otherwise a dead lock may happen
-                // when trying to reconnect in that handler etc.
-                Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
+                OnDisconnected(clientWasConnected, connectResult, exception);
             }
         }
 
-        Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
-        {
-            var clientWasConnected = IsConnected;
-
-            if (!DisconnectIsPendingOrFinished())
-            {
-                return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected);
-            }
-
-            return CompletedTask.Instance;
-        }
-
         bool DisconnectIsPendingOrFinished()
         {
             var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
@@ -648,12 +595,42 @@ namespace MQTTnet.Client
             return eventArgs;
         }
 
+        Task OnConnected(MqttClientConnectResult connectResult)
+        {
+            _logger.Info("Successfully connected");
+
+            if (_connectedEvent.HasHandlers)
+            {
+                var eventArgs = new MqttClientConnectedEventArgs(connectResult);
+                return _connectedEvent.InvokeAsync(eventArgs);
+            }
+
+            return CompletedTask.Instance;
+        }
+
+        void OnDisconnected(bool clientWasConnected, MqttClientConnectResult connectResult, Exception exception)
+        {
+            _logger.Info("Disconnected");
+
+            var eventArgs = new MqttClientDisconnectedEventArgs(
+                clientWasConnected,
+                connectResult,
+                (MqttClientDisconnectReason)_disconnectReason,
+                _disconnectReasonString,
+                _disconnectUserProperties,
+                exception);
+
+            // This handler must be executed in a new thread because otherwise a dead lock may happen
+            // when trying to reconnect in that handler etc.
+            Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
+        }
+
         Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
         {
             var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
             if (extendedAuthenticationExchangeHandler != null)
             {
-                return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this));
+                return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeEventArgs(authPacket));
             }
 
             return CompletedTask.Instance;
@@ -787,15 +764,6 @@ namespace MQTTnet.Client
             }
         }
 
-        Task SendAsync(MqttPacket packet, CancellationToken cancellationToken)
-        {
-            cancellationToken.ThrowIfCancellationRequested();
-
-            _lastPacketSentTimestamp = DateTime.UtcNow;
-
-            return _adapter.SendPacketAsync(packet, cancellationToken);
-        }
-
         void ThrowIfConnected(string message)
         {
             if (IsConnected)
@@ -841,7 +809,7 @@ namespace MQTTnet.Client
             {
                 try
                 {
-                    _mqttClientAlive?.Cancel(false);
+                    _clientAlive?.Cancel(false);
                 }
                 catch (Exception exception)
                 {
@@ -983,62 +951,5 @@ namespace MQTTnet.Client
                 _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)
-            {
-                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.");
-            }
-        }
     }
 }

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -87,17 +87,7 @@ namespace MQTTnet.Client
 
             return client.ConnectAsync(client.Options, cancellationToken);
         }
-
-        public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
-        {
-            if (client == null)
-            {
-                throw new ArgumentNullException(nameof(client));
-            }
-
-            return client.SendExtendedAuthenticationExchangeDataAsync(data, CancellationToken.None);
-        }
-
+        
         public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient mqttClient, MqttTopicFilter topicFilter, CancellationToken cancellationToken = default)
         {
             if (mqttClient == null)

Source/MQTTnet/Formatter/MqttAuthPacketFactory.cs

@@ -0,0 +1,36 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using MQTTnet.Client;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Formatter
+{
+    public sealed class MqttAuthPacketFactory
+    {
+        public MqttAuthPacket Create(MqttAuthPacket serverAuthPacket, MqttExtendedAuthenticationExchangeResponse response)
+        {
+            if (serverAuthPacket == null)
+            {
+                throw new ArgumentNullException(nameof(serverAuthPacket));
+            }
+
+            if (response == null)
+            {
+                throw new ArgumentNullException(nameof(response));
+            }
+
+            return new MqttAuthPacket
+            {
+                ReasonCode = MqttAuthenticateReasonCode.ContinueAuthentication,
+                ReasonString = null,
+                AuthenticationMethod = serverAuthPacket.AuthenticationMethod,
+                AuthenticationData = response.AuthenticationData,
+                UserProperties = response.UserProperties
+            };
+        }
+    }
+}

Source/MQTTnet/Formatter/MqttPacketFactories.cs

@@ -6,6 +6,8 @@ namespace MQTTnet.Formatter
 {
     public static class MqttPacketFactories
     {
+        public static MqttAuthPacketFactory Auth { get; } = new MqttAuthPacketFactory();
+        
         public static MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory();
         
         public static MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory();

Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.LowLevelClient
 {
     public interface ILowLevelMqttClient : IDisposable
     {
-        event Func<InspectMqttPacketEventArgs, Task> InspectPackage;
+        event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
         
         bool IsConnected { get; }
         

Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs

@@ -32,7 +32,7 @@ namespace MQTTnet.LowLevelClient
             _logger = logger.WithSource(nameof(LowLevelMqttClient));
         }
 
-        public event Func<InspectMqttPacketEventArgs, Task> InspectPackage
+        public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
         {
             add => _inspectPacketEvent.AddHandler(value);
             remove => _inspectPacketEvent.RemoveHandler(value);