From d63143bdc9d45c62c48dd39718e0c8616c278528 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 14 Sep 2020 23:49:27 +0200 Subject: [PATCH] ManagedClient: Add new option for disabling auto reconnect. --- .../IManagedMqttClientOptions.cs | 2 + .../ManagedMqttClient.cs | 99 +++++++++++-------- .../ManagedMqttClientOptions.cs | 3 +- .../ManagedMqttClientOptionsBuilder.cs | 17 +++- 4 files changed, 78 insertions(+), 43 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index a705efea..91b2041e 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/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; } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 7f3d14b5..084a0994 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Extensions.ManagedClient { public class ManagedMqttClient : Disposable, IManagedMqttClient { - private readonly BlockingQueue _messageQueue = new BlockingQueue(); + readonly BlockingQueue _messageQueue = new BlockingQueue(); /// /// The subscriptions are managed in 2 separate buckets: @@ -27,21 +27,21 @@ namespace MQTTnet.Extensions.ManagedClient /// any thread and are therefore mutex'ed. get sent to the broker /// at reconnect and are solely owned by . /// - private readonly Dictionary _reconnectSubscriptions = new Dictionary(); - private readonly Dictionary _subscriptions = new Dictionary(); - private readonly HashSet _unsubscriptions = new HashSet(); - private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); + readonly Dictionary _reconnectSubscriptions = new Dictionary(); + readonly Dictionary _subscriptions = new Dictionary(); + readonly HashSet _unsubscriptions = new HashSet(); + 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 ReconnectIfRequiredAsync() + async Task 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; diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index 537e1bb9..d0ed4dc1 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/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); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs index f58453ce..5f7a29fa 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/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)