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()