v4.0.0 (#1357)

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -1,162 +1,155 @@
-using MQTTnet.Client;
-using MQTTnet.Client.Connecting;
-using MQTTnet.Client.Disconnecting;
-using MQTTnet.Client.Publishing;
-using MQTTnet.Client.Receiving;
-using MQTTnet.Diagnostics;
-using MQTTnet.Exceptions;
-using MQTTnet.Internal;
-using MQTTnet.Protocol;
-using MQTTnet.Server;
+// 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 System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Diagnostics.Logger;
+using MQTTnet.Client;
+using MQTTnet.Diagnostics;
+using MQTTnet.Exceptions;
+using MQTTnet.Internal;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+using MQTTnet.Server;
+using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
-    public class ManagedMqttClient : Disposable, IManagedMqttClient
+    public sealed class ManagedMqttClient : Disposable
     {
+        readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
+        readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
+        readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
+
+        readonly MqttNetSourceLogger _logger;
         readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
 
+        readonly AsyncLock _messageQueueLock = new AsyncLock();
+
         /// <summary>
-        /// The subscriptions are managed in 2 separate buckets:
-        /// <see cref="_subscriptions"/> and <see cref="_unsubscriptions"/> are processed during normal operation
-        /// and are moved to the <see cref="_reconnectSubscriptions"/> when they get processed. They can be accessed by
-        /// any thread and are therefore mutex'ed. <see cref="_reconnectSubscriptions"/> get sent to the broker
-        ///  at reconnect and are solely owned by <see cref="MaintainConnectionAsync"/>.
+        ///     The subscriptions are managed in 2 separate buckets:
+        ///     <see
+        ///         cref="_subscriptions" />
+        ///     and
+        ///     <see
+        ///         cref="_unsubscriptions" />
+        ///     are processed during normal operation
+        ///     and are moved to the
+        ///     <see
+        ///         cref="_reconnectSubscriptions" />
+        ///     when they get processed. They can be accessed by
+        ///     any thread and are therefore mutex'ed.
+        ///     <see
+        ///         cref="_reconnectSubscriptions" />
+        ///     get sent to the broker
+        ///     at reconnect and are solely owned by
+        ///     <see
+        ///         cref="MaintainConnectionAsync" />
+        ///     .
         /// </summary>
         readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
 
         readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        readonly HashSet<string> _unsubscriptions = new HashSet<string>();
         readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
-
-        readonly MqttNetSourceLogger _logger;
-
-        readonly AsyncLock _messageQueueLock = new AsyncLock();
+        readonly HashSet<string> _unsubscriptions = new HashSet<string>();
 
         CancellationTokenSource _connectionCancellationToken;
-        CancellationTokenSource _publishingCancellationToken;
         Task _maintainConnectionTask;
+        CancellationTokenSource _publishingCancellationToken;
 
         ManagedMqttClientStorageManager _storageManager;
 
-        public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
+        public ManagedMqttClient(MqttClient mqttClient, IMqttNetLogger logger)
         {
             InternalClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
 
-            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (logger == null)
+            {
+                throw new ArgumentNullException(nameof(logger));
+            }
+
             _logger = logger.WithSource(nameof(ManagedMqttClient));
         }
 
+        public event Func<ApplicationMessageProcessedEventArgs, Task> ApplicationMessageProcessedAsync
+        {
+            add => _applicationMessageProcessedEvent.AddHandler(value);
+            remove => _applicationMessageProcessedEvent.RemoveHandler(value);
+        }
+
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => InternalClient.ApplicationMessageReceivedAsync += value;
+            remove => InternalClient.ApplicationMessageReceivedAsync -= value;
+        }
+
+        public event Func<EventArgs, Task> ConnectedAsync
+        {
+            add => InternalClient.ConnectedAsync += value;
+            remove => InternalClient.ConnectedAsync -= value;
+        }
+
+        public event Func<ConnectingFailedEventArgs, Task> ConnectingFailedAsync
+        {
+            add => _connectingFailedEvent.AddHandler(value);
+            remove => _connectingFailedEvent.RemoveHandler(value);
+        }
+
+        public event Func<EventArgs, Task> ConnectionStateChangedAsync
+        {
+            add => _connectionStateChangedEvent.AddHandler(value);
+            remove => _connectionStateChangedEvent.RemoveHandler(value);
+        }
+
+        public event Func<EventArgs, Task> DisconnectedAsync
+        {
+            add => InternalClient.DisconnectedAsync += value;
+            remove => InternalClient.DisconnectedAsync -= value;
+        }
+
+        public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; }
+
+        public MqttClient InternalClient { get; }
+
         public bool IsConnected => InternalClient.IsConnected;
 
         public bool IsStarted => _connectionCancellationToken != null;
 
