Merge branch 'master' of https://github.com/chkr1011/MQTTnet

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -215,6 +215,11 @@ namespace MQTTnet.Extensions.ManagedClient
 
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
+            foreach (var topicFilter in topicFilters)
+            {
+                MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
+            }
+
             lock (_subscriptions)
             {
                 foreach (var topicFilter in topicFilters)

Source/MQTTnet/Client/MqttClient.cs

@@ -194,6 +194,11 @@ namespace MQTTnet.Client
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
+            foreach (var topicFilter in options.TopicFilters)
+            {
+                MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
+            }
+
             ThrowIfDisposed();
             ThrowIfNotConnected();
 

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Client.Connecting;
+using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.ExtendedAuthenticationExchange;
 using MQTTnet.Client.Options;
@@ -112,7 +112,7 @@ namespace MQTTnet.Client
             return client;
         }
 
-        public static Task ReconnectAsync(this IMqttClient client)
+        public static Task<MqttClientAuthenticateResult> ReconnectAsync(this IMqttClient client)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));
 
@@ -124,6 +124,18 @@ namespace MQTTnet.Client
             return client.ConnectAsync(client.Options);
         }
 
+        public static Task<MqttClientAuthenticateResult> ReconnectAsync(this IMqttClient client, CancellationToken cancellationToken)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            if (client.Options == null)
+            {
+                throw new InvalidOperationException("_ReconnectAsync_ can be used only if _ConnectAsync_ was called before.");
+            }
+
+            return client.ConnectAsync(client.Options, cancellationToken);
+        }
+
         public static Task DisconnectAsync(this IMqttClient client)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));

Source/MQTTnet/Protocol/MqttTopicValidator.cs

@@ -42,5 +42,15 @@ namespace MQTTnet.Protocol
                 }
             }
         }
+
+        public static void ThrowIfInvalidSubscribe(string topic)
+        {
+            if (string.IsNullOrEmpty(topic))
+            {
+                throw new MqttProtocolViolationException("Topic should not be empty.");
+            }
+
+            if (topic.IndexOf("#") != -1 && topic.IndexOf("#") != topic.Length - 1) throw new MqttProtocolViolationException("The character '#' is only allowed as last character");
+        }
     }
 }

Source/MQTTnet/Server/MqttServer.cs

@@ -113,7 +113,12 @@ namespace MQTTnet.Server
         {
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-            
+
+            foreach (var topicFilter in topicFilters)
+            {
+                MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic);
+            }
+
             ThrowIfDisposed();
             ThrowIfNotStarted();