diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml new file mode 100644 index 00000000..4b70f0c4 --- /dev/null +++ b/.github/workflows/codeql-analysis.yml @@ -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 diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 30d6171d..21efd371 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -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 diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings index 0b9e2dc5..e81ffe21 100644 --- a/MQTTnet.sln.DotSettings +++ b/MQTTnet.sln.DotSettings @@ -12,4 +12,5 @@ True True True + True True \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs index a3a06fa2..2941fe8e 100644 --- a/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs @@ -7,12 +7,8 @@ namespace MQTTnet.Extensions.Rpc { public interface IMqttRpcClient : IDisposable { - Task ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel); - - Task ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken); - Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel); - Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken); + Task ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken); } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 3fbec759..453021f3 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -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 ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) + public async Task 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 ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken) - { - return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel, cancellationToken); - } - - public Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel) - { - return ExecuteAsync(timeout, methodName, payload, qualityOfServiceLevel, CancellationToken.None); - } - - public async Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken) + public async Task 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(); +#else + var promise = new TaskCompletionSource(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(); } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs new file mode 100644 index 00000000..005c774f --- /dev/null +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs @@ -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 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); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 16cb334a..da84f415 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -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(identifier)) + using (var packetAwaiter = _packetDispatcher.AddAwaiter(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(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()) { diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index c12c468e..4f081859 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -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; } diff --git a/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs b/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs index c1c2d4eb..8324d4ca 100644 --- a/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs +++ b/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs @@ -5,6 +5,8 @@ namespace MQTTnet.PacketDispatcher { public interface IMqttPacketAwaiter : IDisposable { + MqttPacketAwaiterPacketFilter PacketFilter { get; } + void Complete(MqttBasePacket packet); void Fail(Exception exception); diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs index 6935e746..9ed6e71c 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs @@ -9,12 +9,16 @@ namespace MQTTnet.PacketDispatcher public sealed class MqttPacketAwaiter : IMqttPacketAwaiter where TPacket : MqttBasePacket { readonly TaskCompletionSource _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(); @@ -22,7 +26,9 @@ namespace MQTTnet.PacketDispatcher _taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); #endif } - + + public MqttPacketAwaiterPacketFilter PacketFilter { get; } + public async Task WaitOneAsync(TimeSpan timeout) { using (var timeoutToken = new CancellationTokenSource(timeout)) @@ -82,7 +88,7 @@ namespace MQTTnet.PacketDispatcher public void Dispose() { - _owningPacketDispatcher.RemoveAwaiter(_packetIdentifier); + _owningPacketDispatcher.RemoveAwaiter(this); } } } \ No newline at end of file diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiterPacketFilter.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiterPacketFilter.cs new file mode 100644 index 00000000..a36acc5a --- /dev/null +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiterPacketFilter.cs @@ -0,0 +1,11 @@ +using System; + +namespace MQTTnet.PacketDispatcher +{ + public sealed class MqttPacketAwaiterPacketFilter + { + public Type Type { get; set; } + + public ushort Identifier { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs index 306ca45c..aafabf23 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs @@ -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, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary, IMqttPacketAwaiter>(); + readonly List _awaiters = new List(); - 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(identifier, type); - - if (_awaiters.TryRemove(key, out var awaiter)) + var packetType = packet.GetType(); + var matchingAwaiters = new List(); + + 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 AddAwaiter(ushort packetIdentifier) where TResponsePacket : MqttBasePacket { - if (packet == null) throw new ArgumentNullException(nameof(packet)); + var awaiter = new MqttPacketAwaiter(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 AddAwaiter(ushort? identifier) where TResponsePacket : MqttBasePacket - { - if (!identifier.HasValue) - { - identifier = 0; - } - - var awaiter = new MqttPacketAwaiter(identifier, this); - - var key = new Tuple(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(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(identifier.Value, typeof(TResponsePacket)); - _awaiters.TryRemove(key, out _); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 75083a54..8f33bb8b 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -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); diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 648d74d7..e495a108 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -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 ConnectRpcClientAsync(IMqttRpcClientOptions options) + { + return new MqttRpcClient(await ConnectClientAsync(), options); + } + public IMqttClient CreateClient() { lock (_clients) diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs index 1c651aa3..50168a5f 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs @@ -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; + } + } } } \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs index 947c1045..154f848e 100644 --- a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs @@ -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) {