diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec
index 505dbeb1..00179800 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_.
* [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.
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);
}
}
}