Merge branch 'master' into feature/Single_Connection_Monitoring_Task

Build/MQTTnet.nuspec

@@ -13,6 +13,8 @@
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
     <releaseNotes>
 * [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal).
+* [RpcClient] Moved the RPC topic validation to the topic generation strategy.
+* [RcpClient] Added interface for _MqttRpcClient_.
 * [Server] Reduced async tasks count by moving dedicated keep alive tasks per connection to shared one.
 * [Server] Session takeover and keep alive timeout are now properly set in DISCONNECT packet.
 * [Server] Performance improvements.

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Extensions.Rpc
+{
+    public interface IMqttRpcClient : IDisposable
+    {
+        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
+
+        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
+
+        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
+
+        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
+    }
+}

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -11,12 +11,12 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Extensions.Rpc
 {
-    public sealed class MqttRpcClient : IDisposable
+    public sealed class MqttRpcClient : IMqttRpcClient
     {
-        private readonly ConcurrentDictionary<string, TaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, TaskCompletionSource<byte[]>>();
-        private readonly IMqttClient _mqttClient;
-        private readonly IMqttRpcClientOptions _options;
-        private readonly RpcAwareApplicationMessageReceivedHandler _applicationMessageReceivedHandler;
+        readonly ConcurrentDictionary<string, TaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, TaskCompletionSource<byte[]>>();
+        readonly IMqttClient _mqttClient;
+        readonly IMqttRpcClientOptions _options;
+        readonly RpcAwareApplicationMessageReceivedHandler _applicationMessageReceivedHandler;
 
         [Obsolete("Use MqttRpcClient(IMqttClient mqttClient, IMqttRpcClientOptions options).")]
         public MqttRpcClient(IMqttClient mqttClient) : this(mqttClient, new MqttRpcClientOptions())
@@ -54,11 +54,6 @@ namespace MQTTnet.Extensions.Rpc
         {
             if (methodName == null) throw new ArgumentNullException(nameof(methodName));
 
-            if (methodName.Contains("/") || methodName.Contains("+") || methodName.Contains("#"))
-            {
-                throw new ArgumentException("The method name cannot contain /, + or #.");
-            }
-
             if (!(_mqttClient.ApplicationMessageReceivedHandler is RpcAwareApplicationMessageReceivedHandler))
             {
                 throw new InvalidOperationException("The application message received handler was modified.");
@@ -139,7 +134,7 @@ namespace MQTTnet.Extensions.Rpc
             }
         }
 
-        private Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
+        Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
             if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var tcs))
             {

Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptions.cs

@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Extensions.Rpc.Options
 {
-    public class MqttRpcClientOptions : IMqttRpcClientOptions
+    public sealed class MqttRpcClientOptions : IMqttRpcClientOptions
     {
         public IMqttRpcClientTopicGenerationStrategy TopicGenerationStrategy { get; set; } = new DefaultMqttRpcClientTopicGenerationStrategy();
     }

Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs

@@ -3,7 +3,7 @@ using System;
 
 namespace MQTTnet.Extensions.Rpc.Options
 {
-    public class MqttRpcClientOptionsBuilder
+    public sealed class MqttRpcClientOptionsBuilder
     {
         IMqttRpcClientTopicGenerationStrategy _topicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy();
 

Source/MQTTnet.Extensions.Rpc/Options/MqttRpcTopicPair.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Extensions.Rpc.Options
 {
-    public class MqttRpcTopicPair
+    public sealed class MqttRpcTopicPair
     {
         public string RequestTopic { get; set; }
 

Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs

@@ -2,10 +2,15 @@
 
 namespace MQTTnet.Extensions.Rpc.Options.TopicGeneration
 {
-    public class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
+    public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
     {
         public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
         {
+            if (context.MethodName.Contains("/") || context.MethodName.Contains("+") || context.MethodName.Contains("#"))
+            {
+                throw new ArgumentException("The method name cannot contain /, + or #.");
+            }
+            
             var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{context.MethodName}";
             var responseTopic = requestTopic + "/response";
 

Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs

@@ -3,7 +3,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc.Options.TopicGeneration
 {
-    public class TopicGenerationContext
+    public sealed class TopicGenerationContext
     {
         public string MethodName { get; set; }
 

Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs

@@ -4,9 +4,9 @@ using MQTTnet.Client.Receiving;
 
 namespace MQTTnet.Extensions.Rpc
 {
-    public class RpcAwareApplicationMessageReceivedHandler : IMqttApplicationMessageReceivedHandler
+    public sealed class RpcAwareApplicationMessageReceivedHandler : IMqttApplicationMessageReceivedHandler
     {
-        private readonly Func<MqttApplicationMessageReceivedEventArgs, Task> _handleReceivedApplicationMessageAsync;
+        readonly Func<MqttApplicationMessageReceivedEventArgs, Task> _handleReceivedApplicationMessageAsync;
 
         public RpcAwareApplicationMessageReceivedHandler(
             IMqttApplicationMessageReceivedHandler originalHandler,
@@ -18,14 +18,14 @@ namespace MQTTnet.Extensions.Rpc
 
         public IMqttApplicationMessageReceivedHandler OriginalHandler { get; }
 
-        public async Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
+        public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
             if (OriginalHandler != null)
             {
-                await OriginalHandler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
+                return OriginalHandler.HandleApplicationMessageReceivedAsync(eventArgs);
             }
 
-            await _handleReceivedApplicationMessageAsync(eventArgs).ConfigureAwait(false);
+            return _handleReceivedApplicationMessageAsync(eventArgs);
         }
     }
 }