Fix issues in QoS 2 handling.

This commit is contained in:
Christian
2021-01-07 23:22:45 +01:00
parent 75845d6416
commit 3f5d627e9e
3 changed files with 78 additions and 24 deletions
+4
View File
@@ -17,6 +17,7 @@
* [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato). * [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato).
* [Core] Aligned the format of some log messages. * [Core] Aligned the format of some log messages.
* [Client] Fixed an issue with connection state detection (thanks to @tobiasfrick). * [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!). * [ManagedClient] Removed _AutoReconnect_ since this does no longer work (BREAKING CHANGE!).
* [RpcClient] Moved the RPC topic validation to the topic generation strategy. * [RpcClient] Moved the RPC topic validation to the topic generation strategy.
* [RcpClient] Added interface for _MqttRpcClient_. * [RcpClient] Added interface for _MqttRpcClient_.
@@ -68,6 +69,9 @@ Git commit: $gitCommit
<!-- Images --> <!-- Images -->
<file src="..\Images\Logo_128x128.png" target="images\" /> <file src="..\Images\Logo_128x128.png" target="images\" />
<!-- .NET 5.0 -->
<file src="..\Source\MQTTnet\bin\Release\net5.0\MQTTnet.*" target="lib\net5.0"/>
<!-- .NET Standard 1.3 --> <!-- .NET Standard 1.3 -->
<file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/> <file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
+60 -20
View File
@@ -17,6 +17,7 @@ using System;
using System.Diagnostics; using System.Diagnostics;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Implementations;
namespace MQTTnet.Client namespace MQTTnet.Client
{ {
@@ -551,37 +552,25 @@ namespace MQTTnet.Client
{ {
EnqueueReceivedPublishPacket(publishPacket); EnqueueReceivedPublishPacket(publishPacket);
} }
else if (packet is MqttPubRecPacket pubRecPacket)
{
await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false);
}
else if (packet is MqttPubRelPacket pubRelPacket) else if (packet is MqttPubRelPacket pubRelPacket)
{ {
await SendAsync(new MqttPubCompPacket await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
} }
else if (packet is MqttPingReqPacket) 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) else if (packet is MqttDisconnectPacket disconnectPacket)
{ {
_disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection); await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.Dispatch(packet);
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
}
} }
else if (packet is MqttAuthPacket authPacket) else if (packet is MqttAuthPacket authPacket)
{ {
var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler; await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
if (extendedAuthenticationExchangeHandler != null)
{
await extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)).ConfigureAwait(false);
}
} }
else 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<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{ {
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
@@ -19,7 +19,7 @@ namespace MQTTnet.PacketDispatcher
_awaiters.Clear(); _awaiters.Clear();
} }
public void Dispatch(MqttBasePacket packet) public bool TryDispatch(MqttBasePacket packet)
{ {
if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packet == null) throw new ArgumentNullException(nameof(packet));
@@ -30,7 +30,7 @@ namespace MQTTnet.PacketDispatcher
packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
} }
return; return true;
} }
ushort identifier = 0; ushort identifier = 0;
@@ -45,10 +45,20 @@ namespace MQTTnet.PacketDispatcher
if (_awaiters.TryRemove(key, out var awaiter)) if (_awaiters.TryRemove(key, out var awaiter))
{ {
awaiter.Complete(packet); 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() public void Cancel()