No longer fire _ClientAcknowledgedPublishPacketEvent_ for PUBREC packets (#1550) (#1551)

This commit is contained in:
Christian
2022-11-06 10:29:49 +01:00
committed by GitHub
parent b1a336eb62
commit 7ef137e26e
5 changed files with 22 additions and 32 deletions
+1
View File
@@ -0,0 +1 @@
* [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550)
+1
View File
@@ -242,5 +242,6 @@ See the LICENSE file in the project root for more information.</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@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/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBCOMP/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=Tnet/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/UserDictionary/Words/=Tnet/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary> <s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+2 -11
View File
@@ -99,25 +99,16 @@ namespace MQTTnet.Tests.Server
await LongTestDelay(); await LongTestDelay();
Assert.AreEqual(2, eventArgs.Count); Assert.AreEqual(1, eventArgs.Count);
var firstEvent = eventArgs[0]; var firstEvent = eventArgs[0];
Assert.IsNotNull(firstEvent); Assert.IsNotNull(firstEvent);
Assert.IsNotNull(firstEvent.PublishPacket); Assert.IsNotNull(firstEvent.PublishPacket);
Assert.IsNotNull(firstEvent.AcknowledgePacket); Assert.IsNotNull(firstEvent.AcknowledgePacket);
Assert.IsFalse(firstEvent.IsCompleted); Assert.IsTrue(firstEvent.IsCompleted);
Assert.AreEqual("A", firstEvent.PublishPacket.Topic); Assert.AreEqual("A", firstEvent.PublishPacket.Topic);
var secondEvent = eventArgs[1];
Assert.IsNotNull(secondEvent);
Assert.IsNotNull(secondEvent.PublishPacket);
Assert.IsNotNull(secondEvent.AcknowledgePacket);
Assert.IsTrue(secondEvent.IsCompleted);
Assert.AreEqual("A", secondEvent.PublishPacket.Topic);
} }
} }
} }
@@ -10,15 +10,23 @@ namespace MQTTnet.Server
{ {
public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
{ {
public ClientAcknowledgedPublishPacketEventArgs(string clientId, IDictionary sessionItems, MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
AcknowledgePacket = acknowledgePacket ?? throw new ArgumentNullException(nameof(acknowledgePacket));
}
/// <summary> /// <summary>
/// Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet. /// Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet.
/// </summary> /// </summary>
public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; } public MqttPacketWithIdentifier AcknowledgePacket { get; }
/// <summary> /// <summary>
/// Gets the ID of the client which acknowledged a PUBLISH packet. /// Gets the ID of the client which acknowledged a PUBLISH packet.
/// </summary> /// </summary>
public string ClientId { get; internal set; } public string ClientId { get; }
/// <summary> /// <summary>
/// Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2. /// Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2.
@@ -28,11 +36,11 @@ namespace MQTTnet.Server
/// <summary> /// <summary>
/// Gets the PUBLISH packet which was acknowledged. /// Gets the PUBLISH packet which was acknowledged.
/// </summary> /// </summary>
public MqttPublishPacket PublishPacket { get; internal set; } public MqttPublishPacket PublishPacket { get; }
/// <summary> /// <summary>
/// Gets the session items which contain custom user data per session. /// Gets the session items which contain custom user data per session.
/// </summary> /// </summary>
public IDictionary SessionItems { get; internal set; } public IDictionary SessionItems { get; }
} }
} }
+6 -17
View File
@@ -11,7 +11,6 @@ using MQTTnet.Client;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.PacketDispatcher; using MQTTnet.PacketDispatcher;
using MQTTnet.Packets; using MQTTnet.Packets;
@@ -170,14 +169,7 @@ namespace MQTTnet.Server
{ {
if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers) if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
{ {
var eventArgs = new ClientAcknowledgedPublishPacketEventArgs var eventArgs = new ClientAcknowledgedPublishPacketEventArgs(Id, Session.Items, publishPacket, acknowledgePacket);
{
PublishPacket = publishPacket,
AcknowledgePacket = acknowledgePacket,
ClientId = Id,
SessionItems = Session.Items
};
return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger); return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
} }
@@ -272,17 +264,14 @@ namespace MQTTnet.Server
} }
} }
async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
{ {
var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier); // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
// properly when the client has sent the PUBCOMP packet.
if (acknowledgedPublishPacket != null)
{
await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false);
}
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
return CompletedTask.Instance;
} }
void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)