From d23bbf7bf36595466fe628d148896c1200b1ddf7 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Wed, 14 Sep 2022 14:17:59 +0200 Subject: [PATCH] Improve PingAsync timeout handling. --- .github/workflows/ReleaseNotes.md | 1 + Samples/Client/Client_Connection_Samples.cs | 6 +- .../MQTTnet.Tests/Client/MqttClient_Tests.cs | 37 +- Source/MQTTnet/Client/IMqttClient.cs | 37 +- Source/MQTTnet/Client/MqttClient.cs | 839 +++++++++--------- Source/MQTTnet/Client/MqttClientExtensions.cs | 20 + 6 files changed, 496 insertions(+), 444 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index a4f66c6c..9f904d4a 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,4 +1,5 @@ * [Core] MQTT Packets being sent over web socket transport are now setting the web socket frame boundaries correctly (#1499). * [Client] Keep alive mechanism now uses the configured timeout value from the options (thanks to @Stannieman, #1495). +* [Client] The _PingAsync_ will fallback to the timeout specified in the client options when the cancellation token cannot be cancelled. * [Server] A DISCONNECT packet is no longer sent to MQTT clients < 5.0.0 (thanks to @logicaloud, #1506). * [Server] Improved "take over" process handling. diff --git a/Samples/Client/Client_Connection_Samples.cs b/Samples/Client/Client_Connection_Samples.cs index 62feb648..18b619c4 100644 --- a/Samples/Client/Client_Connection_Samples.cs +++ b/Samples/Client/Client_Connection_Samples.cs @@ -321,14 +321,12 @@ public static class Client_Connection_Samples { try { - // This code will also do the very first connect! So no call to _ConnectAsync_ is required - // in the first place. - if (!mqttClient.IsConnected) + // This code will also do the very first connect! So no call to _ConnectAsync_ is required in the first place. + if (!await mqttClient.TryPingAsync()) { await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); // Subscribe to topics when session is clean etc. - Console.WriteLine("The MQTT client is connected."); } } diff --git a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs index af7c5719..43a72ab4 100644 --- a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs @@ -19,10 +19,12 @@ using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Tests.Mockups; +// ReSharper disable InconsistentNaming + namespace MQTTnet.Tests.Client { [TestClass] - public sealed class Client_Tests : BaseTestClass + public sealed class MqttClient_Tests : BaseTestClass { [DataTestMethod] [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] @@ -145,6 +147,7 @@ namespace MQTTnet.Tests.Client } catch { + // Ignore errors. } await Task.Delay(500); @@ -239,10 +242,10 @@ namespace MQTTnet.Tests.Client for (var i = 0; i < 98; i++) { - Assert.IsFalse(clients[i].IsConnected, $"clients[{i}] is not connected"); + Assert.IsFalse(clients[i].TryPingAsync().GetAwaiter().GetResult(), $"clients[{i}] is not connected"); } - Assert.IsTrue(clients[99].IsConnected); + Assert.IsTrue(clients[99].TryPingAsync().GetAwaiter().GetResult()); Assert.AreEqual(1, clientStatus.Count); Assert.AreEqual(1, sessionStatus.Count); @@ -332,7 +335,7 @@ namespace MQTTnet.Tests.Client var client = await testEnvironment.ConnectClient(); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult()); } } @@ -343,11 +346,11 @@ namespace MQTTnet.Tests.Client { await testEnvironment.StartServer(); var client = await testEnvironment.ConnectClient(); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult()); await client.DisconnectAsync(); - Assert.IsFalse(client.IsConnected); + Assert.IsFalse(client.TryPingAsync().GetAwaiter().GetResult()); } } @@ -591,17 +594,17 @@ namespace MQTTnet.Tests.Client var client = await testEnvironment.ConnectClient(); await Task.Delay(500); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(await client.TryPingAsync()); await server.StopAsync(); await Task.Delay(500); - Assert.IsFalse(client.IsConnected); + Assert.IsFalse(await client.TryPingAsync()); await server.StartAsync(); await Task.Delay(500); await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build()); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(await client.TryPingAsync()); } } @@ -637,6 +640,7 @@ namespace MQTTnet.Tests.Client } catch { + // Ignore errors. } SpinWait.SpinUntil(() => tries >= maxTries, 10000); @@ -656,11 +660,11 @@ namespace MQTTnet.Tests.Client var client = await testEnvironment.ConnectClient(); await Task.Delay(500); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult()); await server.StopAsync(); await Task.Delay(500); - Assert.IsFalse(client.IsConnected); + Assert.IsFalse(client.TryPingAsync().GetAwaiter().GetResult()); for (var i = 0; i < 5; i++) { @@ -671,6 +675,7 @@ namespace MQTTnet.Tests.Client } catch { + // Ignore errors. } } @@ -678,7 +683,7 @@ namespace MQTTnet.Tests.Client await Task.Delay(500); await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build()); - Assert.IsTrue(client.IsConnected); + await client.PingAsync(); } } @@ -828,10 +833,10 @@ namespace MQTTnet.Tests.Client { using (var testEnvironment = new TestEnvironment(TestContext)) { - var server = await testEnvironment.StartServer(); + await testEnvironment.StartServer(); var client = await testEnvironment.ConnectClient(); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult()); client.DisconnectedAsync += e => { Assert.IsTrue(e.ClientWasConnected); @@ -851,7 +856,7 @@ namespace MQTTnet.Tests.Client var server = await testEnvironment.StartServer(); var client = await testEnvironment.ConnectClient(); - Assert.IsTrue(client.IsConnected); + Assert.IsTrue(client.TryPingAsync().GetAwaiter().GetResult()); client.DisconnectedAsync += e => { Assert.IsTrue(e.ClientWasConnected); @@ -941,7 +946,7 @@ namespace MQTTnet.Tests.Client await Task.Delay(500); Assert.IsTrue(messageReceived); - Assert.IsTrue(client1.IsConnected); + Assert.IsTrue(client1.TryPingAsync().GetAwaiter().GetResult()); Assert.IsFalse(disconnectedFired); } } diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index a9ded3a9..ada0ad0f 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -7,32 +7,31 @@ namespace MQTTnet.Client { public interface IMqttClient : IDisposable { - event Func ConnectedAsync; - - event Func ConnectingAsync; - - event Func DisconnectedAsync; - event Func ApplicationMessageReceivedAsync; - + event Func ConnectedAsync; + + event Func ConnectingAsync; + + event Func DisconnectedAsync; + event Func InspectPackage; - + bool IsConnected { get; } - + MqttClientOptions Options { get; } - + Task ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default); - + Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default); - + Task PingAsync(CancellationToken cancellationToken = default); - - Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default); - - Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default); - - Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default); - + Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default); + + Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default); + + Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default); + + Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index fc0d299c..52b774b2 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -2,105 +2,113 @@ // 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.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; using MQTTnet.Protocol; -using System; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Diagnostics; -using MQTTnet.Formatter; -using MQTTnet.Implementations; namespace MQTTnet.Client { public sealed class MqttClient : Disposable, IMqttClient { - readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); - + readonly IMqttClientAdapterFactory _adapterFactory; + readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); + readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); + readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory(); readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory(); - readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); - - readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); - readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); - readonly object _disconnectLock = new object(); readonly AsyncEvent _connectedEvent = new AsyncEvent(); readonly AsyncEvent _connectingEvent = new AsyncEvent(); readonly AsyncEvent _disconnectedEvent = new AsyncEvent(); - readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); + readonly object _disconnectLock = new object(); readonly AsyncEvent _inspectPacketEvent = new AsyncEvent(); - - readonly IMqttClientAdapterFactory _adapterFactory; - readonly IMqttNetLogger _rootLogger; readonly MqttNetSourceLogger _logger; + readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); + readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); - CancellationTokenSource _backgroundCancellationTokenSource; - Task _packetReceiverTask; - Task _keepAlivePacketsSenderTask; - Task _publishPacketReceiverTask; - - AsyncQueue _publishPacketReceiverQueue; + readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); + readonly IMqttNetLogger _rootLogger; IMqttChannelAdapter _adapter; + + CancellationTokenSource _backgroundCancellationTokenSource; bool _cleanDisconnectInitiated; volatile int _connectionStatus; MqttClientDisconnectReason _disconnectReason; + string _disconnectReasonString; + Task _keepAlivePacketsSenderTask; DateTime _lastPacketSentTimestamp; - string _disconnectReasonString; - + Task _packetReceiverTask; + + AsyncQueue _publishPacketReceiverQueue; + Task _publishPacketReceiverTask; + public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) { _adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory)); _rootLogger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger.WithSource(nameof(MqttClient)); } - - public event Func ConnectedAsync - { - add => _connectedEvent.AddHandler(value); - remove => _connectedEvent.RemoveHandler(value); - } - - public event Func ConnectingAsync - { - add => _connectingEvent.AddHandler(value); - remove => _connectingEvent.RemoveHandler(value); - } - - public event Func DisconnectedAsync - { - add => _disconnectedEvent.AddHandler(value); - remove => _disconnectedEvent.RemoveHandler(value); - } - + public event Func ApplicationMessageReceivedAsync { add => _applicationMessageReceivedEvent.AddHandler(value); remove => _applicationMessageReceivedEvent.RemoveHandler(value); } - + + public event Func ConnectedAsync + { + add => _connectedEvent.AddHandler(value); + remove => _connectedEvent.RemoveHandler(value); + } + + public event Func ConnectingAsync + { + add => _connectingEvent.AddHandler(value); + remove => _connectingEvent.RemoveHandler(value); + } + + public event Func DisconnectedAsync + { + add => _disconnectedEvent.AddHandler(value); + remove => _disconnectedEvent.RemoveHandler(value); + } + public event Func InspectPackage { add => _inspectPacketEvent.AddHandler(value); remove => _inspectPacketEvent.RemoveHandler(value); } - + + [Obsolete( + "This property will be removed in the future. Checking for a working connection should be done via calling _PinAsync_. Also other successful traffic can be used to indicate connection status.")] public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected; public MqttClientOptions Options { get; private set; } public async Task ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default) { - if (options == null) throw new ArgumentNullException(nameof(options)); - if (options.ChannelOptions == null) throw new ArgumentException("ChannelOptions are not set."); + if (options == null) + { + throw new ArgumentNullException(nameof(options)); + } + + if (options.ChannelOptions == null) + { + throw new ArgumentException("ChannelOptions are not set."); + } ThrowIfConnected("It is not allowed to connect with a server after the connection is established."); @@ -154,7 +162,7 @@ namespace MQTTnet.Client _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(backgroundCancellationToken), backgroundCancellationToken); @@ -186,7 +194,10 @@ namespace MQTTnet.Client public async Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default) { - if (options is null) throw new ArgumentNullException(nameof(options)); + if (options is null) + { + throw new ArgumentNullException(nameof(options)); + } ThrowIfDisposed(); @@ -213,63 +224,19 @@ namespace MQTTnet.Client } } - public Task PingAsync(CancellationToken cancellationToken = default) + public async Task PingAsync(CancellationToken cancellationToken = default) { - return SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken); - } - - public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default) - { - if (data == null) throw new ArgumentNullException(nameof(data)); - - ThrowIfDisposed(); - ThrowIfNotConnected(); - - var authPacket = new MqttAuthPacket + if (!cancellationToken.CanBeCanceled) { - // This must always be equal to the value from the CONNECT packet. So we use it here to ensure that. - AuthenticationMethod = Options.AuthenticationMethod, - AuthenticationData = data.AuthenticationData, - ReasonString = data.ReasonString, - UserProperties = data.UserProperties - }; - - return SendAsync(authPacket, cancellationToken); - } - - public async Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - foreach (var topicFilter in options.TopicFilters) - { - MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic); + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + await SendAndReceiveAsync(MqttPingReqPacket.Instance, timeout.Token).ConfigureAwait(false); + } + } + else + { + await SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false); } - - ThrowIfDisposed(); - ThrowIfNotConnected(); - - var subscribePacket = _packetFactories.Subscribe.Create(options); - subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - - var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); - - return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket); - } - - public async Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - ThrowIfDisposed(); - ThrowIfNotConnected(); - - var unsubscribePacket = _packetFactories.Unsubscribe.Create(options); - unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - - var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); - - return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket); } public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default) @@ -304,22 +271,67 @@ namespace MQTTnet.Client } } - void Cleanup() + public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default) { - try + if (data == null) { - _backgroundCancellationTokenSource?.Cancel(false); + throw new ArgumentNullException(nameof(data)); } - finally + + ThrowIfDisposed(); + ThrowIfNotConnected(); + + var authPacket = new MqttAuthPacket { - _backgroundCancellationTokenSource?.Dispose(); - _backgroundCancellationTokenSource = null; + // This must always be equal to the value from the CONNECT packet. So we use it here to ensure that. + AuthenticationMethod = Options.AuthenticationMethod, + AuthenticationData = data.AuthenticationData, + ReasonString = data.ReasonString, + UserProperties = data.UserProperties + }; - _publishPacketReceiverQueue?.Dispose(); - _publishPacketReceiverQueue = null; + return SendAsync(authPacket, cancellationToken); + } - _adapter?.Dispose(); + public async Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default) + { + if (options == null) + { + throw new ArgumentNullException(nameof(options)); } + + foreach (var topicFilter in options.TopicFilters) + { + MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic); + } + + ThrowIfDisposed(); + ThrowIfNotConnected(); + + var subscribePacket = _packetFactories.Subscribe.Create(options); + subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + + var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); + + return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket); + } + + public async Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default) + { + if (options == null) + { + throw new ArgumentNullException(nameof(options)); + } + + ThrowIfDisposed(); + ThrowIfNotConnected(); + + var unsubscribePacket = _packetFactories.Unsubscribe.Create(options); + unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + + var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); + + return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket); } protected override void Dispose(bool disposing) @@ -332,6 +344,36 @@ namespace MQTTnet.Client base.Dispose(disposing); } + Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken) + { + if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + { + // no response required + } + else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + { + if (!eventArgs.ProcessingFailed) + { + var pubAckPacket = _packetFactories.PubAck.Create(eventArgs); + return SendAsync(pubAckPacket, cancellationToken); + } + } + else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + { + if (!eventArgs.ProcessingFailed) + { + var pubRecPacket = _packetFactories.PubRec.Create(eventArgs); + return SendAsync(pubRecPacket, cancellationToken); + } + } + else + { + throw new MqttProtocolViolationException("Received a not supported QoS level."); + } + + return PlatformAbstractionLayer.CompletedTask; + } + async Task AuthenticateAsync(MqttClientOptions options, CancellationToken cancellationToken) { MqttClientConnectResult result; @@ -360,32 +402,27 @@ namespace MQTTnet.Client return result; } - void ThrowIfNotConnected() + void Cleanup() { - if (!IsConnected) + try { - throw new MqttCommunicationException("The client is not connected."); + _backgroundCancellationTokenSource?.Cancel(false); + } + finally + { + _backgroundCancellationTokenSource?.Dispose(); + _backgroundCancellationTokenSource = null; + + _publishPacketReceiverQueue?.Dispose(); + _publishPacketReceiverQueue = null; + + _adapter?.Dispose(); } } - void ThrowIfConnected(string message) + MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand) { - if (IsConnected) - { - throw new MqttProtocolViolationException(message); - } - } - - Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult) - { - var clientWasConnected = IsConnected; - - if (!DisconnectIsPendingOrFinished()) - { - return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected); - } - - return PlatformAbstractionLayer.CompletedTask; + return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand); } async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected) @@ -446,28 +483,166 @@ namespace MQTTnet.Client } } - void TryInitiateDisconnect() + Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult) { - lock (_disconnectLock) + var clientWasConnected = IsConnected; + + if (!DisconnectIsPendingOrFinished()) + { + return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + bool DisconnectIsPendingOrFinished() + { + var connectionStatus = (MqttClientConnectionStatus)_connectionStatus; + + do + { + switch (connectionStatus) + { + case MqttClientConnectionStatus.Disconnected: + case MqttClientConnectionStatus.Disconnecting: + return true; + case MqttClientConnectionStatus.Connected: + case MqttClientConnectionStatus.Connecting: + // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards. + // So the first caller will get a "false" and all subsequent ones will get "true". + var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus); + if (curStatus == connectionStatus) + { + return false; + } + + connectionStatus = curStatus; + break; + } + } while (true); + } + + void EnqueueReceivedPublishPacket(MqttPublishPacket publishPacket) + { + try + { + _publishPacketReceiverQueue.Enqueue(publishPacket); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while queueing application message."); + } + } + + async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) + { + var applicationMessage = _applicationMessageFactory.Create(publishPacket); + var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket); + + await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + + return eventArgs; + } + + Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket) + { + var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; + if (extendedAuthenticationExchangeHandler != null) + { + return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) + { + _disconnectReason = (MqttClientDisconnectReason)disconnectPacket.ReasonCode; + _disconnectReasonString = disconnectPacket.ReasonString; + + // Also dispatch disconnect to waiting threads to generate a proper exception. + _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); + + return DisconnectInternalAsync(_packetReceiverTask, null, null); + } + + async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) { try { - _backgroundCancellationTokenSource?.Cancel(false); + var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false); + if (!publishPacketDequeueResult.IsSuccess) + { + return; + } + + var publishPacket = publishPacketDequeueResult.Item; + var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); + + if (eventArgs.AutoAcknowledge) + { + await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { } catch (Exception exception) { - _logger.Warning(exception, "Error while initiating disconnect."); + _logger.Error(exception, "Error while handling application message."); } } } - Task SendAsync(MqttPacket packet, CancellationToken cancellationToken) + Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken) { - cancellationToken.ThrowIfCancellationRequested(); + if (!_packetDispatcher.TryDispatch(pubRecPacket)) + { + // The packet is unknown. Probably due to a restart of the client. + // So wen send this to the server to trigger a full resend of the message. + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); + return SendAsync(pubRelPacket, cancellationToken); + } - _lastPacketSentTimestamp = DateTime.UtcNow; + return PlatformAbstractionLayer.CompletedTask; + } - return _adapter.SendPacketAsync(packet, cancellationToken); + Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) + { + var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + return SendAsync(pubCompPacket, cancellationToken); + } + + async Task PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + + var pubAckPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); + + return _clientPublishResultFactory.Create(pubAckPacket); + } + + async Task PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] + await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); + + return _clientPublishResultFactory.Create(null); + } + + async Task PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); + + var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); + + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + + var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + + return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); } async Task SendAndReceiveAsync(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket @@ -508,124 +683,43 @@ namespace MQTTnet.Client } } - async Task TrySendKeepAliveMessagesAsync(CancellationToken cancellationToken) + Task SendAsync(MqttPacket packet, CancellationToken cancellationToken) { - try + cancellationToken.ThrowIfCancellationRequested(); + + _lastPacketSentTimestamp = DateTime.UtcNow; + + return _adapter.SendPacketAsync(packet, cancellationToken); + } + + void ThrowIfConnected(string message) + { + if (IsConnected) { - _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."); + throw new MqttProtocolViolationException(message); } } - async Task TryReceivePacketsAsync(CancellationToken cancellationToken) + void ThrowIfNotConnected() { - try + if (!IsConnected) { - _logger.Verbose("Start receiving packets."); - - while (!cancellationToken.IsCancellationRequested) - { - MqttPacket packet; - var packetTask = _adapter.ReceivePacketAsync(cancellationToken); - - if (packetTask.IsCompleted) - { - packet = packetTask.Result; - } - else - { - packet = await packetTask.ConfigureAwait(false); - } - - if (cancellationToken.IsCancellationRequested) - { - return; - } - - if (packet == null) - { - await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false); - - return; - } - - await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); - } + throw new MqttCommunicationException("The client is not connected."); } - catch (Exception exception) + } + + void TryInitiateDisconnect() + { + lock (_disconnectLock) { - if (_cleanDisconnectInitiated) + try { - return; + _backgroundCancellationTokenSource?.Cancel(false); } - - if (exception is OperationCanceledException) + catch (Exception exception) { + _logger.Warning(exception, "Error while initiating disconnect."); } - else if (exception is MqttCommunicationException) - { - _logger.Warning(exception, "Communication error while receiving packets."); - } - else - { - _logger.Error(exception, "Error while receiving packets."); - } - - _packetDispatcher.FailAll(exception); - - await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false); - } - finally - { - _logger.Verbose("Stopped receiving packets."); } } @@ -694,158 +788,125 @@ namespace MQTTnet.Client } } - void EnqueueReceivedPublishPacket(MqttPublishPacket publishPacket) + async Task TryReceivePacketsAsync(CancellationToken cancellationToken) { try { - _publishPacketReceiverQueue.Enqueue(publishPacket); - } - catch (Exception exception) - { - _logger.Error(exception, "Error while queueing application message."); - } - } + _logger.Verbose("Start receiving packets."); - async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken) - { - while (!cancellationToken.IsCancellationRequested) - { - try + while (!cancellationToken.IsCancellationRequested) { - var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false); - if (!publishPacketDequeueResult.IsSuccess) + MqttPacket packet; + var packetTask = _adapter.ReceivePacketAsync(cancellationToken); + + if (packetTask.IsCompleted) + { + packet = packetTask.Result; + } + else + { + packet = await packetTask.ConfigureAwait(false); + } + + if (cancellationToken.IsCancellationRequested) { return; } - var publishPacket = publishPacketDequeueResult.Item; - var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); - - if (eventArgs.AutoAcknowledge) + if (packet == null) { - await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false); + await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false); + + return; } + + await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); } - catch (OperationCanceledException) + } + catch (Exception exception) + { + if (_cleanDisconnectInitiated) + { + return; + } + + if (exception is OperationCanceledException) { } - catch (Exception exception) + else if (exception is MqttCommunicationException) { - _logger.Error(exception, "Error while handling application message."); + _logger.Warning(exception, "Communication error while receiving packets."); + } + else + { + _logger.Error(exception, "Error while receiving packets."); + } + + _packetDispatcher.FailAll(exception); + + await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false); + } + finally + { + _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); } } - } - - Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken) - { - if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + catch (Exception exception) { - // no response required - } - else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) - { - if (!eventArgs.ProcessingFailed) + if (_cleanDisconnectInitiated) { - var pubAckPacket = _packetFactories.PubAck.Create(eventArgs); - return SendAsync(pubAckPacket, cancellationToken); + return; } - } - else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) - { - if (!eventArgs.ProcessingFailed) + + if (exception is OperationCanceledException) { - var pubRecPacket = _packetFactories.PubRec.Create(eventArgs); - return SendAsync(pubRecPacket, cancellationToken); + 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); } - else + finally { - throw new MqttProtocolViolationException("Received a not supported QoS level."); + _logger.Verbose("Stopped sending keep alive packets."); } - - return PlatformAbstractionLayer.CompletedTask; - } - - Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken) - { - if (!_packetDispatcher.TryDispatch(pubRecPacket)) - { - // The packet is unknown. Probably due to a restart of the client. - // So wen send this to the server to trigger a full resend of the message. - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); - return SendAsync(pubRelPacket, cancellationToken); - } - - return PlatformAbstractionLayer.CompletedTask; - } - - Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) - { - var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - return SendAsync(pubCompPacket, cancellationToken); - } - - Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) - { - _disconnectReason = (MqttClientDisconnectReason) disconnectPacket.ReasonCode; - _disconnectReasonString = disconnectPacket.ReasonString; - - // Also dispatch disconnect to waiting threads to generate a proper exception. - _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); - - return DisconnectInternalAsync(_packetReceiverTask, null, null); - } - - Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket) - { - var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; - if (extendedAuthenticationExchangeHandler != null) - { - return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)); - } - - return PlatformAbstractionLayer.CompletedTask; - } - - async Task PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(publishPacket, cancellationToken) - .ConfigureAwait(false); - - return _clientPublishResultFactory.Create(null); - } - - async Task PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - - var pubAckPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - - return _clientPublishResultFactory.Create(pubAckPacket); - } - - async Task PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - - var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - - var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); - - return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); - } - - async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) - { - var applicationMessage = _applicationMessageFactory.Create(publishPacket); - var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket); - - await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); - - return eventArgs; } async Task WaitForTaskAsync(Task task, Task sender) @@ -878,37 +939,5 @@ namespace MQTTnet.Client { } } - - bool DisconnectIsPendingOrFinished() - { - var connectionStatus = (MqttClientConnectionStatus)_connectionStatus; - - do - { - switch (connectionStatus) - { - case MqttClientConnectionStatus.Disconnected: - case MqttClientConnectionStatus.Disconnecting: - return true; - case MqttClientConnectionStatus.Connected: - case MqttClientConnectionStatus.Connecting: - // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards. - // So the first caller will get a "false" and all subsequent ones will get "true". - var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus); - if (curStatus == connectionStatus) - { - return false; - } - - connectionStatus = curStatus; - break; - } - } while (true); - } - - MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand) - { - return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand); - } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index 8e8ffb65..9f936bdd 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -14,6 +14,26 @@ namespace MQTTnet.Client { public static class MqttClientExtensions { + public static async Task TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default) + { + if (client == null) + { + throw new ArgumentNullException(nameof(client)); + } + + try + { + await client.PingAsync(cancellationToken).ConfigureAwait(false); + return true; + } + catch + { + // Ignore errors. + } + + return false; + } + public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null) { if (client == null)