From 4481869d42e6c2d0a3205918e33291f1de97d297 Mon Sep 17 00:00:00 2001 From: dcharles Date: Tue, 7 May 2024 10:16:32 -0700 Subject: [PATCH] fix: time out and cancel propagation in managed client (#1987) * fix: time out and cancel propagation in managed client * Update release notes * Refactor code --------- Co-authored-by: christian <6939810+chkr1011@users.noreply.github.com> --- .github/workflows/ReleaseNotes.md | 3 +- .../ManagedMqttClient.cs | 70 +++++---- .../ManagedMqttClient_Tests.cs | 140 +++++++++++++++++- 3 files changed, 183 insertions(+), 30 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index ecac6db0..9aab3300 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -3,7 +3,8 @@ * [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs). * [Client] The client will now throw a _MqttClientUnexpectedDisconnectReceivedException_ when publishing a QoS 0 message which leads to a server disconnect (BREAKING CHANGE!, #1974, thanks to @fazho). * [Client] Exposed the certificate selection event handler in client options (#1984). -* [Server] The server will no longer send _NoMatchingSubscribers_ when the actual subscription was non success (#1965, BREAKING CHANGE!). +* [Server] The server will no longer send _NoMatchingSubscribers_ when the actual subscription was non success (#1965, BREAKING CHANGE!). * [Server] Fixed broken support for _null_ in _AddServer_ method in ASP.NET integration (#1981). * [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene). +* [ManagedClient] Fixed race condition when server shuts down while subscribing (#1987, thanks to @marve). * [TopicTemplate] Added new extension which provides a template engine for topics (#1932, thanks to @simonthum). diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 7450237d..109eb69a 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -19,6 +19,8 @@ namespace MQTTnet.Extensions.ManagedClient { public sealed class ManagedMqttClient : Disposable, IManagedMqttClient { + readonly MqttNetSourceLogger _logger; + readonly AsyncEvent _interceptingPublishMessageEvent = new AsyncEvent(); readonly AsyncEvent _applicationMessageProcessedEvent = new AsyncEvent(); readonly AsyncEvent _applicationMessageSkippedEvent = new AsyncEvent(); @@ -27,9 +29,7 @@ namespace MQTTnet.Extensions.ManagedClient readonly AsyncEvent _synchronizingSubscriptionsFailedEvent = new AsyncEvent(); readonly AsyncEvent _subscriptionsChangedEvent = new AsyncEvent(); - readonly MqttNetSourceLogger _logger; readonly BlockingQueue _messageQueue = new BlockingQueue(); - readonly AsyncLock _messageQueueLock = new AsyncLock(); /// @@ -83,21 +83,19 @@ namespace MQTTnet.Extensions.ManagedClient add => _applicationMessageSkippedEvent.AddHandler(value); remove => _applicationMessageSkippedEvent.RemoveHandler(value); } - + public event Func ApplicationMessageProcessedAsync { add => _applicationMessageProcessedEvent.AddHandler(value); remove => _applicationMessageProcessedEvent.RemoveHandler(value); } - public event Func InterceptPublishMessageAsync { add => _interceptingPublishMessageEvent.AddHandler(value); remove => _interceptingPublishMessageEvent.RemoveHandler(value); } - public event Func ApplicationMessageReceivedAsync { add => InternalClient.ApplicationMessageReceivedAsync += value; @@ -149,7 +147,7 @@ namespace MQTTnet.Extensions.ManagedClient public ManagedMqttClientOptions Options { get; private set; } public int PendingApplicationMessagesCount => _messageQueue.Count; - + public async Task EnqueueAsync(MqttApplicationMessage applicationMessage) { ThrowIfDisposed(); @@ -277,7 +275,7 @@ namespace MQTTnet.Extensions.ManagedClient ThrowIfDisposed(); _isCleanDisconnect = cleanDisconnect; - + StopPublishing(); StopMaintainingConnection(); @@ -369,6 +367,13 @@ namespace MQTTnet.Extensions.ManagedClient return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime; } + CancellationTokenSource NewTimeoutToken(CancellationToken linkedToken) + { + var newTimeoutToken = CancellationTokenSource.CreateLinkedTokenSource(linkedToken); + newTimeoutToken.CancelAfter(Options.ClientOptions.Timeout); + return newTimeoutToken; + } + async Task HandleSubscriptionExceptionAsync(Exception exception, List addedSubscriptions, List removedSubscriptions) { _logger.Warning(exception, "Synchronizing subscriptions failed."); @@ -411,7 +416,7 @@ namespace MQTTnet.Extensions.ManagedClient { if (_isCleanDisconnect) { - using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout)) + using (var disconnectTimeout = NewTimeoutToken(CancellationToken.None)) { await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false); } @@ -461,7 +466,7 @@ namespace MQTTnet.Extensions.ManagedClient cancellationToken.ThrowIfCancellationRequested(); - await TryPublishQueuedMessageAsync(message).ConfigureAwait(false); + await TryPublishQueuedMessageAsync(message, cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException) @@ -477,7 +482,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - async Task PublishReconnectSubscriptionsAsync() + async Task PublishReconnectSubscriptionsAsync(CancellationToken cancellationToken) { _logger.Info("Publishing subscriptions at reconnect"); @@ -489,20 +494,20 @@ namespace MQTTnet.Extensions.ManagedClient { topicFilters = new List(); SendSubscribeUnsubscribeResult subscribeUnsubscribeResult; - + foreach (var sub in _reconnectSubscriptions) { topicFilters.Add(sub.Value); if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null, cancellationToken).ConfigureAwait(false); topicFilters.Clear(); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); } } - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null, cancellationToken).ConfigureAwait(false); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); } } @@ -555,13 +560,13 @@ namespace MQTTnet.Extensions.ManagedClient if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null, cancellationToken).ConfigureAwait(false); addedTopicFilters.Clear(); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); } } - - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null, cancellationToken).ConfigureAwait(false); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); var removedTopicFilters = new List(); @@ -571,13 +576,13 @@ namespace MQTTnet.Extensions.ManagedClient if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters, cancellationToken).ConfigureAwait(false); removedTopicFilters.Clear(); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); } } - subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters, cancellationToken).ConfigureAwait(false); await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false); } } @@ -592,7 +597,7 @@ namespace MQTTnet.Extensions.ManagedClient MqttClientConnectResult connectResult = null; try { - using (var connectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout)) + using (var connectTimeout = NewTimeoutToken(cancellationToken)) { connectResult = await InternalClient.ConnectAsync(Options.ClientOptions, connectTimeout.Token).ConfigureAwait(false); } @@ -611,7 +616,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - async Task SendSubscribeUnsubscribe(List addedSubscriptions, List removedSubscriptions) + async Task SendSubscribeUnsubscribe(List addedSubscriptions, List removedSubscriptions, CancellationToken cancellationToken) { var subscribeResults = new List(); var unsubscribeResults = new List(); @@ -626,8 +631,11 @@ namespace MQTTnet.Extensions.ManagedClient unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription); } - var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false); - unsubscribeResults.Add(unsubscribeResult); + using (var unsubscribeTimeout = NewTimeoutToken(cancellationToken)) + { + var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build(), unsubscribeTimeout.Token).ConfigureAwait(false); + unsubscribeResults.Add(unsubscribeResult); + } //clear because these worked, maybe the subscribe below will fail, only report those removedSubscriptions.Clear(); @@ -642,8 +650,11 @@ namespace MQTTnet.Extensions.ManagedClient subscribeOptionsBuilder.WithTopicFilter(addedSubscription); } - var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false); - subscribeResults.Add(subscribeResult); + using (var subscribeTimeout = NewTimeoutToken(cancellationToken)) + { + var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build(), subscribeTimeout.Token).ConfigureAwait(false); + subscribeResults.Add(subscribeResult); + } } } catch (Exception exception) @@ -697,7 +708,7 @@ namespace MQTTnet.Extensions.ManagedClient { var oldConnectionState = InternalClient.IsConnected; var connectionState = await ReconnectIfRequiredAsync(cancellationToken).ConfigureAwait(false); - + if (connectionState == ReconnectionResult.NotConnected) { StopPublishing(); @@ -705,7 +716,7 @@ namespace MQTTnet.Extensions.ManagedClient } else if (connectionState == ReconnectionResult.Reconnected) { - await PublishReconnectSubscriptionsAsync().ConfigureAwait(false); + await PublishReconnectSubscriptionsAsync(cancellationToken).ConfigureAwait(false); StartPublishing(); } else if (connectionState == ReconnectionResult.Recovered) @@ -735,7 +746,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message) + async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message, CancellationToken cancellationToken) { Exception transmitException = null; bool acceptPublish = true; @@ -750,7 +761,10 @@ namespace MQTTnet.Extensions.ManagedClient if (acceptPublish) { - await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); + using (var publishTimeout = NewTimeoutToken(cancellationToken)) + { + await InternalClient.PublishAsync(message.ApplicationMessage, publishTimeout.Token).ConfigureAwait(false); + } } using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync diff --git a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs index e8e10005..ed459677 100644 --- a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs @@ -7,12 +7,14 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; +using Microsoft.CodeAnalysis.CSharp.Syntax; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Extensions.ManagedClient; using MQTTnet.Formatter; using MQTTnet.Internal; +using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; @@ -574,16 +576,152 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient } } + [TestMethod] + public async Task Subscribe_Does_Not_Hang_On_Server_Stop() + { + var timeout = TimeSpan.FromSeconds(2); + var testTimeout = timeout * 2; + const string topic = "test_topic_2"; + using (var testEnvironment = CreateTestEnvironment()) + using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout)) + { + testEnvironment.IgnoreClientLogErrors = true; + bool reject = true; + var receivedOnServer = new SemaphoreSlim(0, 1); + var failedOnClient = new SemaphoreSlim(0, 1); + testEnvironment.Server.InterceptingInboundPacketAsync += e => + { + if (e.Packet is MqttSubscribePacket) + { + if (reject) + { + e.ProcessPacket = false; + } + receivedOnServer.Release(); + } + return Task.CompletedTask; + }; + managedClient.SynchronizingSubscriptionsFailedAsync += e => + { + failedOnClient.Release(); + return Task.CompletedTask; + }; + + await managedClient.SubscribeAsync(topic); + Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout)); + Assert.IsTrue(await failedOnClient.WaitAsync(testTimeout)); + + reject = false; + await managedClient.SubscribeAsync(topic); + Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout)); + } + } + + [TestMethod] + public async Task Unsubscribe_Does_Not_Hang_On_Server_Stop() + { + var timeout = TimeSpan.FromSeconds(2); + var testTimeout = timeout * 2; + const string topic = "test_topic_2"; + using (var testEnvironment = CreateTestEnvironment()) + using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout)) + { + testEnvironment.IgnoreClientLogErrors = true; + bool reject = true; + var receivedOnServer = new SemaphoreSlim(0, 1); + var failedOnClient = new SemaphoreSlim(0, 1); + testEnvironment.Server.InterceptingInboundPacketAsync += e => + { + if (e.Packet is MqttUnsubscribePacket) + { + if (reject) + { + e.ProcessPacket = false; + } + receivedOnServer.Release(); + } + else if (e.Packet is MqttSubscribePacket) + { + receivedOnServer.Release(); + } + return Task.CompletedTask; + }; + managedClient.SynchronizingSubscriptionsFailedAsync += e => + { + failedOnClient.Release(); + return Task.CompletedTask; + }; + + await managedClient.SubscribeAsync(topic); + Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout)); + + await managedClient.UnsubscribeAsync(topic); + Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout)); + Assert.IsTrue(await failedOnClient.WaitAsync(testTimeout)); + + reject = false; + await managedClient.UnsubscribeAsync(topic); + Assert.IsTrue(await receivedOnServer.WaitAsync(testTimeout)); + } + } + + [TestMethod] + public async Task Publish_Does_Not_Hang_On_Server_Error() + { + var timeout = TimeSpan.FromSeconds(2); + var testTimeout = timeout * 2; + + const string topic = "test_topic_42"; + + using (var testEnvironment = CreateTestEnvironment()) + using (var managedClient = await CreateManagedClientAsync(testEnvironment, timeout: timeout)) + { + testEnvironment.IgnoreClientLogErrors = true; + bool reject = true; + var receivedOnServer = new TaskCompletionSource(); + managedClient.ApplicationMessageProcessedAsync += e => Task.FromResult(reject &= e.Exception is null); + testEnvironment.Server.InterceptingInboundPacketAsync += e => + { + if (e.Packet is MqttPublishPacket) + { + if (reject) + { + e.ProcessPacket = false; + } + else + { + receivedOnServer.TrySetResult(); + } + } + return Task.CompletedTask; + }; + + + await managedClient.EnqueueAsync(new MqttApplicationMessage { Topic = topic, Payload = new byte[] { 1 }, Retain = true, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); + + var timeoutTask = Task.Delay(testTimeout); + + var firstDone = await Task.WhenAny(receivedOnServer.Task, timeoutTask); + Assert.AreEqual(receivedOnServer.Task, firstDone, "Client is hung on publish!"); + } + } + async Task CreateManagedClientAsync( TestEnvironment testEnvironment, IMqttClient underlyingClient = null, TimeSpan? connectionCheckInterval = null, - string host = "localhost") + string host = "localhost", + TimeSpan? timeout = null) { await testEnvironment.StartServer(); var clientOptions = new MqttClientOptionsBuilder().WithTcpServer(host, testEnvironment.ServerPort); + if (timeout != null) + { + clientOptions.WithTimeout(timeout.Value); + } + var managedOptions = new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build(); // Use a short connection check interval so that subscription operations are performed quickly