mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-15 00:15:15 +08:00
Merge branch 'master' of https://github.com/chkr1011/MQTTnet into feature/pending-appmsg-acknowledge
# Conflicts: # Source/MQTTnet/Client/MqttClient.cs
This commit is contained in:
tamas.kurucsai
@@ -0,0 +1,67 @@
|
||||
# For most projects, this workflow file will not need changing; you simply need
|
||||
# to commit it to your repository.
|
||||
#
|
||||
# You may wish to alter this file to override the set of languages analyzed,
|
||||
# or to provide custom queries or build logic.
|
||||
#
|
||||
# ******** NOTE ********
|
||||
# We have attempted to detect the languages in your repository. Please check
|
||||
# the `language` matrix defined below to confirm you have the correct set of
|
||||
# supported CodeQL languages.
|
||||
#
|
||||
name: "CodeQL"
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ master ]
|
||||
pull_request:
|
||||
# The branches below must be a subset of the branches above
|
||||
branches: [ master ]
|
||||
schedule:
|
||||
- cron: '23 9 * * 6'
|
||||
|
||||
jobs:
|
||||
analyze:
|
||||
name: Analyze
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
language: [ 'csharp' ]
|
||||
# CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python' ]
|
||||
# Learn more:
|
||||
# https://docs.github.com/en/free-pro-team@latest/github/finding-security-vulnerabilities-and-errors-in-your-code/configuring-code-scanning#changing-the-languages-that-are-analyzed
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v2
|
||||
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: Initialize CodeQL
|
||||
uses: github/codeql-action/init@v1
|
||||
with:
|
||||
languages: ${{ matrix.language }}
|
||||
# If you wish to specify custom queries, you can do so here or in a config file.
|
||||
# By default, queries listed here will override any specified in a config file.
|
||||
# Prefix the list here with "+" to use these queries and those in the config file.
|
||||
# queries: ./path/to/local/query, your-org/your-repo/queries@main
|
||||
|
||||
# Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
|
||||
# If this step fails, then you should remove it and run the build manually (see below)
|
||||
- name: Autobuild
|
||||
uses: github/codeql-action/autobuild@v1
|
||||
|
||||
# ℹ️ Command-line programs to run using the OS shell.
|
||||
# 📚 https://git.io/JvXDl
|
||||
|
||||
# ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
|
||||
# and modify them (or add more) to build your code if your project
|
||||
# uses a compiled language
|
||||
|
||||
#- run: |
|
||||
# make bootstrap
|
||||
# make release
|
||||
|
||||
- name: Perform CodeQL Analysis
|
||||
uses: github/codeql-action/analyze@v1
|
||||
@@ -15,7 +15,10 @@
|
||||
* [Core] Fixed some issues in nuget packages.
|
||||
* [Client] Changed exception types so that proper exceptions are thrown when connecting (#1082).
|
||||
* [Client] Exposed Dup flag in application messages.
|
||||
* [Client] Improved keep alive message sending.
|
||||
* [RpcClient] Fixed an issue when using a custom application message reveived handler (#1093).
|
||||
* [RpcClient] Fixed a race condition when using timeout and cancellation token at the same time (BREAKING CHANGE!).
|
||||
* [Server] Improved keep alive message sending.
|
||||
|
||||
Git commit: $gitCommit
|
||||
</releaseNotes>
|
||||
|
||||
@@ -12,4 +12,5 @@
|
||||
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
|
||||
@@ -7,12 +7,8 @@ namespace MQTTnet.Extensions.Rpc
|
||||
{
|
||||
public interface IMqttRpcClient : IDisposable
|
||||
{
|
||||
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
|
||||
|
||||
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
|
||||
|
||||
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
|
||||
|
||||
Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
|
||||
Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
|
||||
}
|
||||
}
|
||||
@@ -5,9 +5,9 @@ using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
|
||||
using MQTTnet.Protocol;
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Client.Subscribing;
|
||||
using MQTTnet.Implementations;
|
||||
|
||||
namespace MQTTnet.Extensions.Rpc
|
||||
@@ -36,22 +36,27 @@ namespace MQTTnet.Extensions.Rpc
|
||||
_mqttClient.ApplicationMessageReceivedHandler = _applicationMessageReceivedHandler;
|
||||
}
|
||||
|
||||
public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel, CancellationToken.None);
|
||||
using (var timeoutToken = new CancellationTokenSource(timeout))
|
||||
{
|
||||
try
|
||||
{
|
||||
return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, timeoutToken.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException exception)
|
||||
{
|
||||
if (timeoutToken.IsCancellationRequested)
|
||||
{
|
||||
throw new MqttCommunicationTimedOutException(exception);
|
||||
}
|
||||
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
|
||||
{
|
||||
return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
return ExecuteAsync(timeout, methodName, payload, qualityOfServiceLevel, CancellationToken.None);
|
||||
}
|
||||
|
||||
public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
|
||||
public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
|
||||
{
|
||||
if (methodName == null) throw new ArgumentNullException(nameof(methodName));
|
||||
|
||||
@@ -89,48 +94,33 @@ namespace MQTTnet.Extensions.Rpc
|
||||
|
||||
try
|
||||
{
|
||||
#if NET452
|
||||
var promise = new TaskCompletionSource<byte[]>();
|
||||
#else
|
||||
var promise = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
#endif
|
||||
|
||||
if (!_waitingCalls.TryAdd(responseTopic, promise))
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
}
|
||||
|
||||
await _mqttClient.SubscribeAsync(responseTopic, qualityOfServiceLevel).ConfigureAwait(false);
|
||||
await _mqttClient.PublishAsync(requestMessage).ConfigureAwait(false);
|
||||
var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
|
||||
.WithTopicFilter(responseTopic, qualityOfServiceLevel)
|
||||
.Build();
|
||||
|
||||
using (var timeoutCts = new CancellationTokenSource(timeout))
|
||||
using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token))
|
||||
await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
|
||||
await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
using (cancellationToken.Register(() => { promise.TrySetCanceled(); }))
|
||||
{
|
||||
linkedCts.Token.Register(() =>
|
||||
{
|
||||
if (!promise.Task.IsCompleted && !promise.Task.IsFaulted && !promise.Task.IsCanceled)
|
||||
{
|
||||
promise.TrySetCanceled();
|
||||
}
|
||||
});
|
||||
|
||||
try
|
||||
{
|
||||
var result = await promise.Task.ConfigureAwait(false);
|
||||
timeoutCts.Cancel(false);
|
||||
return result;
|
||||
}
|
||||
catch (OperationCanceledException exception)
|
||||
{
|
||||
if (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new MqttCommunicationTimedOutException(exception);
|
||||
}
|
||||
else
|
||||
{
|
||||
throw;
|
||||
}
|
||||
}
|
||||
return await promise.Task.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_waitingCalls.TryRemove(responseTopic, out _);
|
||||
|
||||
await _mqttClient.UnsubscribeAsync(responseTopic).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
@@ -142,11 +132,15 @@ namespace MQTTnet.Extensions.Rpc
|
||||
return PlatformAbstractionLayer.CompletedTask;
|
||||
}
|
||||
|
||||
#if NET452
|
||||
Task.Run(() => promise.TrySetResult(eventArgs.ApplicationMessage.Payload));
|
||||
#else
|
||||
promise.TrySetResult(eventArgs.ApplicationMessage.Payload);
|
||||
#endif
|
||||
|
||||
// Set this message to handled to that other code can avoid execution etc.
|
||||
eventArgs.IsHandled = true;
|
||||
|
||||
|
||||
return PlatformAbstractionLayer.CompletedTask;
|
||||
}
|
||||
|
||||
@@ -162,4 +156,4 @@ namespace MQTTnet.Extensions.Rpc
|
||||
_waitingCalls.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
using System;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Extensions.Rpc
|
||||
{
|
||||
public static class MqttRpcClientExtensions
|
||||
{
|
||||
public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
|
||||
{
|
||||
if (client == null) throw new ArgumentNullException(nameof(client));
|
||||
|
||||
var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);
|
||||
|
||||
return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,6 @@ using MQTTnet.PacketDispatcher;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Protocol;
|
||||
using System;
|
||||
using System.Diagnostics;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Implementations;
|
||||
@@ -25,8 +24,6 @@ namespace MQTTnet.Client
|
||||
{
|
||||
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
|
||||
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
|
||||
volatile int _lastSentAtTicks;
|
||||
volatile int _lastReceivedAtTicks;
|
||||
readonly object _disconnectLock = new object();
|
||||
|
||||
readonly IMqttClientAdapterFactory _adapterFactory;
|
||||
@@ -45,6 +42,8 @@ namespace MQTTnet.Client
|
||||
bool _isConnected;
|
||||
MqttClientDisconnectReason _disconnectReason;
|
||||
|
||||
DateTime _lastPacketSentTimestamp;
|
||||
|
||||
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
|
||||
{
|
||||
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
||||
@@ -79,7 +78,7 @@ namespace MQTTnet.Client
|
||||
Options = options;
|
||||
|
||||
_packetIdentifierProvider.Reset();
|
||||
_packetDispatcher.Cancel();
|
||||
_packetDispatcher.CancelAll();
|
||||
|
||||
_backgroundCancellationTokenSource = new CancellationTokenSource();
|
||||
var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
|
||||
@@ -102,8 +101,7 @@ namespace MQTTnet.Client
|
||||
authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_lastSentAtTicks = Environment.TickCount;
|
||||
_lastReceivedAtTicks = Environment.TickCount;
|
||||
_lastPacketSentTimestamp = DateTime.UtcNow;
|
||||
|
||||
if (Options.KeepAlivePeriod != TimeSpan.Zero)
|
||||
{
|
||||
@@ -391,7 +389,7 @@ namespace MQTTnet.Client
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
_lastReceivedAtTicks = Environment.TickCount;
|
||||
_lastPacketSentTimestamp = DateTime.UtcNow;
|
||||
|
||||
return _adapter.SendPacketAsync(packet, cancellationToken);
|
||||
}
|
||||
@@ -400,18 +398,17 @@ namespace MQTTnet.Client
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
ushort identifier = 0;
|
||||
if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier > 0)
|
||||
ushort packetIdentifier = 0;
|
||||
if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier)
|
||||
{
|
||||
identifier = packetWithIdentifier.PacketIdentifier;
|
||||
packetIdentifier = packetWithIdentifier.PacketIdentifier;
|
||||
}
|
||||
|
||||
using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(identifier))
|
||||
using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(packetIdentifier))
|
||||
{
|
||||
try
|
||||
{
|
||||
_lastSentAtTicks = Environment.TickCount;
|
||||
await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
|
||||
await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
@@ -446,15 +443,15 @@ namespace MQTTnet.Client
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Values described here: [MQTT-3.1.2-24].
|
||||
var waitTime = keepAlivePeriod - TimeSpan.FromMilliseconds(unchecked(Environment.TickCount - _lastSentAtTicks));
|
||||
var timeWithoutPacketSent = DateTime.UtcNow - _lastPacketSentTimestamp;
|
||||
|
||||
if (waitTime <= TimeSpan.Zero)
|
||||
if (timeWithoutPacketSent > keepAlivePeriod)
|
||||
{
|
||||
await SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false);
|
||||
await PingAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Wait a fixed time in all cases. Calculation of the remaining time is complicated
|
||||
// due to some edge cases and was buggy in the past. Now we wait half a second because the
|
||||
// due to some edge cases and was buggy in the past. Now we wait several ms because the
|
||||
// min keep alive value is one second so that the server will wait 1.5 seconds for a PING
|
||||
// packet.
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken).ConfigureAwait(false);
|
||||
@@ -538,7 +535,7 @@ namespace MQTTnet.Client
|
||||
_logger.Error(exception, "Error while receiving packets.");
|
||||
}
|
||||
|
||||
_packetDispatcher.Dispatch(exception);
|
||||
_packetDispatcher.FailAll(exception);
|
||||
|
||||
if (!DisconnectIsPending())
|
||||
{
|
||||
@@ -555,8 +552,6 @@ namespace MQTTnet.Client
|
||||
{
|
||||
try
|
||||
{
|
||||
_lastReceivedAtTicks = Environment.TickCount;
|
||||
|
||||
if (packet is MqttPublishPacket publishPacket)
|
||||
{
|
||||
EnqueueReceivedPublishPacket(publishPacket);
|
||||
@@ -569,10 +564,6 @@ namespace MQTTnet.Client
|
||||
{
|
||||
await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else if (packet is MqttPingReqPacket)
|
||||
{
|
||||
await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else if (packet is MqttDisconnectPacket disconnectPacket)
|
||||
{
|
||||
await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
|
||||
@@ -581,11 +572,22 @@ namespace MQTTnet.Client
|
||||
{
|
||||
await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
|
||||
}
|
||||
else if (packet is MqttPingRespPacket)
|
||||
{
|
||||
_packetDispatcher.TryDispatch(packet);
|
||||
}
|
||||
else if (packet is MqttPingReqPacket)
|
||||
{
|
||||
throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a Client to the Server only.");
|
||||
}
|
||||
else
|
||||
{
|
||||
_packetDispatcher.Dispatch(packet);
|
||||
if (!_packetDispatcher.TryDispatch(packet))
|
||||
{
|
||||
throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
if (_cleanDisconnectInitiated)
|
||||
@@ -605,7 +607,7 @@ namespace MQTTnet.Client
|
||||
_logger.Error(exception, "Error while receiving packets.");
|
||||
}
|
||||
|
||||
_packetDispatcher.Dispatch(exception);
|
||||
_packetDispatcher.FailAll(exception);
|
||||
|
||||
if (!DisconnectIsPending())
|
||||
{
|
||||
@@ -725,7 +727,7 @@ namespace MQTTnet.Client
|
||||
_disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection);
|
||||
|
||||
// Also dispatch disconnect to waiting threads to generate a proper exception.
|
||||
_packetDispatcher.Dispatch(disconnectPacket);
|
||||
_packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
|
||||
|
||||
if (!DisconnectIsPending())
|
||||
{
|
||||
|
||||
@@ -252,7 +252,11 @@ namespace MQTTnet.Client.Options
|
||||
{
|
||||
if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
|
||||
|
||||
_tlsParameters = new MqttClientOptionsBuilderTlsParameters();
|
||||
_tlsParameters = new MqttClientOptionsBuilderTlsParameters
|
||||
{
|
||||
UseTls = true
|
||||
};
|
||||
|
||||
optionsBuilder(_tlsParameters);
|
||||
return this;
|
||||
}
|
||||
|
||||
@@ -5,6 +5,8 @@ namespace MQTTnet.PacketDispatcher
|
||||
{
|
||||
public interface IMqttPacketAwaiter : IDisposable
|
||||
{
|
||||
MqttPacketAwaiterPacketFilter PacketFilter { get; }
|
||||
|
||||
void Complete(MqttBasePacket packet);
|
||||
|
||||
void Fail(Exception exception);
|
||||
|
||||
@@ -9,12 +9,16 @@ namespace MQTTnet.PacketDispatcher
|
||||
public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
|
||||
{
|
||||
readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
|
||||
readonly ushort? _packetIdentifier;
|
||||
readonly MqttPacketDispatcher _owningPacketDispatcher;
|
||||
|
||||
public MqttPacketAwaiter(ushort? packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
|
||||
public MqttPacketAwaiter(ushort packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
|
||||
{
|
||||
_packetIdentifier = packetIdentifier;
|
||||
PacketFilter = new MqttPacketAwaiterPacketFilter
|
||||
{
|
||||
Type = typeof(TPacket),
|
||||
Identifier = packetIdentifier
|
||||
};
|
||||
|
||||
_owningPacketDispatcher = owningPacketDispatcher ?? throw new ArgumentNullException(nameof(owningPacketDispatcher));
|
||||
#if NET452
|
||||
_taskCompletionSource = new TaskCompletionSource<MqttBasePacket>();
|
||||
@@ -22,7 +26,9 @@ namespace MQTTnet.PacketDispatcher
|
||||
_taskCompletionSource = new TaskCompletionSource<MqttBasePacket>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
public MqttPacketAwaiterPacketFilter PacketFilter { get; }
|
||||
|
||||
public async Task<TPacket> WaitOneAsync(TimeSpan timeout)
|
||||
{
|
||||
using (var timeoutToken = new CancellationTokenSource(timeout))
|
||||
@@ -82,7 +88,7 @@ namespace MQTTnet.PacketDispatcher
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_owningPacketDispatcher.RemoveAwaiter<TPacket>(_packetIdentifier);
|
||||
_owningPacketDispatcher.RemoveAwaiter(this);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.PacketDispatcher
|
||||
{
|
||||
public sealed class MqttPacketAwaiterPacketFilter
|
||||
{
|
||||
public Type Type { get; set; }
|
||||
|
||||
public ushort Identifier { get; set; }
|
||||
}
|
||||
}
|
||||
@@ -1,103 +1,101 @@
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Packets;
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace MQTTnet.PacketDispatcher
|
||||
{
|
||||
public sealed class MqttPacketDispatcher
|
||||
{
|
||||
readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
|
||||
readonly List<IMqttPacketAwaiter> _awaiters = new List<IMqttPacketAwaiter>();
|
||||
|
||||
public void Dispatch(Exception exception)
|
||||
public void FailAll(Exception exception)
|
||||
{
|
||||
foreach (var awaiter in _awaiters)
|
||||
{
|
||||
awaiter.Value.Fail(exception);
|
||||
}
|
||||
if (exception == null) throw new ArgumentNullException(nameof(exception));
|
||||
|
||||
_awaiters.Clear();
|
||||
lock (_awaiters)
|
||||
{
|
||||
foreach (var awaiter in _awaiters)
|
||||
{
|
||||
awaiter.Fail(exception);
|
||||
}
|
||||
|
||||
_awaiters.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
public void CancelAll()
|
||||
{
|
||||
lock (_awaiters)
|
||||
{
|
||||
foreach (var entry in _awaiters)
|
||||
{
|
||||
entry.Cancel();
|
||||
}
|
||||
|
||||
_awaiters.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
public bool TryDispatch(MqttBasePacket packet)
|
||||
{
|
||||
if (packet == null) throw new ArgumentNullException(nameof(packet));
|
||||
|
||||
if (packet is MqttDisconnectPacket disconnectPacket)
|
||||
{
|
||||
foreach (var packetAwaiter in _awaiters)
|
||||
{
|
||||
packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
ushort identifier = 0;
|
||||
if (packet is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier > 0)
|
||||
if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
|
||||
{
|
||||
identifier = packetWithIdentifier.PacketIdentifier;
|
||||
}
|
||||
|
||||
var type = packet.GetType();
|
||||
var key = new Tuple<ushort, Type>(identifier, type);
|
||||
|
||||
if (_awaiters.TryRemove(key, out var awaiter))
|
||||
var packetType = packet.GetType();
|
||||
var matchingAwaiters = new List<IMqttPacketAwaiter>();
|
||||
|
||||
lock (_awaiters)
|
||||
{
|
||||
awaiter.Complete(packet);
|
||||
return true;
|
||||
for (var i = _awaiters.Count - 1; i >= 0; i--)
|
||||
{
|
||||
var entry = _awaiters[i];
|
||||
|
||||
// Note: The PingRespPacket will also arrive here and has NO identifier but there
|
||||
// is code which waits for it. So the code must be able to deal with filters which
|
||||
// are referring to the type only (identifier is 0)!
|
||||
if (entry.PacketFilter.Type != packetType || entry.PacketFilter.Identifier != identifier)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
matchingAwaiters.Add(entry);
|
||||
_awaiters.RemoveAt(i);
|
||||
}
|
||||
}
|
||||
|
||||
foreach (var matchingEntry in matchingAwaiters)
|
||||
{
|
||||
matchingEntry.Complete(packet);
|
||||
}
|
||||
|
||||
return false;
|
||||
return matchingAwaiters.Count > 0;
|
||||
}
|
||||
|
||||
public void Dispatch(MqttBasePacket packet)
|
||||
|
||||
public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttBasePacket
|
||||
{
|
||||
if (packet == null) throw new ArgumentNullException(nameof(packet));
|
||||
var awaiter = new MqttPacketAwaiter<TResponsePacket>(packetIdentifier, this);
|
||||
|
||||
if (!TryDispatch(packet))
|
||||
lock (_awaiters)
|
||||
{
|
||||
throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
|
||||
_awaiters.Add(awaiter);
|
||||
}
|
||||
}
|
||||
|
||||
public void Cancel()
|
||||
{
|
||||
foreach (var awaiter in _awaiters)
|
||||
{
|
||||
awaiter.Value.Cancel();
|
||||
}
|
||||
|
||||
_awaiters.Clear();
|
||||
}
|
||||
|
||||
public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
|
||||
{
|
||||
if (!identifier.HasValue)
|
||||
{
|
||||
identifier = 0;
|
||||
}
|
||||
|
||||
var awaiter = new MqttPacketAwaiter<TResponsePacket>(identifier, this);
|
||||
|
||||
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
|
||||
if (!_awaiters.TryAdd(key, awaiter))
|
||||
{
|
||||
throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{key.Item2.Name}' with identifier {key.Item1}.");
|
||||
}
|
||||
|
||||
|
||||
return awaiter;
|
||||
}
|
||||
|
||||
public void RemoveAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
|
||||
public void RemoveAwaiter(IMqttPacketAwaiter awaiter)
|
||||
{
|
||||
if (!identifier.HasValue)
|
||||
if (awaiter == null) throw new ArgumentNullException(nameof(awaiter));
|
||||
|
||||
lock (_awaiters)
|
||||
{
|
||||
identifier = 0;
|
||||
_awaiters.Remove(awaiter);
|
||||
}
|
||||
|
||||
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
|
||||
_awaiters.TryRemove(key, out _);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -212,6 +212,10 @@ namespace MQTTnet.Server
|
||||
{
|
||||
await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else if (packet is MqttPingRespPacket)
|
||||
{
|
||||
throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only.");
|
||||
}
|
||||
else if (packet is MqttDisconnectPacket)
|
||||
{
|
||||
Session.WillMessage = null;
|
||||
@@ -222,7 +226,10 @@ namespace MQTTnet.Server
|
||||
}
|
||||
else
|
||||
{
|
||||
_packetDispatcher.Dispatch(packet);
|
||||
if (!_packetDispatcher.TryDispatch(packet))
|
||||
{
|
||||
throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -255,7 +262,7 @@ namespace MQTTnet.Server
|
||||
Session.WillMessage = null;
|
||||
}
|
||||
|
||||
_packetDispatcher.Cancel();
|
||||
_packetDispatcher.CancelAll();
|
||||
|
||||
_logger.Info("Client '{0}': Connection stopped.", ClientId);
|
||||
|
||||
|
||||
@@ -9,6 +9,8 @@ using System.Diagnostics;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Extensions.Rpc;
|
||||
using MQTTnet.Extensions.Rpc.Options;
|
||||
using MQTTnet.LowLevelClient;
|
||||
|
||||
namespace MQTTnet.Tests.Mockups
|
||||
@@ -77,6 +79,11 @@ namespace MQTTnet.Tests.Mockups
|
||||
};
|
||||
}
|
||||
|
||||
public async Task<IMqttRpcClient> ConnectRpcClientAsync(IMqttRpcClientOptions options)
|
||||
{
|
||||
return new MqttRpcClient(await ConnectClientAsync(), options);
|
||||
}
|
||||
|
||||
public IMqttClient CreateClient()
|
||||
{
|
||||
lock (_clients)
|
||||
|
||||
@@ -109,15 +109,6 @@ namespace MQTTnet.Tests.Mockups
|
||||
return Implementation.StartAsync(options);
|
||||
}
|
||||
|
||||
public void ConnectionValidator(MqttConnectionValidatorContext ctx)
|
||||
{
|
||||
if (!ctx.ClientId.StartsWith(TestContext.TestName))
|
||||
{
|
||||
TestEnvironment.TrackException(new InvalidOperationException($"Invalid client ID used ({ctx.ClientId}). It must start with UnitTest name."));
|
||||
ctx.ReasonCode = Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
|
||||
}
|
||||
}
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
return Implementation.StopAsync();
|
||||
@@ -137,5 +128,14 @@ namespace MQTTnet.Tests.Mockups
|
||||
{
|
||||
Implementation.Dispose();
|
||||
}
|
||||
|
||||
void ConnectionValidator(MqttConnectionValidatorContext ctx)
|
||||
{
|
||||
if (!ctx.ClientId.StartsWith(TestContext.TestName))
|
||||
{
|
||||
TestEnvironment.TrackException(new InvalidOperationException($"Invalid client ID used ({ctx.ClientId}). It must start with UnitTest name."));
|
||||
ctx.ReasonCode = Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -81,18 +81,19 @@ namespace MQTTnet.Tests
|
||||
|
||||
var requestSender = await testEnvironment.ConnectClientAsync();
|
||||
|
||||
var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
|
||||
var rpcClient = await testEnvironment.ConnectRpcClientAsync(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
|
||||
|
||||
await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
|
||||
async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
|
||||
{
|
||||
using (var testEnvironment = new TestEnvironment(TestContext))
|
||||
{
|
||||
await testEnvironment.StartServerAsync();
|
||||
var responseSender = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(protocolVersion));
|
||||
await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
|
||||
await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", qosLevel);
|
||||
|
||||
responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
|
||||
{
|
||||
@@ -101,14 +102,16 @@ namespace MQTTnet.Tests
|
||||
|
||||
var requestSender = await testEnvironment.ConnectClientAsync();
|
||||
|
||||
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));
|
||||
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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
|
||||
class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
|
||||
{
|
||||
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user