From 1df67989349fdcabb4cb8570e455ba9625aba0ce Mon Sep 17 00:00:00 2001 From: Christian Date: Wed, 11 Apr 2018 22:19:38 +0200 Subject: [PATCH] Fix issues in RPC extension. --- Build/MQTTnet.Extensions.Rpc.nuspec | 32 +++++++++++-------- Build/MQTTnet.nuspec | 20 ++---------- .../MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 30 ++++++----------- .../MQTTnet.Extensions.Rpc/SampleCCode.c | 29 ++++++++++++++++- .../MqttApplicationMessageBuilder.cs | 9 ++++-- .../Packets/MqttUnsubscribe.cs | 2 +- MQTTnet.sln | 1 + .../MainPage.xaml.cs | 5 +-- 8 files changed, 70 insertions(+), 58 deletions(-) diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index ec09c1de..2f2fe9b7 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -2,7 +2,7 @@ MQTTnet.Extensions.Rpc - 2.6.0 + 2.7.4 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -12,36 +12,42 @@ This is a extension library which allows executing synchronous device calls including a response using MQTTnet. * Initial version. - Copyright Christian Kratky 2016-2017 + Copyright Christian Kratky 2016-2018 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 - + - + - - + - - + - - + - - + + + + - + + + + + + + + \ No newline at end of file diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 2b7bf1e6..cb859b16 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.7.4 + 2.7.5 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,47 +10,34 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * [Client] Fixed a deadlock while the client disconnects. -* [Client] Fixed broken support for protocol version 3.1.0. -* [Server] The _MqttTcpServerAdapter_ is now added to the ASP.NET services. -* [Server] _MqttServerAdapter_ is renamed to _MqttTcpServerAdapter_ (BREAKING CHANGE!). -* [Server] The server no longer sends the will message of a client if the disconnect was clean (via _Disconnect_ packet). -* [Server] The application message interceptor now allows closing the connection. -* [Server] Added a new flag for the _ClientDisconnected_ event which contains a value indicating whether the disconnect was clean (via _Disconnect_ packet). + * [Client] Added new overloads for the message builder. Copyright Christian Kratky 2016-2018 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 - - - - - - - @@ -62,7 +49,6 @@ - - + \ No newline at end of file diff --git a/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 938fcb4c..78836ea4 100644 --- a/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Extensions/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Concurrent; +using System.Text; using System.Threading.Tasks; using MQTTnet.Client; using MQTTnet.Internal; @@ -9,11 +10,9 @@ namespace MQTTnet.Extensions.Rpc { public sealed class MqttRpcClient : IDisposable { - private const string ResponseTopic = "$MQTTnet.RPC/+/+/response"; private readonly ConcurrentDictionary> _waitingCalls = new ConcurrentDictionary>(); private readonly IMqttClient _mqttClient; - private bool _isEnabled; - + public MqttRpcClient(IMqttClient mqttClient) { _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); @@ -21,16 +20,9 @@ namespace MQTTnet.Extensions.Rpc _mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived; } - public async Task EnableAsync() + public Task ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) { - await _mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(ResponseTopic).WithAtLeastOnceQoS().Build()); - _isEnabled = true; - } - - public async Task DisableAsync() - { - await _mqttClient.UnsubscribeAsync(ResponseTopic); - _isEnabled = false; + return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel); } public async Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel) @@ -42,12 +34,7 @@ namespace MQTTnet.Extensions.Rpc throw new ArgumentException("The method name cannot contain /, + or #."); } - if (!_isEnabled) - { - throw new InvalidOperationException("The RPC client is not enabled."); - } - - var requestTopic = $"$MQTTnet.RPC/{Guid.NewGuid():N}/{methodName}"; + var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{methodName}"; var responseTopic = requestTopic + "/response"; var requestMessage = new MqttApplicationMessageBuilder() @@ -64,18 +51,21 @@ namespace MQTTnet.Extensions.Rpc throw new InvalidOperationException(); } - await _mqttClient.PublishAsync(requestMessage); + await _mqttClient.SubscribeAsync(responseTopic, qualityOfServiceLevel).ConfigureAwait(false); + await _mqttClient.PublishAsync(requestMessage).ConfigureAwait(false); + return await tcs.Task.TimeoutAfter(timeout); } finally { _waitingCalls.TryRemove(responseTopic, out _); + await _mqttClient.UnsubscribeAsync(responseTopic).ConfigureAwait(false); } } private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs eventArgs) { - if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out TaskCompletionSource tcs)) + if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var tcs)) { return; } diff --git a/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c b/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c index 5f282702..1afa32ce 100644 --- a/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c +++ b/Extensions/MQTTnet.Extensions.Rpc/SampleCCode.c @@ -1 +1,28 @@ - \ No newline at end of file +// If using the MQTT client PubSubClient it must be ensured that the request topic for each method is subscribed like the following. +_mqttClient.subscribe("MQTTnet.RPC/+/ping"); +_mqttClient.subscribe("MQTTnet.RPC/+/do_something"); + +// It is not allowed to change the structure of the topic. Otherwise RPC will not work. So method names can be separated using +// an _ or . but no +, # or /.. If it is required to distinguish between devices own rules can be defined like the following. +_mqttClient.subscribe("MQTTnet.RPC/+/deviceA.ping"); +_mqttClient.subscribe("MQTTnet.RPC/+/deviceB.ping"); +_mqttClient.subscribe("MQTTnet.RPC/+/deviceC.getTemperature"); + +// Within the callback of the mqtt client the topic must be checked if it belongs to MQTTnet RPC. The following code shows one +// possible way. +void mqtt_Callback(char *topic, byte *payload, unsigned int payloadLength) +{ + String topicString = String(topic); + + if (topicString.startsWith("MQTTnet.RPC/")) { + String responseTopic = topicString + String("/response"); + + if (topicString.endsWith("/deviceA.ping")) { + mqtt_publish(responseTopic, "pong", false); + return; + } + } +} + +// Do not send response message with the _retain_ flag set to true. +// All required data for a RPC call and the result must be placed into the payload. \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageBuilder.cs b/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageBuilder.cs index ce949b85..79c65ece 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageBuilder.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttApplicationMessageBuilder.cs @@ -32,7 +32,12 @@ namespace MQTTnet return this; } - public MqttApplicationMessageBuilder WithPayload(MemoryStream payload) + public MqttApplicationMessageBuilder WithPayload(Stream payload) + { + return WithPayload(payload, payload.Length - payload.Position); + } + + public MqttApplicationMessageBuilder WithPayload(Stream payload, long length) { if (payload == null) { @@ -46,7 +51,7 @@ namespace MQTTnet } else { - _payload = new byte[payload.Length - payload.Position]; + _payload = new byte[length]; payload.Read(_payload, 0, _payload.Length); } diff --git a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs index 6c24a477..a4d84535 100644 --- a/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs +++ b/Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Packets public override string ToString() { var topicFiltersText = string.Join(",", TopicFilters); - return "Subscribe: [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]"; + return "Unsubscribe: [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]"; } } } diff --git a/MQTTnet.sln b/MQTTnet.sln index cece21c5..fab76088 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -18,6 +18,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1 appveyor.yml = appveyor.yml Build\build.ps1 = Build\build.ps1 Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec + Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec Build\MQTTnet.nuspec = Build\MQTTnet.nuspec EndProjectSection EndProject diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index da0c2f31..d907f1d8 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -350,14 +350,11 @@ namespace MQTTnet.TestApp.UniversalWindows payload = Convert.FromBase64String(RpcPayload.Text); } - try { var rpcClient = new MqttRpcClient(_mqttClient); - await rpcClient.EnableAsync(); var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), RpcMethod.Text, payload, qos); - await rpcClient.DisableAsync(); - + RpcResponses.Items.Add(RpcMethod.Text + " >>> " + Encoding.UTF8.GetString(response)); } catch (MqttCommunicationTimedOutException)