ManagedClient: Add new option for disabling auto reconnect.

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs

@@ -8,6 +8,8 @@ namespace MQTTnet.Extensions.ManagedClient
     {
         IMqttClientOptions ClientOptions { get; }
 
+        bool AutoReconnect { get; }
+
         TimeSpan AutoReconnectDelay { get; }
 
         TimeSpan ConnectionCheckInterval { get; }

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -18,7 +18,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClient : Disposable, IManagedMqttClient
     {
-        private readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
+        readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
 
         /// <summary>
         /// The subscriptions are managed in 2 separate buckets:
@@ -27,21 +27,21 @@ namespace MQTTnet.Extensions.ManagedClient
         /// 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>
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly HashSet<string> _unsubscriptions = new HashSet<string>();
-        private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
+        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);
 
-        private readonly IMqttClient _mqttClient;
-        private readonly IMqttNetScopedLogger _logger;
+        readonly IMqttClient _mqttClient;
+        readonly IMqttNetScopedLogger _logger;
 
-        private readonly AsyncLock _messageQueueLock = new AsyncLock();
+        readonly AsyncLock _messageQueueLock = new AsyncLock();
 
-        private CancellationTokenSource _connectionCancellationToken;
-        private CancellationTokenSource _publishingCancellationToken;
-        private Task _maintainConnectionTask;
+        CancellationTokenSource _connectionCancellationToken;
+        CancellationTokenSource _publishingCancellationToken;
+        Task _maintainConnectionTask;
 
-        private ManagedMqttClientStorageManager _storageManager;
+        ManagedMqttClientStorageManager _storageManager;
 
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
         {
@@ -52,8 +52,11 @@ namespace MQTTnet.Extensions.ManagedClient
         }
 
         public bool IsConnected => _mqttClient.IsConnected;
+
         public bool IsStarted => _connectionCancellationToken != null;
+        
         public int PendingApplicationMessagesCount => _messageQueue.Count;
+        
         public IManagedMqttClientOptions Options { get; private set; }
 
         public IMqttClientConnectedHandler ConnectedHandler
@@ -88,8 +91,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
             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;
@@ -105,11 +107,16 @@ namespace MQTTnet.Extensions.ManagedClient
                 }
             }
 
-            _connectionCancellationToken = new CancellationTokenSource();
-
-            _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token);
-            _maintainConnectionTask.Forget(_logger);
+            if (Options.AutoReconnect)
+            {
+                var cancellationTokenSource = new CancellationTokenSource();
+                var cancellationToken = cancellationTokenSource.Token;
+                _connectionCancellationToken = cancellationTokenSource;
 
+                _maintainConnectionTask = Task.Run(() => MaintainConnectionAsync(cancellationToken), cancellationToken);
+                _maintainConnectionTask.Forget(_logger);
+            }
+            
             _logger.Info("Started");
         }
 
@@ -260,10 +267,11 @@ namespace MQTTnet.Extensions.ManagedClient
                 _mqttClient.Dispose();
                 _subscriptionsQueuedSignal.Dispose();
             }
+
             base.Dispose(disposing);
         }
 
-        private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
+        async Task MaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -303,7 +311,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
+        async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -346,7 +354,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
+        async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -384,7 +392,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
+        async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
             try
@@ -449,7 +457,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             var endTime = DateTime.UtcNow + timeout;
             while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
@@ -501,7 +509,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task PublishReconnectSubscriptionsAsync()
+        async Task PublishReconnectSubscriptionsAsync()
         {
             _logger.Info("Publishing subscriptions at reconnect");
 
@@ -519,7 +527,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task HandleSubscriptionExceptionAsync(Exception exception)
+        async Task HandleSubscriptionExceptionAsync(Exception exception)
         {
             _logger.Warning(exception, "Synchronizing subscriptions failed.");
 
@@ -530,7 +538,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task<ReconnectionResult> ReconnectIfRequiredAsync()
+        async Task<ReconnectionResult> ReconnectIfRequiredAsync()
         {
             if (_mqttClient.IsConnected)
             {
@@ -554,34 +562,47 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private void StartPublishing()
+        void StartPublishing()
         {
             if (_publishingCancellationToken != null)
             {
                 StopPublishing();
             }
 
-            var cts = new CancellationTokenSource();
-            _publishingCancellationToken = cts;
+            var cancellationTokenSource = new CancellationTokenSource();
+            var cancellationToken = cancellationTokenSource.Token;
+            _publishingCancellationToken = cancellationTokenSource;
 
-            Task.Run(() => PublishQueuedMessagesAsync(cts.Token), cts.Token).Forget(_logger);
+            Task.Run(() => PublishQueuedMessagesAsync(cancellationToken), cancellationToken).Forget(_logger);
         }
 
-        private void StopPublishing()
+        void StopPublishing()
         {
-            _publishingCancellationToken?.Cancel(false);
-            _publishingCancellationToken?.Dispose();
-            _publishingCancellationToken = null;
+            try
+            {
+                _publishingCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _publishingCancellationToken?.Dispose();
+                _publishingCancellationToken = null;
+            }
         }
 
-        private void StopMaintainingConnection()
+        void StopMaintainingConnection()
         {
-            _connectionCancellationToken?.Cancel(false);
-            _connectionCancellationToken?.Dispose();
-            _connectionCancellationToken = null;
+            try
+            {
+                _connectionCancellationToken?.Cancel(false);
+            }
+            finally
+            {
+                _connectionCancellationToken?.Dispose();
+                _connectionCancellationToken = null;
+            }
         }
 
-        private TimeSpan GetRemainingTime(DateTime endTime)
+        static TimeSpan GetRemainingTime(DateTime endTime)
         {
             var remainingTime = endTime - DateTime.UtcNow;
             return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs

@@ -1,5 +1,4 @@
 using System;
-using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
@@ -9,6 +8,8 @@ namespace MQTTnet.Extensions.ManagedClient
     {
         public IMqttClientOptions ClientOptions { get; set; }
 
+        public bool AutoReconnect { get; set; } = true;
+
         public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
 
         public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1);

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs

@@ -1,5 +1,4 @@
 using System;
-using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
@@ -7,8 +6,8 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClientOptionsBuilder
     {
-        private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
-        private MqttClientOptionsBuilder _clientOptionsBuilder;
+        readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
+        MqttClientOptionsBuilder _clientOptionsBuilder;
 
         public ManagedMqttClientOptionsBuilder WithMaxPendingMessages(int value)
         {
@@ -70,6 +69,18 @@ namespace MQTTnet.Extensions.ManagedClient
             return this;
         }
 
+        public ManagedMqttClientOptionsBuilder WithAutoReconnect()
+        {
+            _options.AutoReconnect = true;
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithoutAutoReconnect()
+        {
+            _options.AutoReconnect = false;
+            return this;
+        }
+        
         public ManagedMqttClientOptions Build()
         {
             if (_clientOptionsBuilder != null)