From 3f5d627e9e3b5cd93b9a73fd472766a5e278e07c Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 7 Jan 2021 23:22:45 +0100 Subject: [PATCH] Fix issues in QoS 2 handling. --- Build/MQTTnet.nuspec | 4 + Source/MQTTnet/Client/MqttClient.cs | 80 ++++++++++++++----- .../PacketDispatcher/MqttPacketDispatcher.cs | 18 ++++- 3 files changed, 78 insertions(+), 24 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index a843b524..a5aad211 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -17,6 +17,7 @@ * [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato). * [Core] Aligned the format of some log messages. * [Client] Fixed an issue with connection state detection (thanks to @tobiasfrick). +* [Client] Fixed an issue when receiving partial QoS 2 packets after a reconnect. * [ManagedClient] Removed _AutoReconnect_ since this does no longer work (BREAKING CHANGE!). * [RpcClient] Moved the RPC topic validation to the topic generation strategy. * [RcpClient] Added interface for _MqttRpcClient_. @@ -68,6 +69,9 @@ Git commit: $gitCommit + + + diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 75d4e996..a6901fe4 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -17,6 +17,7 @@ using System; using System.Diagnostics; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Implementations; namespace MQTTnet.Client { @@ -551,37 +552,25 @@ namespace MQTTnet.Client { EnqueueReceivedPublishPacket(publishPacket); } + else if (packet is MqttPubRecPacket pubRecPacket) + { + await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false); + } else if (packet is MqttPubRelPacket pubRelPacket) { - await SendAsync(new MqttPubCompPacket - { - PacketIdentifier = pubRelPacket.PacketIdentifier, - ReasonCode = MqttPubCompReasonCode.Success - }, cancellationToken).ConfigureAwait(false); + await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); } else if (packet is MqttPingReqPacket) { - await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); + await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false); } else if (packet is MqttDisconnectPacket disconnectPacket) { - _disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection); - - // Also dispatch disconnect to waiting threads to generate a proper exception. - _packetDispatcher.Dispatch(packet); - - if (!DisconnectIsPending()) - { - await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false); - } + await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false); } else if (packet is MqttAuthPacket authPacket) { - var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; - if (extendedAuthenticationExchangeHandler != null) - { - await extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)).ConfigureAwait(false); - } + await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false); } else { @@ -685,6 +674,57 @@ namespace MQTTnet.Client } } + Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken) + { + if (!_packetDispatcher.TryDispatch(pubRecPacket)) + { + // The packet is unknown. Probably due to a restart of the client. + // So wen send this to the server to trigger a full resend of the message. + return SendAsync(new MqttPubRelPacket + { + PacketIdentifier = pubRecPacket.PacketIdentifier, + ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound + }, cancellationToken); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) + { + return SendAsync(new MqttPubCompPacket + { + PacketIdentifier = pubRelPacket.PacketIdentifier, + ReasonCode = MqttPubCompReasonCode.Success + }, cancellationToken); + } + + Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) + { + _disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection); + + // Also dispatch disconnect to waiting threads to generate a proper exception. + _packetDispatcher.Dispatch(disconnectPacket); + + if (!DisconnectIsPending()) + { + return DisconnectInternalAsync(_packetReceiverTask, null, null); + } + + return PlatformAbstractionLayer.CompletedTask; + } + + Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket) + { + var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; + if (extendedAuthenticationExchangeHandler != null) + { + return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)); + } + + return PlatformAbstractionLayer.CompletedTask; + } + async Task PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs index 8844a967..306ca45c 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs @@ -19,7 +19,7 @@ namespace MQTTnet.PacketDispatcher _awaiters.Clear(); } - public void Dispatch(MqttBasePacket packet) + public bool TryDispatch(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -30,7 +30,7 @@ namespace MQTTnet.PacketDispatcher packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); } - return; + return true; } ushort identifier = 0; @@ -45,10 +45,20 @@ namespace MQTTnet.PacketDispatcher if (_awaiters.TryRemove(key, out var awaiter)) { awaiter.Complete(packet); - return; + return true; } - throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time."); + return false; + } + + public void Dispatch(MqttBasePacket packet) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + + if (!TryDispatch(packet)) + { + throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time."); + } } public void Cancel()