From 2716db04e318a0526fe0fb2c38422ca40b5287d1 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 1 Nov 2020 12:00:12 +0100 Subject: [PATCH 1/2] Move the RPC topic validation to the topic generation strategy. --- .../MQTTnet.Extensions.Rpc/IMqttRpcClient.cs | 18 ++++++++++++++++++ Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 17 ++++++----------- .../Options/MqttRpcClientOptions.cs | 2 +- .../Options/MqttRpcClientOptionsBuilder.cs | 2 +- .../Options/MqttRpcTopicPair.cs | 2 +- ...aultMqttRpcClientTopicGenerationStrategy.cs | 7 ++++++- .../TopicGeneration/TopicGenerationContext.cs | 2 +- ...pcAwareApplicationMessageReceivedHandler.cs | 10 +++++----- 8 files changed, 39 insertions(+), 21 deletions(-) create mode 100644 Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs diff --git a/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs new file mode 100644 index 00000000..a3a06fa2 --- /dev/null +++ b/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 ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel); + + Task ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken); + + Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel); + + Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 6a6a93e7..c225230e 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/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> _waitingCalls = new ConcurrentDictionary>(); - private readonly IMqttClient _mqttClient; - private readonly IMqttRpcClientOptions _options; - private readonly RpcAwareApplicationMessageReceivedHandler _applicationMessageReceivedHandler; + readonly ConcurrentDictionary> _waitingCalls = new ConcurrentDictionary>(); + 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)) { diff --git a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptions.cs b/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptions.cs index 0dc45175..617653ab 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptions.cs +++ b/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(); } diff --git a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs index 69277ac1..9f0e986f 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs +++ b/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(); diff --git a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcTopicPair.cs b/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcTopicPair.cs index 12bccce5..56f768ae 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcTopicPair.cs +++ b/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; } diff --git a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs index 6d1f4bc7..39bbde7f 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs +++ b/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"; diff --git a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs index dc7263f9..cd82c52a 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs +++ b/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; } diff --git a/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs b/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs index 9d90fb19..d7b75fc9 100644 --- a/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs +++ b/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 _handleReceivedApplicationMessageAsync; + readonly Func _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); } } } From 1a9460776b4d889aa78d0ea10978595f492435e3 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 1 Nov 2020 12:03:19 +0100 Subject: [PATCH 2/2] Update docs. --- Build/MQTTnet.nuspec | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 42042a0e..97d0d54b 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,6 +13,8 @@ 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. * [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_. Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor