Make packet inspector fully async (#1941)

* Make packet inspector methods async.

* Update release notes
This commit is contained in:
Christian
2024-03-02 09:28:56 +01:00
committed by GitHub
parent f82d382f23
commit c204f6d8b2
3 changed files with 43 additions and 33 deletions
+1 -5
View File
@@ -1,5 +1 @@
* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919). * [Core] The package inspector is now fully async (#1941).
* [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).
+12 -3
View File
@@ -141,7 +141,8 @@ namespace MQTTnet.Adapter
try try
{ {
PacketInspector?.BeginReceivePacket(); var localPacketInspector = PacketInspector;
localPacketInspector?.BeginReceivePacket();
ReceivedMqttPacket receivedPacket; ReceivedMqttPacket receivedPacket;
var receivedPacketTask = ReceiveAsync(cancellationToken); var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -159,7 +160,10 @@ namespace MQTTnet.Adapter
return null; return null;
} }
PacketInspector?.EndReceivePacket(); if (localPacketInspector != null)
{
await localPacketInspector.EndReceivePacket().ConfigureAwait(false);
}
Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength); Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
@@ -215,8 +219,13 @@ namespace MQTTnet.Adapter
try try
{ {
var packetBuffer = PacketFormatterAdapter.Encode(packet); 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); _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation) if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
+30 -25
View File
@@ -4,6 +4,7 @@
using System; using System;
using System.IO; using System.IO;
using System.Threading.Tasks;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
@@ -12,16 +13,20 @@ namespace MQTTnet.Adapter
{ {
public sealed class MqttPacketInspector public sealed class MqttPacketInspector
{ {
readonly MqttNetSourceLogger _logger;
readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent; readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent;
readonly MqttNetSourceLogger _logger;
MemoryStream _receivedPacketBuffer; MemoryStream _receivedPacketBuffer;
public MqttPacketInspector(AsyncEvent<InspectMqttPacketEventArgs> asyncEvent, IMqttNetLogger logger) public MqttPacketInspector(AsyncEvent<InspectMqttPacketEventArgs> asyncEvent, IMqttNetLogger logger)
{ {
_asyncEvent = asyncEvent ?? throw new ArgumentNullException(nameof(asyncEvent)); _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)); _logger = logger.WithSource(nameof(MqttPacketInspector));
} }
@@ -36,28 +41,15 @@ namespace MQTTnet.Adapter
{ {
_receivedPacketBuffer = new MemoryStream(); _receivedPacketBuffer = new MemoryStream();
} }
_receivedPacketBuffer?.SetLength(0); _receivedPacketBuffer?.SetLength(0);
} }
public void EndReceivePacket() public Task BeginSendPacket(MqttPacketBuffer buffer)
{ {
if (!_asyncEvent.HasHandlers) if (!_asyncEvent.HasHandlers)
{ {
return; return CompletedTask.Instance;
}
var buffer = _receivedPacketBuffer.ToArray();
_receivedPacketBuffer.SetLength(0);
InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
}
public void BeginSendPacket(MqttPacketBuffer buffer)
{
if (!_asyncEvent.HasHandlers)
{
return;
} }
// Create a copy of the actual packet so that the inspector gets no access // 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. // intended for debugging etc. so that this is OK.
var bufferCopy = buffer.ToArray(); 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) public void FillReceiveBuffer(byte[] buffer)
@@ -74,16 +79,16 @@ namespace MQTTnet.Adapter
{ {
return; return;
} }
_receivedPacketBuffer?.Write(buffer, 0, buffer.Length); _receivedPacketBuffer?.Write(buffer, 0, buffer.Length);
} }
void InspectPacket(byte[] buffer, MqttPacketFlowDirection direction) async Task InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
{ {
try try
{ {
var eventArgs = new InspectMqttPacketEventArgs(direction, buffer); var eventArgs = new InspectMqttPacketEventArgs(direction, buffer);
_asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult(); await _asyncEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
} }
catch (Exception exception) catch (Exception exception)
{ {