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)