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

Use the net452 supporting async task completion source (#1607)

Browse Source
This commit is contained in:
Christian
2022-12-04 13:39:50 +01:00
committed by GitHub
parent 180e7f79d4
commit d572490a67
1 changed files with 39 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
+39 -42
View File
@@ -2,23 +2,23 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Client;
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc
{
public sealed class MqttRpcClient : IMqttRpcClient
{
readonly ConcurrentDictionary<string, TaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, TaskCompletionSource<byte[]>>();
readonly IMqttClient _mqttClient;
readonly MqttRpcClientOptions _options;
readonly ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, AsyncTaskCompletionSource<byte[]>>();
public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options)
{
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
@@ -27,6 +27,18 @@ namespace MQTTnet.Extensions.Rpc
_mqttClient.ApplicationMessageReceivedAsync += HandleApplicationMessageReceivedAsync;
}
public void Dispose()
{
_mqttClient.ApplicationMessageReceivedAsync -= HandleApplicationMessageReceivedAsync;
foreach (var tcs in _waitingCalls)
{
tcs.Value.TrySetCanceled();
}
_waitingCalls.Clear();
}
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
{
using (var timeoutToken = new CancellationTokenSource(timeout))
@@ -49,15 +61,19 @@ namespace MQTTnet.Extensions.Rpc
public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default)
{
if (methodName == null) throw new ArgumentNullException(nameof(methodName));
var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(new TopicGenerationContext
if (methodName == null)
{
MethodName = methodName,
QualityOfServiceLevel = qualityOfServiceLevel,
MqttClient = _mqttClient,
Options = _options
});
throw new ArgumentNullException(nameof(methodName));
}
var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(
new TopicGenerationContext
{
MethodName = methodName,
QualityOfServiceLevel = qualityOfServiceLevel,
MqttClient = _mqttClient,
Options = _options
});
var requestTopic = topicNames.RequestTopic;
var responseTopic = topicNames.ResponseTopic;
@@ -72,8 +88,7 @@ namespace MQTTnet.Extensions.Rpc
throw new MqttProtocolViolationException("RPC response topic is empty.");
}
var requestMessage = new MqttApplicationMessageBuilder()
.WithTopic(requestTopic)
var requestMessage = new MqttApplicationMessageBuilder().WithTopic(requestTopic)
.WithPayload(payload)
.WithQualityOfServiceLevel(qualityOfServiceLevel)
.WithResponseTopic(responseTopic)
@@ -81,25 +96,23 @@ namespace MQTTnet.Extensions.Rpc
try
{
#if NET452
var awaitable = new TaskCompletionSource<byte[]>();
#else
var awaitable = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
#endif
var awaitable = new AsyncTaskCompletionSource<byte[]>();
if (!_waitingCalls.TryAdd(responseTopic, awaitable))
{
throw new InvalidOperationException();
}
var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(responseTopic, qualityOfServiceLevel)
.Build();
var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(responseTopic, qualityOfServiceLevel).Build();
await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
using (cancellationToken.Register(() => { awaitable.TrySetCanceled(); }))
using (cancellationToken.Register(
() =>
{
awaitable.TrySetCanceled();
}))
{
return await awaitable.Task.ConfigureAwait(false);
}
@@ -107,8 +120,8 @@ namespace MQTTnet.Extensions.Rpc
finally
{
_waitingCalls.TryRemove(responseTopic, out _);
await _mqttClient.UnsubscribeAsync(responseTopic).ConfigureAwait(false);
await _mqttClient.UnsubscribeAsync(responseTopic, CancellationToken.None).ConfigureAwait(false);
}
}
@@ -119,28 +132,12 @@ namespace MQTTnet.Extensions.Rpc
return CompletedTask.Instance;
}
#if NET452
Task.Run(() => awaitable.TrySetResult(eventArgs.ApplicationMessage.Payload));
#else
awaitable.TrySetResult(eventArgs.ApplicationMessage.Payload);
#endif
// Set this message to handled to that other code can avoid execution etc.
eventArgs.IsHandled = true;
return CompletedTask.Instance;
}
public void Dispose()
{
_mqttClient.ApplicationMessageReceivedAsync -= HandleApplicationMessageReceivedAsync;
foreach (var tcs in _waitingCalls)
{
tcs.Value.TrySetCanceled();
}
_waitingCalls.Clear();
}
}
}