From d003ae1f342e0db7af470bcc5d69f0e8346c76c1 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Mon, 21 Nov 2022 21:35:36 +0100 Subject: [PATCH] Add event which is fired before a publish packet is released --- MQTTnet.sln.DotSettings | 1 + .../Diagnostics/PackageInspection_Samples.cs | 2 +- .../Diagnostics/PacketInspection_Tests.cs | 2 +- Source/MQTTnet.Tests/Server/General.cs | 2 +- Source/MQTTnet/Client/IMqttClient.cs | 4 +- Source/MQTTnet/Client/MqttClient.cs | 37 +++++++++-- ...ientBeforeReleasePublishPacketEventArgs.cs | 18 +++++ .../Publishing/MqttClientPublishResult.cs | 16 +++-- .../MqttClientPublishResultFactory.cs | 66 ++++++++----------- .../Formatter/MqttPubRelPacketFactory.cs | 4 +- Source/MQTTnet/Server/Internal/MqttClient.cs | 2 +- 11 files changed, 98 insertions(+), 56 deletions(-) create mode 100644 Source/MQTTnet/Client/Publishing/MqttClientBeforeReleasePublishPacketEventArgs.cs diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings index 83eed4bd..f6c48da4 100644 --- a/MQTTnet.sln.DotSettings +++ b/MQTTnet.sln.DotSettings @@ -243,5 +243,6 @@ See the LICENSE file in the project root for more information. True True True + True True True \ No newline at end of file diff --git a/Samples/Diagnostics/PackageInspection_Samples.cs b/Samples/Diagnostics/PackageInspection_Samples.cs index c4d4b7ac..f1eee1ef 100644 --- a/Samples/Diagnostics/PackageInspection_Samples.cs +++ b/Samples/Diagnostics/PackageInspection_Samples.cs @@ -26,7 +26,7 @@ public static class PackageInspection_Samples .WithTcpServer("broker.hivemq.com") .Build(); - mqttClient.InspectPackage += OnInspectPackage; + mqttClient.InspectPackageAsync += OnInspectPackage; await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); diff --git a/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs b/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs index 5589a37d..48b38162 100644 --- a/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs +++ b/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs @@ -30,7 +30,7 @@ namespace MQTTnet.Tests.Diagnostics var packets = new List(); - mqttClient.InspectPackage += eventArgs => + mqttClient.InspectPackageAsync += eventArgs => { packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer)); return CompletedTask.Instance; diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index eeb75fc6..cc74db79 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -954,7 +954,7 @@ namespace MQTTnet.Tests.Server var client = testEnvironment.CreateClient(); - client.InspectPackage += e => + client.InspectPackageAsync += e => { if (e.Buffer.Length > 0) { diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index f0787e75..ed01d7ad 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -9,13 +9,15 @@ namespace MQTTnet.Client { event Func ApplicationMessageReceivedAsync; + event Func BeforeReleasePublishPacketAsync; + event Func ConnectedAsync; event Func ConnectingAsync; event Func DisconnectedAsync; - event Func InspectPackage; + event Func InspectPackageAsync; bool IsConnected { get; } diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 4b752194..43c713af 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -21,6 +21,7 @@ namespace MQTTnet.Client readonly IMqttClientAdapterFactory _adapterFactory; readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); + readonly AsyncEvent _beforeReleasePublishPacketEvent = new AsyncEvent(); readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory(); readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); @@ -63,6 +64,12 @@ namespace MQTTnet.Client remove => _applicationMessageReceivedEvent.RemoveHandler(value); } + public event Func BeforeReleasePublishPacketAsync + { + add => _beforeReleasePublishPacketEvent.AddHandler(value); + remove => _beforeReleasePublishPacketEvent.RemoveHandler(value); + } + public event Func ConnectedAsync { add => _connectedEvent.AddHandler(value); @@ -81,7 +88,7 @@ namespace MQTTnet.Client remove => _disconnectedEvent.RemoveHandler(value); } - public event Func InspectPackage + public event Func InspectPackageAsync { add => _inspectPacketEvent.AddHandler(value); remove => _inspectPacketEvent.RemoveHandler(value); @@ -259,7 +266,9 @@ namespace MQTTnet.Client } } - public async Task ReleasePublishPacketAsync(MqttClientReleasePublishPacketOptions options, CancellationToken cancellationToken = default) + public async Task ReleasePublishPacketAsync( + MqttClientReleasePublishPacketOptions options, + CancellationToken cancellationToken = default) { if (options == null) { @@ -281,8 +290,23 @@ namespace MQTTnet.Client throw new InvalidOperationException("Manually releasing PUBLISH packets requires the feature _AutoReleasePublishPackets_ disabled in the client options."); } + var eventArgs = new MqttClientBeforeReleasePublishPacketEventArgs(options.PacketIdentifier); + await _beforeReleasePublishPacketEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + var pubRelPacket = _packetFactories.PubRel.Create(options); - var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + + MqttPubCompPacket pubCompPacket; + if (cancellationToken.CanBeCanceled) + { + pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + } + else + { + using (var timeout = new CancellationTokenSource(Options.Timeout)) + { + pubCompPacket = await SendAndReceiveAsync(pubRelPacket, timeout.Token).ConfigureAwait(false); + } + } return new MqttClientReleasePublishPacketResult(pubCompPacket.PacketIdentifier, pubCompPacket.ReasonString, pubCompPacket.UserProperties); } @@ -628,7 +652,7 @@ namespace MQTTnet.Client { // 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. - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.PacketIdentifierNotFound); return SendAsync(pubRelPacket, cancellationToken); } @@ -669,7 +693,10 @@ namespace MQTTnet.Client return _clientPublishResultFactory.Create(pubRecPacket); } - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + var eventArgs = new MqttClientBeforeReleasePublishPacketEventArgs(pubRecPacket.PacketIdentifier); + await _beforeReleasePublishPacketEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.Success); var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); } diff --git a/Source/MQTTnet/Client/Publishing/MqttClientBeforeReleasePublishPacketEventArgs.cs b/Source/MQTTnet/Client/Publishing/MqttClientBeforeReleasePublishPacketEventArgs.cs new file mode 100644 index 00000000..81757781 --- /dev/null +++ b/Source/MQTTnet/Client/Publishing/MqttClientBeforeReleasePublishPacketEventArgs.cs @@ -0,0 +1,18 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; + +namespace MQTTnet.Client +{ + public sealed class MqttClientBeforeReleasePublishPacketEventArgs : EventArgs + { + public MqttClientBeforeReleasePublishPacketEventArgs(ushort packetIdentifier) + { + PacketIdentifier = packetIdentifier; + } + + public ushort PacketIdentifier { get; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index 6fd2dcd8..240fcf36 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -9,6 +9,14 @@ namespace MQTTnet.Client { public sealed class MqttClientPublishResult { + public MqttClientPublishResult(ushort? packetIdentifier, MqttClientPublishReasonCode reasonCode, string reasonString, IReadOnlyCollection userProperties) + { + PacketIdentifier = packetIdentifier; + ReasonCode = reasonCode; + ReasonString = reasonString; + UserProperties = userProperties; + } + /// /// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or /// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit @@ -19,19 +27,19 @@ namespace MQTTnet.Client /// /// Gets the packet identifier which was used for this publish. /// - public ushort? PacketIdentifier { get; set; } + public ushort? PacketIdentifier { get; } /// /// Gets or sets the reason code. /// Hint: MQTT 5 feature only. /// - public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success; + public MqttClientPublishReasonCode ReasonCode { get; } /// /// Gets or sets the reason string. /// Hint: MQTT 5 feature only. /// - public string ReasonString { get; set; } + public string ReasonString { get; } /// /// Gets or sets the user properties. @@ -42,6 +50,6 @@ namespace MQTTnet.Client /// The feature is very similar to the HTTP header concept. /// Hint: MQTT 5 feature only. /// - public IReadOnlyCollection UserProperties { get; internal set; } + public IReadOnlyCollection UserProperties { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs index c8550a6b..17bcf065 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using System.Collections.Generic; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -10,8 +11,8 @@ namespace MQTTnet.Client { public sealed class MqttClientPublishResultFactory { - static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult(); static readonly IReadOnlyCollection EmptyUserProperties = new List(); + static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult(null, MqttClientPublishReasonCode.Success, null, EmptyUserProperties); public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket) { @@ -21,70 +22,55 @@ namespace MQTTnet.Client return EmptySuccessResult; } - var result = new MqttClientPublishResult - { - // Both enums have the same values. So it can be easily converted. - ReasonCode = (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode, - PacketIdentifier = pubAckPacket.PacketIdentifier, - ReasonString = pubAckPacket.ReasonString, - UserProperties = pubAckPacket.UserProperties ?? EmptyUserProperties - }; + var result = new MqttClientPublishResult( + pubAckPacket.PacketIdentifier, + (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode, + pubAckPacket.ReasonString, + pubAckPacket.UserProperties ?? EmptyUserProperties); return result; } public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket) { - return new MqttClientPublishResult + if (pubRecPacket == null) { - PacketIdentifier = pubRecPacket.PacketIdentifier, - ReasonCode = (MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode, - ReasonString = pubRecPacket.ReasonString, - UserProperties = pubRecPacket.UserProperties - }; + throw new ArgumentNullException(nameof(pubRecPacket)); + } + + return new MqttClientPublishResult( + pubRecPacket.PacketIdentifier, + (MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode, + pubRecPacket.ReasonString, + pubRecPacket.UserProperties ?? EmptyUserProperties); } - + public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) { if (pubRecPacket == null || pubCompPacket == null) { - return new MqttClientPublishResult - { - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError - }; + return new MqttClientPublishResult(null, MqttClientPublishReasonCode.UnspecifiedError, null, EmptyUserProperties); } - MqttClientPublishResult result; - // The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC. if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound) { - result = new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.UnspecifiedError, - ReasonString = pubCompPacket.ReasonString, - UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties - }; - - return result; + return new MqttClientPublishResult( + pubCompPacket.PacketIdentifier, + MqttClientPublishReasonCode.UnspecifiedError, + pubCompPacket.ReasonString, + pubCompPacket.UserProperties ?? EmptyUserProperties); } - result = new MqttClientPublishResult - { - PacketIdentifier = pubCompPacket.PacketIdentifier, - ReasonCode = MqttClientPublishReasonCode.Success, - ReasonString = pubCompPacket.ReasonString, - UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties - }; + var reasonCode = MqttClientPublishReasonCode.Success; if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) { // Both enums share the same values. - result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode; + reasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode; } - return result; + return new MqttClientPublishResult(pubCompPacket.PacketIdentifier, reasonCode, pubCompPacket.ReasonString, pubCompPacket.UserProperties ?? EmptyUserProperties); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs index 9b999c54..92c9d1fd 100644 --- a/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Formatter { public sealed class MqttPubRelPacketFactory { - public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode) + public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttPubRelReasonCode reasonCode) { if (pubRecPacket == null) { @@ -21,7 +21,7 @@ namespace MQTTnet.Formatter return new MqttPubRelPacket { PacketIdentifier = pubRecPacket.PacketIdentifier, - ReasonCode = (MqttPubRelReasonCode)(int)reasonCode + ReasonCode = reasonCode }; } diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index f505b00c..378353c3 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -273,7 +273,7 @@ namespace MQTTnet.Server { // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished // properly when the client has sent the PUBCOMP packet. - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.Success); Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); return CompletedTask.Instance;