Add event which is fired before a publish packet is released

This commit is contained in:
Christian
2022-11-21 21:35:36 +01:00
parent 301c4411e5
commit d003ae1f34
11 changed files with 98 additions and 56 deletions
+1
View File
@@ -243,5 +243,6 @@ See the LICENSE file in the project root for more information.</s:String>
<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/=PUBCOMP/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBREC/@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>
@@ -26,7 +26,7 @@ public static class PackageInspection_Samples
.WithTcpServer("broker.hivemq.com") .WithTcpServer("broker.hivemq.com")
.Build(); .Build();
mqttClient.InspectPackage += OnInspectPackage; mqttClient.InspectPackageAsync += OnInspectPackage;
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
@@ -30,7 +30,7 @@ namespace MQTTnet.Tests.Diagnostics
var packets = new List<string>(); var packets = new List<string>();
mqttClient.InspectPackage += eventArgs => mqttClient.InspectPackageAsync += eventArgs =>
{ {
packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer)); packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer));
return CompletedTask.Instance; return CompletedTask.Instance;
+1 -1
View File
@@ -954,7 +954,7 @@ namespace MQTTnet.Tests.Server
var client = testEnvironment.CreateClient(); var client = testEnvironment.CreateClient();
client.InspectPackage += e => client.InspectPackageAsync += e =>
{ {
if (e.Buffer.Length > 0) if (e.Buffer.Length > 0)
{ {
+3 -1
View File
@@ -9,13 +9,15 @@ namespace MQTTnet.Client
{ {
event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync; event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
event Func<MqttClientBeforeReleasePublishPacketEventArgs, Task> BeforeReleasePublishPacketAsync;
event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync; event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync; event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync;
event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync; event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
event Func<InspectMqttPacketEventArgs, Task> InspectPackage; event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
bool IsConnected { get; } bool IsConnected { get; }
+32 -5
View File
@@ -21,6 +21,7 @@ namespace MQTTnet.Client
readonly IMqttClientAdapterFactory _adapterFactory; readonly IMqttClientAdapterFactory _adapterFactory;
readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>(); readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
readonly AsyncEvent<MqttClientBeforeReleasePublishPacketEventArgs> _beforeReleasePublishPacketEvent = new AsyncEvent<MqttClientBeforeReleasePublishPacketEventArgs>();
readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory(); readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory();
@@ -63,6 +64,12 @@ namespace MQTTnet.Client
remove => _applicationMessageReceivedEvent.RemoveHandler(value); remove => _applicationMessageReceivedEvent.RemoveHandler(value);
} }
public event Func<MqttClientBeforeReleasePublishPacketEventArgs, Task> BeforeReleasePublishPacketAsync
{
add => _beforeReleasePublishPacketEvent.AddHandler(value);
remove => _beforeReleasePublishPacketEvent.RemoveHandler(value);
}
public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
{ {
add => _connectedEvent.AddHandler(value); add => _connectedEvent.AddHandler(value);
@@ -81,7 +88,7 @@ namespace MQTTnet.Client
remove => _disconnectedEvent.RemoveHandler(value); remove => _disconnectedEvent.RemoveHandler(value);
} }
public event Func<InspectMqttPacketEventArgs, Task> InspectPackage public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
{ {
add => _inspectPacketEvent.AddHandler(value); add => _inspectPacketEvent.AddHandler(value);
remove => _inspectPacketEvent.RemoveHandler(value); remove => _inspectPacketEvent.RemoveHandler(value);
@@ -259,7 +266,9 @@ namespace MQTTnet.Client
} }
} }
public async Task<MqttClientReleasePublishPacketResult> ReleasePublishPacketAsync(MqttClientReleasePublishPacketOptions options, CancellationToken cancellationToken = default) public async Task<MqttClientReleasePublishPacketResult> ReleasePublishPacketAsync(
MqttClientReleasePublishPacketOptions options,
CancellationToken cancellationToken = default)
{ {
if (options == null) 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."); 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 pubRelPacket = _packetFactories.PubRel.Create(options);
var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
MqttPubCompPacket pubCompPacket;
if (cancellationToken.CanBeCanceled)
{
pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
}
else
{
using (var timeout = new CancellationTokenSource(Options.Timeout))
{
pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, timeout.Token).ConfigureAwait(false);
}
}
return new MqttClientReleasePublishPacketResult(pubCompPacket.PacketIdentifier, pubCompPacket.ReasonString, pubCompPacket.UserProperties); 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. // 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. // 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); return SendAsync(pubRelPacket, cancellationToken);
} }
@@ -669,7 +693,10 @@ namespace MQTTnet.Client
return _clientPublishResultFactory.Create(pubRecPacket); 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<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false); var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
} }
@@ -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; }
}
}
@@ -9,6 +9,14 @@ namespace MQTTnet.Client
{ {
public sealed class MqttClientPublishResult public sealed class MqttClientPublishResult
{ {
public MqttClientPublishResult(ushort? packetIdentifier, MqttClientPublishReasonCode reasonCode, string reasonString, IReadOnlyCollection<MqttUserProperty> userProperties)
{
PacketIdentifier = packetIdentifier;
ReasonCode = reasonCode;
ReasonString = reasonString;
UserProperties = userProperties;
}
/// <summary> /// <summary>
/// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or /// 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 /// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit
@@ -19,19 +27,19 @@ namespace MQTTnet.Client
/// <summary> /// <summary>
/// Gets the packet identifier which was used for this publish. /// Gets the packet identifier which was used for this publish.
/// </summary> /// </summary>
public ushort? PacketIdentifier { get; set; } public ushort? PacketIdentifier { get; }
/// <summary> /// <summary>
/// Gets or sets the reason code. /// Gets or sets the reason code.
/// Hint: MQTT 5 feature only. /// Hint: MQTT 5 feature only.
/// </summary> /// </summary>
public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success; public MqttClientPublishReasonCode ReasonCode { get; }
/// <summary> /// <summary>
/// Gets or sets the reason string. /// Gets or sets the reason string.
/// Hint: MQTT 5 feature only. /// Hint: MQTT 5 feature only.
/// </summary> /// </summary>
public string ReasonString { get; set; } public string ReasonString { get; }
/// <summary> /// <summary>
/// Gets or sets the user properties. /// Gets or sets the user properties.
@@ -42,6 +50,6 @@ namespace MQTTnet.Client
/// The feature is very similar to the HTTP header concept. /// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only. /// Hint: MQTT 5 feature only.
/// </summary> /// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; } public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
} }
} }
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic; using System.Collections.Generic;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
@@ -10,8 +11,8 @@ namespace MQTTnet.Client
{ {
public sealed class MqttClientPublishResultFactory public sealed class MqttClientPublishResultFactory
{ {
static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult();
static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>(); static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>();
static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult(null, MqttClientPublishReasonCode.Success, null, EmptyUserProperties);
public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket) public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket)
{ {
@@ -21,70 +22,55 @@ namespace MQTTnet.Client
return EmptySuccessResult; return EmptySuccessResult;
} }
var result = new MqttClientPublishResult var result = new MqttClientPublishResult(
{ pubAckPacket.PacketIdentifier,
// Both enums have the same values. So it can be easily converted. (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode,
ReasonCode = (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode, pubAckPacket.ReasonString,
PacketIdentifier = pubAckPacket.PacketIdentifier, pubAckPacket.UserProperties ?? EmptyUserProperties);
ReasonString = pubAckPacket.ReasonString,
UserProperties = pubAckPacket.UserProperties ?? EmptyUserProperties
};
return result; return result;
} }
public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket) public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket)
{ {
return new MqttClientPublishResult if (pubRecPacket == null)
{ {
PacketIdentifier = pubRecPacket.PacketIdentifier, throw new ArgumentNullException(nameof(pubRecPacket));
ReasonCode = (MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode, }
ReasonString = pubRecPacket.ReasonString,
UserProperties = pubRecPacket.UserProperties return new MqttClientPublishResult(
}; pubRecPacket.PacketIdentifier,
(MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode,
pubRecPacket.ReasonString,
pubRecPacket.UserProperties ?? EmptyUserProperties);
} }
public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket) public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
{ {
if (pubRecPacket == null || pubCompPacket == null) if (pubRecPacket == null || pubCompPacket == null)
{ {
return new MqttClientPublishResult return new MqttClientPublishResult(null, MqttClientPublishReasonCode.UnspecifiedError, null, EmptyUserProperties);
{
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError
};
} }
MqttClientPublishResult result;
// The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC. // The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC.
if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound) if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound)
{ {
result = new MqttClientPublishResult return new MqttClientPublishResult(
{ pubCompPacket.PacketIdentifier,
PacketIdentifier = pubCompPacket.PacketIdentifier, MqttClientPublishReasonCode.UnspecifiedError,
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError, pubCompPacket.ReasonString,
ReasonString = pubCompPacket.ReasonString, pubCompPacket.UserProperties ?? EmptyUserProperties);
UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties
};
return result;
} }
result = new MqttClientPublishResult var reasonCode = MqttClientPublishReasonCode.Success;
{
PacketIdentifier = pubCompPacket.PacketIdentifier,
ReasonCode = MqttClientPublishReasonCode.Success,
ReasonString = pubCompPacket.ReasonString,
UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties
};
if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success) if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success)
{ {
// Both enums share the same values. // 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);
} }
} }
} }
@@ -11,7 +11,7 @@ namespace MQTTnet.Formatter
{ {
public sealed class MqttPubRelPacketFactory public sealed class MqttPubRelPacketFactory
{ {
public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode) public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttPubRelReasonCode reasonCode)
{ {
if (pubRecPacket == null) if (pubRecPacket == null)
{ {
@@ -21,7 +21,7 @@ namespace MQTTnet.Formatter
return new MqttPubRelPacket return new MqttPubRelPacket
{ {
PacketIdentifier = pubRecPacket.PacketIdentifier, PacketIdentifier = pubRecPacket.PacketIdentifier,
ReasonCode = (MqttPubRelReasonCode)(int)reasonCode ReasonCode = reasonCode
}; };
} }
+1 -1
View File
@@ -273,7 +273,7 @@ namespace MQTTnet.Server
{ {
// Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished // Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
// properly when the client has sent the PUBCOMP packet. // 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)); Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
return CompletedTask.Instance; return CompletedTask.Instance;