-        public IMqttClient InternalClient { get; }
+        public ManagedMqttClientOptions Options { get; private set; }
 
         public int PendingApplicationMessagesCount => _messageQueue.Count;
 
-        public IManagedMqttClientOptions Options { get; private set; }
-
-        public IMqttClientConnectedHandler ConnectedHandler
-        {
-            get => InternalClient.ConnectedHandler;
-            set => InternalClient.ConnectedHandler = value;
-        }
-
-        public IMqttClientDisconnectedHandler DisconnectedHandler
-        {
-            get => InternalClient.DisconnectedHandler;
-            set => InternalClient.DisconnectedHandler = value;
-        }
-
-        public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
-        {
-            get => InternalClient.ApplicationMessageReceivedHandler;
-            set => InternalClient.ApplicationMessageReceivedHandler = value;
-        }
-
-        public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; }
-
-        public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; }
-
-        public IConnectingFailedHandler ConnectingFailedHandler { get; set; }
-
         public ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
 
-        public async Task StartAsync(IManagedMqttClientOptions options)
+        public async Task EnqueueAsync(MqttApplicationMessage applicationMessage)
         {
             ThrowIfDisposed();
 
-            if (options == null) throw new ArgumentNullException(nameof(options));
-            if (options.ClientOptions == null) throw new ArgumentException("The client options are not set.", nameof(options));
-
-            if (!_maintainConnectionTask?.IsCompleted ?? false) throw new InvalidOperationException("The managed client is already started.");
-
-            Options = options;
-
-            if (options.Storage != null)
+            if (applicationMessage == null)
             {
-                _storageManager = new ManagedMqttClientStorageManager(options.Storage);
-                var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
-
-                foreach (var message in messages)
-                {
-                    _messageQueue.Enqueue(message);
-                }
+                throw new ArgumentNullException(nameof(applicationMessage));
             }
 
-            var cancellationTokenSource = new CancellationTokenSource();
-            var cancellationToken = cancellationTokenSource.Token;
-            _connectionCancellationToken = cancellationTokenSource;
-
-            _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(cancellationToken), cancellationToken);
-            _maintainConnectionTask.RunInBackground(_logger);
-
-            _logger.Info("Started");
+            var managedMqttApplicationMessage = new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage);
+            await EnqueueAsync(managedMqttApplicationMessage.Build()).ConfigureAwait(false);
         }
 
-        public async Task StopAsync()
+        public async Task EnqueueAsync(ManagedMqttApplicationMessage applicationMessage)
         {
             ThrowIfDisposed();
 
-            StopPublishing();
-            StopMaintainingConnection();
-
-            _messageQueue.Clear();
-
-            if (_maintainConnectionTask != null)
+            if (applicationMessage == null)
             {
-                await Task.WhenAny(_maintainConnectionTask);
-                _maintainConnectionTask = null;
+                throw new ArgumentNullException(nameof(applicationMessage));
             }
-        }
 
