Expose PING method for clients.

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -1,8 +1,9 @@
-using System;
+using MQTTnet.Client.Connecting;
-using System.Collections.Generic;
-using System.Threading.Tasks;
-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<TopicFilter> topicFilters);
+        Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters);
         Task UnsubscribeAsync(IEnumerable<string> topics);
 
         Task PublishAsync(ManagedMqttApplicationMessage applicationMessages);

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<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
@@ -199,7 +204,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
+        public Task SubscribeAsync(IEnumerable<MqttTopicFilter> 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<TopicFilter> subscriptions;
+                List<MqttTopicFilter> subscriptions;
                 HashSet<string> 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<string>(_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);
                 }
             }

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs

@@ -1,12 +1,12 @@
-using System;
+using MQTTnet.Client.Connecting;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-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)

Source/MQTTnet/Client/IMqttClient.cs

@@ -1,12 +1,12 @@
-using System;
+using MQTTnet.Client.Connecting;
-using System.Threading;
-using System.Threading.Tasks;
-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<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);
         Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken);
+        Task PingAsync(CancellationToken cancellationToken);
 
         Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken);
         Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken);

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<MqttPingRespPacket>(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;
         }
     }
 }

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -131,7 +131,7 @@ namespace MQTTnet.Client
             return client.DisconnectAsync(new MqttClientDisconnectOptions());
         }
 
-        public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, params TopicFilter[] topicFilters)
+        public static Task<MqttClientSubscribeResult> 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<MqttClientSubscribeResult> 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<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient client, params string[] topicFilters)

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<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
         {
             return Implementation.PublishAsync(applicationMessage, cancellationToken);