YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-20 11:45:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

MqttRpcClient now adds ResponseTopic to MqttApplicationMessage when using MQTT v5 (#1295)

Browse Source 
* MqttRpcClient now adds ResponseTopic to MqttApplicationMessage when using MQTT v5

* No need for conditional protocol check (ResponseTopic property won't be added by PacketFormatter in <MQTT v5)

* Cleanup: Remove using
This commit is contained in:
Michi
2021-11-28 12:41:56 +01:00
committed by GitHub
parent 2d94abdde3
commit 25ddfc83a8
2 changed files with 100 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
+3 -2
View File
@@ -1,4 +1,4 @@
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Extensions.Rpc.Options; using MQTTnet.Extensions.Rpc.Options;
using MQTTnet.Extensions.Rpc.Options.TopicGeneration; using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
@@ -90,6 +90,7 @@ namespace MQTTnet.Extensions.Rpc
.WithTopic(requestTopic) .WithTopic(requestTopic)
.WithPayload(payload) .WithPayload(payload)
.WithQualityOfServiceLevel(qualityOfServiceLevel) .WithQualityOfServiceLevel(qualityOfServiceLevel)
.WithResponseTopic(responseTopic)
.Build(); .Build();
try try
@@ -99,7 +100,7 @@ namespace MQTTnet.Extensions.Rpc
#else #else
var awaitable = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously); var awaitable = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
#endif #endif
if (!_waitingCalls.TryAdd(responseTopic, awaitable)) if (!_waitingCalls.TryAdd(responseTopic, awaitable))
{ {
throw new InvalidOperationException(); throw new InvalidOperationException();
Tests/MQTTnet.Core.Tests/RPC_Tests.cs
+97 -4
View File
@@ -56,6 +56,25 @@ namespace MQTTnet.Tests
return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500); return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500);
} }
[TestMethod]
public Task Execute_Success_With_QoS_0_MQTT_V5_Use_ResponseTopic()
{
return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtMostOnce);
}
[TestMethod]
public Task Execute_Success_With_QoS_1_MQTT_V5_Use_ResponseTopic()
{
return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtLeastOnce);
}
[TestMethod]
public Task Execute_Success_With_QoS_2_MQTT_V5_Use_ResponseTopic()
{
return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.ExactlyOnce);
}
[TestMethod] [TestMethod]
[ExpectedException(typeof(MqttCommunicationTimedOutException))] [ExpectedException(typeof(MqttCommunicationTimedOutException))]
public async Task Execute_Timeout() public async Task Execute_Timeout()
@@ -63,7 +82,7 @@ namespace MQTTnet.Tests
using (var testEnvironment = new TestEnvironment(TestContext)) using (var testEnvironment = new TestEnvironment(TestContext))
{ {
await testEnvironment.StartServer(); await testEnvironment.StartServer();
var requestSender = await testEnvironment.ConnectClient(); var requestSender = await testEnvironment.ConnectClient();
var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()); var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build());
@@ -81,8 +100,8 @@ namespace MQTTnet.Tests
var requestSender = await testEnvironment.ConnectClient(); var requestSender = await testEnvironment.ConnectClient();
var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build()); var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()).Build());
await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce); await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
} }
} }
@@ -105,12 +124,86 @@ namespace MQTTnet.Tests
using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build())) using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
{ {
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel); var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
Assert.AreEqual("pong", Encoding.UTF8.GetString(response)); Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
} }
} }
} }
async Task Execute_Success_MQTT_V5(MqttQualityOfServiceLevel qosLevel)
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
await testEnvironment.StartServer();
var responseSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", qosLevel);
responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
{
await responseSender.PublishAsync(e.ApplicationMessage.ResponseTopic, "pong");
});
var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
{
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
}
}
}
[TestMethod]
public async Task Execute_Success_MQTT_V5_Mixed_Clients()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
await testEnvironment.StartServer();
var responseSender = await testEnvironment.ConnectClient();
await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
{
Assert.IsNull(e.ApplicationMessage.ResponseTopic);
await responseSender.PublishAsync(e.ApplicationMessage.Topic + "/response", "pong");
});
var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
{
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
}
}
}
[TestMethod]
[ExpectedException(typeof(MqttCommunicationTimedOutException))]
public async Task Execute_Timeout_MQTT_V5_Mixed_Clients()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
await testEnvironment.StartServer();
var responseSender = await testEnvironment.ConnectClient();
await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
{
Assert.IsNull(e.ApplicationMessage.ResponseTopic);
});
var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
{
var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
}
}
}
class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
{ {
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context) public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)