-        public Task PingAsync(CancellationToken cancellationToken)
-        {
-            return InternalClient.PingAsync(cancellationToken);
-        }
-
-        public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
-            await PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build()).ConfigureAwait(false);
-            return new MqttClientPublishResult();
-        }
-
-        public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
-        {
-            ThrowIfDisposed();
-
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-            if (Options == null) throw new InvalidOperationException("call StartAsync before publishing messages");
+            if (Options == null)
+            {
+                throw new InvalidOperationException("call StartAsync before publishing messages");
+            }
 
             MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
 
@@ -210,11 +203,77 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        public Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
+        public Task PingAsync(CancellationToken cancellationToken)
+        {
+            return InternalClient.PingAsync(cancellationToken);
+        }
+
+        public async Task StartAsync(ManagedMqttClientOptions options)
         {
             ThrowIfDisposed();
 
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            if (options.ClientOptions == null)
+            {
+                throw new ArgumentException("The client options are not set.", nameof(options));
+            }
+
+            if (!_maintainConnectionTask?.IsCompleted ?? false)
+            {
+                throw new InvalidOperationException("The managed client is already started.");
+            }
+
+            Options = options;
+
+            if (options.Storage != null)
+            {
+                _storageManager = new ManagedMqttClientStorageManager(options.Storage);
+                var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
+
+                foreach (var message in messages)
+                {
+                    _messageQueue.Enqueue(message);
+                }
+            }
+
+            var cancellationTokenSource = new CancellationTokenSource();
+            var cancellationToken = cancellationTokenSource.Token;
+            _connectionCancellationToken = cancellationTokenSource;
+
+            _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(cancellationToken), cancellationToken);
+            _maintainConnectionTask.RunInBackground(_logger);
+
+            _logger.Info("Started");
+        }
+
+        public async Task StopAsync()
+        {
+            ThrowIfDisposed();
+
+            StopPublishing();
+            StopMaintainingConnection();
+
+            _messageQueue.Clear();
+
+            if (_maintainConnectionTask != null)
+            {
+                await Task.WhenAny(_maintainConnectionTask);
+                _maintainConnectionTask = null;
+            }
+        }
+
+        public Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)
+        {
+            ThrowIfDisposed();
+
+            if (topicFilters == null)
+            {
+                throw new ArgumentNullException(nameof(topicFilters));
+            }
 
             foreach (var topicFilter in topicFilters)
             {
@@ -235,11 +294,14 @@ namespace MQTTnet.Extensions.ManagedClient
             return Task.FromResult(0);
         }
 
-        public Task UnsubscribeAsync(IEnumerable<string> topics)
+        public Task UnsubscribeAsync(ICollection<string> topics)
         {
             ThrowIfDisposed();
 
-            if (topics == null) throw new ArgumentNullException(nameof(topics));
+            if (topics == null)
+            {
+                throw new ArgumentNullException(nameof(topics));
+            }
 
             lock (_subscriptions)
             {
@@ -277,6 +339,23 @@ namespace MQTTnet.Extensions.ManagedClient
             base.Dispose(disposing);
         }
 
+        static TimeSpan GetRemainingTime(DateTime endTime)
+        {
+            var remainingTime = endTime - DateTime.UtcNow;
+            return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;
+        }
+
+        async Task HandleSubscriptionExceptionAsync(Exception exception)
+        {
+            _logger.Warning(exception, "Synchronizing subscriptions failed.");
+
+            var synchronizingSubscriptionsFailedHandler = SynchronizingSubscriptionsFailedHandler;
+            if (SynchronizingSubscriptionsFailedHandler != null)
+            {
+                await synchronizingSubscriptionsFailedHandler.HandleSynchronizingSubscriptionsFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
+            }
+        }
+
         async Task MaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
@@ -299,9 +378,9 @@ namespace MQTTnet.Extensions.ManagedClient
                 {
                     try
                     {
-                        using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.CommunicationTimeout))
+                        using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
                         {
-                            await InternalClient.DisconnectAsync(disconnectTimeout.Token).ConfigureAwait(false);
+                            await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
                         }
                     }
                     catch (OperationCanceledException)
@@ -326,49 +405,6 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
-        {
-            try
-            {
-                var connectionState = await ReconnectIfRequiredAsync(cancellationToken).ConfigureAwait(false);
-                if (connectionState == ReconnectionResult.NotConnected)
-                {
-                    StopPublishing();
-                    await Task.Delay(Options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
-                    return;
-                }
-
-                if (connectionState == ReconnectionResult.Reconnected)
-                {
-                    await PublishReconnectSubscriptionsAsync().ConfigureAwait(false);
-                    StartPublishing();
-                    return;
-                }
-
-                if (connectionState == ReconnectionResult.Recovered)
-                {
-                    StartPublishing();
-                    return;
-                }
-
-                if (connectionState == ReconnectionResult.StillConnected)
-                {
-                    await PublishSubscriptionsAsync(Options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
-                }
-            }
-            catch (OperationCanceledException)
-            {
-            }
-            catch (MqttCommunicationException exception)
-            {
-                _logger.Warning(exception, "Communication error while maintaining connection.");
-            }
-            catch (Exception exception)
-            {
-                _logger.Error(exception, "Error exception while maintaining connection.");
-            }
-        }
-
         async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
         {
             try
@@ -407,6 +443,259 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
+        async Task PublishReconnectSubscriptionsAsync()
+        {
+            _logger.Info("Publishing subscriptions at reconnect");
+
+            try
+            {
+                if (_reconnectSubscriptions.Any())
+                {
+                    var subscriptions = _reconnectSubscriptions.Select(
+                        i => new MqttTopicFilter
+                        {
+                            Topic = i.Key,
+                            QualityOfServiceLevel = i.Value
+                        });
+
+                    var topicFilters = new List<MqttTopicFilter>();
+
+                    foreach (var sub in subscriptions)
+                    {
+                        topicFilters.Add(sub);
+
+                        if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
+                        {
+                            await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                            topicFilters.Clear();
+                        }
+                    }
+
+                    await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
+                }
+            }
+            catch (Exception exception)
+            {
+                await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
+            }
+        }
+
+        async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            var endTime = DateTime.UtcNow + timeout;
+
+            while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
+            {
+                List<MqttTopicFilter> subscriptions;
+                HashSet<string> unsubscriptions;
+
+                lock (_subscriptions)
+                {
+                    subscriptions = _subscriptions.Select(
+                            i => new MqttTopicFilter
+                            {
+                                Topic = i.Key,
+                                QualityOfServiceLevel = i.Value
+                            })
+                        .ToList();
+
+                    _subscriptions.Clear();
+
+                    unsubscriptions = new HashSet<string>(_unsubscriptions);
+                    _unsubscriptions.Clear();
+                }
+
+                if (!subscriptions.Any() && !unsubscriptions.Any())
+                {
+                    continue;
+                }
+
+                _logger.Verbose("Publishing {0} added and {1} removed subscriptions", subscriptions.Count, unsubscriptions.Count);
+
+                foreach (var unsubscription in unsubscriptions)
+                {
+                    _reconnectSubscriptions.Remove(unsubscription);
+                }
+
+                foreach (var subscription in subscriptions)
+                {
+                    _reconnectSubscriptions[subscription.Topic] = subscription.QualityOfServiceLevel;
+                }
+
+                var addedTopicFilters = new List<MqttTopicFilter>();
+                foreach (var subscription in subscriptions)
+                {
+                    addedTopicFilters.Add(subscription);
+
+                    if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
+                    {
+                        await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+                        addedTopicFilters.Clear();
+                    }
+                }
+
+                await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
+
+                var removedTopicFilters = new List<string>();
+                foreach (var unSub in unsubscriptions)
+                {
+                    removedTopicFilters.Add(unSub);
+
+                    if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
+                    {
+                        await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+                        removedTopicFilters.Clear();
+                    }
+                }
+
+                await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
+            }
+        }
+
+        async Task<ReconnectionResult> ReconnectIfRequiredAsync(CancellationToken cancellationToken)
+        {
+            if (InternalClient.IsConnected)
+            {
+                return ReconnectionResult.StillConnected;
+            }
+
+            MqttClientConnectResult connectResult = null;
+            try
+            {
+                using (var connectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
+                {
+                    connectResult = await InternalClient.ConnectAsync(Options.ClientOptions, connectTimeout.Token).ConfigureAwait(false);
+                }
+
+                if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
+                {
+                    throw new MqttCommunicationException($"Client connected but server denied connection with reason '{connectResult.ResultCode}'.");
+                }
+
+                return connectResult.IsSessionPresent ? ReconnectionResult.Recovered : ReconnectionResult.Reconnected;
+            }
+            catch (Exception exception)
+            {
+                await _connectingFailedEvent.InvokeAsync(new ConnectingFailedEventArgs(connectResult, exception));
+                return ReconnectionResult.NotConnected;
+            }
+        }
+
+        async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
+        {
+            try
+            {
+                if (removedSubscriptions != null && removedSubscriptions.Any())
+                {
+                    var unsubscribeOptionsBuilder = new MqttClientUnsubscribeOptionsBuilder();
+
+                    foreach (var removedSubscription in removedSubscriptions)
+                    {
+                        unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription);
+                    }
+
+                    await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                }
+
+                if (addedSubscriptions != null && addedSubscriptions.Any())
+                {
+                    var subscribeOptionsBuilder = new MqttClientSubscribeOptionsBuilder();
+
+                    foreach (var addedSubscription in addedSubscriptions)
+                    {
+                        subscribeOptionsBuilder.WithTopicFilter(addedSubscription);
+                    }
+
+                    await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
+                }
+            }
+            catch (Exception exception)
+            {
+                await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
+            }
+        }
+
+        void StartPublishing()
+        {
+            StopPublishing();
+
+            var cancellationTokenSource = new CancellationTokenSource();
+            var cancellationToken = cancellationTokenSource.Token;
+            _publishingCancellationToken = cancellationTokenSource;
+
+            Task.Run(() => PublishQueuedMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger);
+        }
+
+        void StopMaintainingConnection()
+        {
+            try
+            {
+                _connectionCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _connectionCancellationToken?.Dispose();
+                _connectionCancellationToken = null;
+            }
+        }
+
+        void StopPublishing()
+        {
+            try
+            {
+                _publishingCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _publishingCancellationToken?.Dispose();
+                _publishingCancellationToken = null;
+            }
+        }
+
+        async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
+        {
+            try
+            {
+                var oldConnectionState = InternalClient.IsConnected;
+                var connectionState = await ReconnectIfRequiredAsync(cancellationToken).ConfigureAwait(false);
+                
+                if (connectionState == ReconnectionResult.NotConnected)
+                {
+                    StopPublishing();
+                    await Task.Delay(Options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
+                }
+                else if (connectionState == ReconnectionResult.Reconnected)
+                {
+                    await PublishReconnectSubscriptionsAsync().ConfigureAwait(false);
+                    StartPublishing();
+                }
+                else if (connectionState == ReconnectionResult.Recovered)
+                {
+                    StartPublishing();
+                }
+                else if (connectionState == ReconnectionResult.StillConnected)
+                {
+                    await PublishSubscriptionsAsync(Options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
+                }
+
+                if (oldConnectionState != InternalClient.IsConnected)
+                {
+                    await _connectionStateChangedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (MqttCommunicationException exception)
+            {
+                _logger.Warning(exception, "Communication error while maintaining connection.");
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error exception while maintaining connection.");
+            }
+        }
+
         async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
@@ -463,220 +752,12 @@ namespace MQTTnet.Extensions.ManagedClient
             }
             finally
             {
-                var eventHandler = ApplicationMessageProcessedHandler;
-                if (eventHandler != null)
+                if (_applicationMessageProcessedEvent.HasHandlers)
                 {
-                    var eventArguments = new ApplicationMessageProcessedEventArgs(message, transmitException);
-                    await eventHandler.HandleApplicationMessageProcessedAsync(eventArguments).ConfigureAwait(false);
+                    var eventArgs = new ApplicationMessageProcessedEventArgs(message, transmitException);
+                    await _applicationMessageProcessedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
             }
         }
-
-        async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
-        {
-            var endTime = DateTime.UtcNow + timeout;
-
-            while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
-            {
-                List<MqttTopicFilter> subscriptions;
-                HashSet<string> unsubscriptions;
-
-                lock (_subscriptions)
-                {
-                    subscriptions = _subscriptions.Select(i => new MqttTopicFilter
-                    {
-                        Topic = i.Key, 
-                        QualityOfServiceLevel = i.Value
-                    }).ToList();
-                    
-                    _subscriptions.Clear();
-                    
-                    unsubscriptions = new HashSet<string>(_unsubscriptions);
-                    _unsubscriptions.Clear();
-                }
-
-                if (!subscriptions.Any() && !unsubscriptions.Any())
-                {
-                    continue;
-                }
-
-                _logger.Verbose("Publishing {0} added and {1} removed subscriptions", subscriptions.Count, unsubscriptions.Count);
-
-                foreach (var unsubscription in unsubscriptions)
-                {
-                    _reconnectSubscriptions.Remove(unsubscription);
-                }
-
-                foreach (var subscription in subscriptions)
-                {
-                    _reconnectSubscriptions[subscription.Topic] = subscription.QualityOfServiceLevel;
-                }
-
-                var addedTopicFilters = new List<MqttTopicFilter>();
-                foreach (var subscription in subscriptions)
-                {
-                    addedTopicFilters.Add(subscription);
-                    
-                    if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
-                    {
-                        await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
-                        addedTopicFilters.Clear();
-                    }
-                }
-
-                await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
-
-                var removedTopicFilters = new List<string>();
-                foreach (var unSub in unsubscriptions)
-                {
-                    removedTopicFilters.Add(unSub);
-                    
-                    if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
-                    {
-                        await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
-                        removedTopicFilters.Clear();
-                    }
-                }
-
-                await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
-            }
-        }
-
-        async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
-        {
-            try
-            {
-                if (removedSubscriptions != null && removedSubscriptions.Any())
-                {
-                    await InternalClient.UnsubscribeAsync(removedSubscriptions.ToArray()).ConfigureAwait(false);
-                }
-
-                if (addedSubscriptions != null && addedSubscriptions.Any())
-                {
-                    await InternalClient.SubscribeAsync(addedSubscriptions.ToArray()).ConfigureAwait(false);
-                }
-            }
-            catch (Exception exception)
-            {
-                await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
-            }
-        }
-
-        async Task PublishReconnectSubscriptionsAsync()
-        {
-            _logger.Info("Publishing subscriptions at reconnect");
-
-            try
-            {
-                if (_reconnectSubscriptions.Any())
-                {
-                    var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter
-                    {
-                        Topic = i.Key, 
-                        QualityOfServiceLevel = i.Value
-                    });
-                    
-                    var topicFilters = new List<MqttTopicFilter>();
-                    
-                    foreach (var sub in subscriptions)
-                    {
-                        topicFilters.Add(sub);
-                        
-                        if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
-                        {
-                            await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
-                            topicFilters.Clear();
-                        }
-                    }
-
-                    await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
-                }
-            }
-            catch (Exception exception)
-            {
-                await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
-            }
-        }
-
-        async Task HandleSubscriptionExceptionAsync(Exception exception)
-        {
-            _logger.Warning(exception, "Synchronizing subscriptions failed.");
-
-            var synchronizingSubscriptionsFailedHandler = SynchronizingSubscriptionsFailedHandler;
-            if (SynchronizingSubscriptionsFailedHandler != null)
-            {
-                await synchronizingSubscriptionsFailedHandler.HandleSynchronizingSubscriptionsFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
-            }
-        }
-
-        async Task<ReconnectionResult> ReconnectIfRequiredAsync(CancellationToken cancellationToken)
-        {
-            if (InternalClient.IsConnected)
-            {
-                return ReconnectionResult.StillConnected;
-            }
-
-            try
-            {
-                var result = await InternalClient.ConnectAsync(Options.ClientOptions, cancellationToken).ConfigureAwait(false);
-                return result.IsSessionPresent ? ReconnectionResult.Recovered : ReconnectionResult.Reconnected;
-            }
-            catch (Exception exception)
-            {
-                var connectingFailedHandler = ConnectingFailedHandler;
-                if (connectingFailedHandler != null)
-                {
-                    await connectingFailedHandler.HandleConnectingFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
-                }
-
-                return ReconnectionResult.NotConnected;
-            }
-        }
-
-        void StartPublishing()
-        {
-            if (_publishingCancellationToken != null)
-            {
-                StopPublishing();
-            }
-
-            var cancellationTokenSource = new CancellationTokenSource();
-            var cancellationToken = cancellationTokenSource.Token;
-            _publishingCancellationToken = cancellationTokenSource;
-
-            Task.Run(() => PublishQueuedMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger);
-        }
-
-        void StopPublishing()
-        {
-            try
-            {
-                _publishingCancellationToken?.Cancel(false);
-            }
-            finally
-            {
-                _publishingCancellationToken?.Dispose();
-                _publishingCancellationToken = null;
-            }
-        }
-
-        void StopMaintainingConnection()
-        {
-            try
-            {
-                _connectionCancellationToken?.Cancel(false);
-            }
-            finally
-            {
-                _connectionCancellationToken?.Dispose();
-                _connectionCancellationToken = null;
-            }
-        }
-
-        static TimeSpan GetRemainingTime(DateTime endTime)
-        {
-            var remainingTime = endTime - DateTime.UtcNow;
-            return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;
-        }
     }
 }