Move the RPC topic validation to the topic generation strategy.

This commit is contained in:
Christian Kratky
2020-11-01 12:00:12 +01:00
parent 34ae54d542
commit 2716db04e3
8 changed files with 39 additions and 21 deletions
@@ -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);
}
}
+6 -11
View File
@@ -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))
{
@@ -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();
}
@@ -3,7 +3,7 @@ using System;
namespace MQTTnet.Extensions.Rpc.Options
{
public class MqttRpcClientOptionsBuilder
public sealed class MqttRpcClientOptionsBuilder
{
IMqttRpcClientTopicGenerationStrategy _topicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy();
@@ -1,6 +1,6 @@
namespace MQTTnet.Extensions.Rpc.Options
{
public class MqttRpcTopicPair
public sealed class MqttRpcTopicPair
{
public string RequestTopic { get; set; }
@@ -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";
@@ -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; }
@@ -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);
}
}
}