From c204f6d8b264a304e506851ee59f852fb8020de5 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 2 Mar 2024 09:28:56 +0100 Subject: [PATCH] Make packet inspector fully async (#1941) * Make packet inspector methods async. * Update release notes --- .github/workflows/ReleaseNotes.md | 6 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 15 ++++- Source/MQTTnet/Adapter/MqttPacketInspector.cs | 55 ++++++++++--------- 3 files changed, 43 insertions(+), 33 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index f8f5777c..4983fc8d 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,5 +1 @@ -* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919). -* [ManagedClient] Added support for intercepting (and cancelling) publish messages (#1915, thanks to @hannasm). -* [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj). -* [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883). -* [Server] Fixed "service not registered" exception in ASP.NET integration (#1889). +* [Core] The package inspector is now fully async (#1941). diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 9015825b..fcfffc99 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -141,7 +141,8 @@ namespace MQTTnet.Adapter try { - PacketInspector?.BeginReceivePacket(); + var localPacketInspector = PacketInspector; + localPacketInspector?.BeginReceivePacket(); ReceivedMqttPacket receivedPacket; var receivedPacketTask = ReceiveAsync(cancellationToken); @@ -159,7 +160,10 @@ namespace MQTTnet.Adapter return null; } - PacketInspector?.EndReceivePacket(); + if (localPacketInspector != null) + { + await localPacketInspector.EndReceivePacket().ConfigureAwait(false); + } Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength); @@ -215,8 +219,13 @@ namespace MQTTnet.Adapter try { var packetBuffer = PacketFormatterAdapter.Encode(packet); - PacketInspector?.BeginSendPacket(packetBuffer); + var localPacketInspector = PacketInspector; + if (localPacketInspector != null) + { + await localPacketInspector.BeginSendPacket(packetBuffer).ConfigureAwait(false); + } + _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet); if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation) diff --git a/Source/MQTTnet/Adapter/MqttPacketInspector.cs b/Source/MQTTnet/Adapter/MqttPacketInspector.cs index a83de0e2..4f0c089e 100644 --- a/Source/MQTTnet/Adapter/MqttPacketInspector.cs +++ b/Source/MQTTnet/Adapter/MqttPacketInspector.cs @@ -4,6 +4,7 @@ using System; using System.IO; +using System.Threading.Tasks; using MQTTnet.Diagnostics; using MQTTnet.Formatter; using MQTTnet.Internal; @@ -12,16 +13,20 @@ namespace MQTTnet.Adapter { public sealed class MqttPacketInspector { - readonly MqttNetSourceLogger _logger; readonly AsyncEvent _asyncEvent; - + readonly MqttNetSourceLogger _logger; + MemoryStream _receivedPacketBuffer; - + public MqttPacketInspector(AsyncEvent asyncEvent, IMqttNetLogger logger) { _asyncEvent = asyncEvent ?? throw new ArgumentNullException(nameof(asyncEvent)); - - if (logger == null) throw new ArgumentNullException(nameof(logger)); + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + _logger = logger.WithSource(nameof(MqttPacketInspector)); } @@ -36,28 +41,15 @@ namespace MQTTnet.Adapter { _receivedPacketBuffer = new MemoryStream(); } - + _receivedPacketBuffer?.SetLength(0); } - public void EndReceivePacket() + public Task BeginSendPacket(MqttPacketBuffer buffer) { if (!_asyncEvent.HasHandlers) { - return; - } - - var buffer = _receivedPacketBuffer.ToArray(); - _receivedPacketBuffer.SetLength(0); - - InspectPacket(buffer, MqttPacketFlowDirection.Inbound); - } - - public void BeginSendPacket(MqttPacketBuffer buffer) - { - if (!_asyncEvent.HasHandlers) - { - return; + return CompletedTask.Instance; } // Create a copy of the actual packet so that the inspector gets no access @@ -65,7 +57,20 @@ namespace MQTTnet.Adapter // intended for debugging etc. so that this is OK. var bufferCopy = buffer.ToArray(); - InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound); + return InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound); + } + + public Task EndReceivePacket() + { + if (!_asyncEvent.HasHandlers) + { + return CompletedTask.Instance; + } + + var buffer = _receivedPacketBuffer.ToArray(); + _receivedPacketBuffer.SetLength(0); + + return InspectPacket(buffer, MqttPacketFlowDirection.Inbound); } public void FillReceiveBuffer(byte[] buffer) @@ -74,16 +79,16 @@ namespace MQTTnet.Adapter { return; } - + _receivedPacketBuffer?.Write(buffer, 0, buffer.Length); } - void InspectPacket(byte[] buffer, MqttPacketFlowDirection direction) + async Task InspectPacket(byte[] buffer, MqttPacketFlowDirection direction) { try { var eventArgs = new InspectMqttPacketEventArgs(direction, buffer); - _asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult(); + await _asyncEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) {