Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add possibility to specify parameter to RPC invocation #1798

Merged
merged 4 commits into from
Sep 1, 2023
Merged

Add possibility to specify parameter to RPC invocation #1798

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
02e54b0
Add possibility to specify parameter to RPC invocation which will be …
Temppus Jul 24, 2023
8ec6356
Make parameters optional
Temppus Aug 27, 2023
6d397ad
Update ReleaseNotes.md
chkr1011 Sep 1, 2023
5112648
Merge branch 'master' into master
chkr1011 Sep 1, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Protocol;
Expand All @@ -7,8 +8,8 @@ namespace MQTTnet.Extensions.Rpc
{
public interface IMqttRpcClient : IDisposable
{
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null);

Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default);
Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default);
}
}
9 changes: 5 additions & 4 deletions Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
Expand Down Expand Up @@ -42,13 +43,13 @@ public void Dispose()
_waitingCalls.Clear();
}

public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null)
{
using (var timeoutToken = new CancellationTokenSource(timeout))
{
try
{
return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, timeoutToken.Token).ConfigureAwait(false);
return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, parameters, timeoutToken.Token).ConfigureAwait(false);
}
catch (OperationCanceledException exception)
{
Expand All @@ -62,14 +63,14 @@ public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte
}
}

public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default)
public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
{
if (methodName == null)
{
throw new ArgumentNullException(nameof(methodName));
}

var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
var context = new TopicGenerationContext(_mqttClient, _options, methodName, parameters, qualityOfServiceLevel);
var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);

var requestTopic = topicNames.RequestTopic;
Expand Down
5 changes: 3 additions & 2 deletions Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Protocol;
Expand All @@ -11,13 +12,13 @@ namespace MQTTnet.Extensions.Rpc
{
public static class MqttRpcClientExtensions
{
public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, IDictionary<string,object> parameters = null)
{
if (client == null) throw new ArgumentNullException(nameof(client));

var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);

return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel);
return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel, parameters);
}
}
}
9 changes: 8 additions & 1 deletion Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,23 +3,30 @@
// See the LICENSE file in the project root for more information.

using System;
using System.Collections.Generic;
using MQTTnet.Client;
using MQTTnet.Protocol;

namespace MQTTnet.Extensions.Rpc
{
public sealed class TopicGenerationContext
{
public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName, MqttQualityOfServiceLevel qualityOfServiceLevel)
static readonly IDictionary<string, object> EmptyParameters = new Dictionary<string, object>();

public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName,
IDictionary<string, object> parameters, MqttQualityOfServiceLevel qualityOfServiceLevel)
{
MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
Parameters = parameters ?? EmptyParameters;
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please leave the parameter null because otherwise the user can modify the instance of the "empty" dictionary because it is not read only (immutable).

Copy link
Contributor Author

@Temppus Temppus Aug 14, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @chkr1011 for review.
Would it be ok if I changed type for parameters from IDictionary to IReadOnlyDictionary ? I would like to avoid checking for null in client code if not necessary.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That would indeed circumvent the issue but I am afraid that maybe some users want to change these values and use it as a way to return data to the caller as well. So I would leave it null here. If you want to avoid checking for null all the time (what I can understand) you may pass an "Empty" immutable instance on your own.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Made parameters nullable as suggested.

QualityOfServiceLevel = qualityOfServiceLevel;
MqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
Options = options ?? throw new ArgumentNullException(nameof(options));
}

public string MethodName { get; }

public IDictionary<string, object> Parameters { get; }

public IMqttClient MqttClient { get; }

public MqttRpcClientOptions Options { get; }
Expand Down
61 changes: 61 additions & 0 deletions Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
Expand Down Expand Up @@ -45,6 +46,34 @@ public async Task Execute_Success_MQTT_V5_Mixed_Clients()
}
}

[TestMethod]
public async Task Execute_Success_Parameters_Propagated_Correctly()
{
var paramValue = "123";
var parameters = new Dictionary<string, object>
{
{ TestParametersTopicGenerationStrategy.ExpectedParamName, "123" },
};

using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();

var responseSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
await responseSender.SubscribeAsync($"MQTTnet.RPC/+/ping/{paramValue}");

responseSender.ApplicationMessageReceivedAsync += e => responseSender.PublishStringAsync(e.ApplicationMessage.Topic + "/response", "pong");

using (var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder()
.WithTopicGenerationStrategy(new TestParametersTopicGenerationStrategy()).Build()))
{
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce, parameters);

Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
}
}
}

[TestMethod]
public Task Execute_Success_With_QoS_0()
{
Expand Down Expand Up @@ -215,12 +244,44 @@ class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
{
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
{
if (context.Parameters == null)
{
throw new InvalidOperationException("Parameters dictionary can not be null");
}

return new MqttRpcTopicPair
{
RequestTopic = "a",
ResponseTopic = "b"
};
}
}

class TestParametersTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
{
internal const string ExpectedParamName = "test_param_name";

public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
{
if (context.Parameters == null)
{
throw new InvalidOperationException("Parameters dictionary can not be null");
}

if (!context.Parameters.TryGetValue(ExpectedParamName, out var paramValue))
{
throw new InvalidOperationException($"Parameter with name {ExpectedParamName} not present");
}

var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{context.MethodName}/{paramValue}";
var responseTopic = requestTopic + "/response";

return new MqttRpcTopicPair
{
RequestTopic = requestTopic,
ResponseTopic = responseTopic
};
}
}
}
}