From c2ac4fffaf67609665d44e96036e4eefb668153c Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 18 Apr 2020 01:38:48 +0200 Subject: [PATCH] Expose PING method for clients. --- .../IManagedMqttClient.cs | 14 +++++----- .../ManagedMqttClient.cs | 13 +++++++--- .../ManagedMqttClientExtensions.cs | 16 ++++++------ Source/MQTTnet/Client/IMqttClient.cs | 13 +++++----- Source/MQTTnet/Client/MqttClient.cs | 26 ++++++++++++++----- Source/MQTTnet/Client/MqttClientExtensions.cs | 6 ++--- .../Mockups/TestClientWrapper.cs | 5 ++++ 7 files changed, 59 insertions(+), 34 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index 9919081f..285ceda8 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -1,8 +1,9 @@ -using System; -using System.Collections.Generic; -using System.Threading.Tasks; -using MQTTnet.Client.Connecting; +using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Extensions.ManagedClient { @@ -21,11 +22,12 @@ namespace MQTTnet.Extensions.ManagedClient IConnectingFailedHandler ConnectingFailedHandler { get; set; } ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; } - + Task StartAsync(IManagedMqttClientOptions options); Task StopAsync(); + Task PingAsync(CancellationToken cancellationToken); - Task SubscribeAsync(IEnumerable topicFilters); + Task SubscribeAsync(IEnumerable topicFilters); Task UnsubscribeAsync(IEnumerable topics); Task PublishAsync(ManagedMqttApplicationMessage applicationMessages); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 10d7d77c..43f76d99 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -129,6 +129,11 @@ namespace MQTTnet.Extensions.ManagedClient } } + public Task PingAsync(CancellationToken cancellationToken) + { + return _mqttClient.PingAsync(cancellationToken); + } + public async Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -199,7 +204,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - public Task SubscribeAsync(IEnumerable topicFilters) + public Task SubscribeAsync(IEnumerable topicFilters) { ThrowIfDisposed(); @@ -449,12 +454,12 @@ namespace MQTTnet.Extensions.ManagedClient var endTime = DateTime.UtcNow + timeout; while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false)) { - List subscriptions; + List subscriptions; HashSet unsubscriptions; lock (_subscriptions) { - subscriptions = _subscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList(); + subscriptions = _subscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList(); _subscriptions.Clear(); unsubscriptions = new HashSet(_unsubscriptions); _unsubscriptions.Clear(); @@ -504,7 +509,7 @@ namespace MQTTnet.Extensions.ManagedClient { if (_reconnectSubscriptions.Any()) { - var subscriptions = _reconnectSubscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }); + var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }); await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs index b6387918..33da56e6 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs @@ -1,12 +1,12 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Client.Connecting; +using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Protocol; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Extensions.ManagedClient { @@ -108,7 +108,7 @@ namespace MQTTnet.Extensions.ManagedClient return client; } - public static Task SubscribeAsync(this IManagedMqttClient client, params TopicFilter[] topicFilters) + public static Task SubscribeAsync(this IManagedMqttClient client, params MqttTopicFilter[] topicFilters) { if (client == null) throw new ArgumentNullException(nameof(client)); @@ -120,7 +120,7 @@ namespace MQTTnet.Extensions.ManagedClient if (client == null) throw new ArgumentNullException(nameof(client)); if (topic == null) throw new ArgumentNullException(nameof(topic)); - return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); + return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); } public static Task SubscribeAsync(this IManagedMqttClient client, string topic) @@ -128,7 +128,7 @@ namespace MQTTnet.Extensions.ManagedClient if (client == null) throw new ArgumentNullException(nameof(client)); if (topic == null) throw new ArgumentNullException(nameof(topic)); - return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build()); + return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).Build()); } public static Task UnsubscribeAsync(this IManagedMqttClient client, params string[] topicFilters) diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index 059f2e60..58cfa0c1 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -1,12 +1,12 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Client.Connecting; +using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.ExtendedAuthenticationExchange; using MQTTnet.Client.Options; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; +using System; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Client { @@ -16,11 +16,12 @@ namespace MQTTnet.Client IMqttClientOptions Options { get; } IMqttClientConnectedHandler ConnectedHandler { get; set; } - + IMqttClientDisconnectedHandler DisconnectedHandler { get; set; } - + Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken); Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken); + Task PingAsync(CancellationToken cancellationToken); Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken); Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken); diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 3f812320..52929eff 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -40,7 +40,8 @@ namespace MQTTnet.Client IMqttChannelAdapter _adapter; bool _cleanDisconnectInitiated; - long _disconnectGate; + long _isDisconnectPending; + bool _isConnected; public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) { @@ -56,7 +57,13 @@ namespace MQTTnet.Client public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } - public bool IsConnected { get; private set; } + public bool IsConnected + { + get + { + return _isConnected || Interlocked.Read(ref _isDisconnectPending) != 0; + } + } public IMqttClientOptions Options { get; private set; } @@ -81,7 +88,7 @@ namespace MQTTnet.Client _backgroundCancellationTokenSource = new CancellationTokenSource(); var backgroundCancellationToken = _backgroundCancellationTokenSource.Token; - _disconnectGate = 0; + _isDisconnectPending = 0; var adapter = _adapterFactory.CreateClientAdapter(options, _logger); _adapter = adapter; @@ -107,7 +114,7 @@ namespace MQTTnet.Client _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken); } - IsConnected = true; + _isConnected = true; _logger.Info("Connected."); @@ -157,6 +164,11 @@ namespace MQTTnet.Client } } + public async Task PingAsync(CancellationToken cancellationToken) + { + await SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false); + } + public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken) { if (data == null) throw new ArgumentNullException(nameof(data)); @@ -283,7 +295,7 @@ namespace MQTTnet.Client void ThrowIfNotConnected() { - if (!IsConnected || Interlocked.Read(ref _disconnectGate) == 1) throw new MqttCommunicationException("The client is not connected."); + if (!IsConnected || Interlocked.Read(ref _isDisconnectPending) == 1) throw new MqttCommunicationException("The client is not connected."); } void ThrowIfConnected(string message) @@ -296,7 +308,7 @@ namespace MQTTnet.Client var clientWasConnected = IsConnected; TryInitiateDisconnect(); - IsConnected = false; + _isConnected = false; try { @@ -749,7 +761,7 @@ namespace MQTTnet.Client bool DisconnectIsPending() { - return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0; + return Interlocked.CompareExchange(ref _isDisconnectPending, 1, 0) != 0; } } } diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index aed3e45c..2408a14a 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -131,7 +131,7 @@ namespace MQTTnet.Client return client.DisconnectAsync(new MqttClientDisconnectOptions()); } - public static Task SubscribeAsync(this IMqttClient client, params TopicFilter[] topicFilters) + public static Task SubscribeAsync(this IMqttClient client, params MqttTopicFilter[] topicFilters) { if (client == null) throw new ArgumentNullException(nameof(client)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); @@ -147,7 +147,7 @@ namespace MQTTnet.Client if (client == null) throw new ArgumentNullException(nameof(client)); if (topic == null) throw new ArgumentNullException(nameof(topic)); - return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); + return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); } public static Task SubscribeAsync(this IMqttClient client, string topic) @@ -155,7 +155,7 @@ namespace MQTTnet.Client if (client == null) throw new ArgumentNullException(nameof(client)); if (topic == null) throw new ArgumentNullException(nameof(topic)); - return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build()); + return client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topic).Build()); } public static Task UnsubscribeAsync(this IMqttClient client, params string[] topicFilters) diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs index e52133c3..5fd449db 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs @@ -73,6 +73,11 @@ namespace MQTTnet.Tests.Mockups Implementation.Dispose(); } + public Task PingAsync(CancellationToken cancellationToken) + { + return Implementation.PingAsync(cancellationToken); + } + public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) { return Implementation.PublishAsync(applicationMessage, cancellationToken);