YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-19 02:53:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Optimize SendPacketAsync method for MQTTnet.AspNetCore (#1582)

Browse Source 
* Avoid to call MqttPacketBuffer.Join() when SendPacketAsync

* Safely cleanup the PacketFormatterAdapter

* Keep buffer.Join when mqtt over webSocket

* null checking for IHttpContextFeature.HttpContext
This commit is contained in:
老九
2022-11-28 21:02:13 +01:00
committed by GitHub
parent e45b207375
commit 41df2ec5da
1 changed files with 31 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+29 -12
View File
@@ -8,20 +8,20 @@ using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp; using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using System; using System;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Internal;
namespace MQTTnet.AspNetCore namespace MQTTnet.AspNetCore
{ {
public sealed class MqttConnectionContext : IMqttChannelAdapter public sealed class MqttConnectionContext : IMqttChannelAdapter
{ {
readonly AsyncLock _writerLock = new AsyncLock(); readonly AsyncLock _writerLock = new AsyncLock();
readonly bool? _isOverWebSocket;
PipeReader _input; PipeReader _input;
PipeWriter _output; PipeWriter _output;
@@ -30,6 +30,12 @@ namespace MQTTnet.AspNetCore
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
Connection = connection ?? throw new ArgumentNullException(nameof(connection)); Connection = connection ?? throw new ArgumentNullException(nameof(connection));
var feature = connection.Features.Get<IHttpContextFeature>();
if (feature != null && feature.HttpContext != null)
{
_isOverWebSocket = feature.HttpContext.WebSockets.IsWebSocketRequest;
}
if (Connection.Transport != null) if (Connection.Transport != null)
{ {
_input = Connection.Transport.Input; _input = Connection.Transport.Input;
@@ -169,20 +175,31 @@ namespace MQTTnet.AspNetCore
public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken) public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
{ {
var formatter = PacketFormatterAdapter;
using (await _writerLock.EnterAsync(cancellationToken).ConfigureAwait(false)) using (await _writerLock.EnterAsync(cancellationToken).ConfigureAwait(false))
{ {
var buffer = formatter.Encode(packet); try
var msg = buffer.Join().AsMemory();
var output = _output;
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
if (result.IsCompleted)
{ {
BytesSent += msg.Length; var buffer = PacketFormatterAdapter.Encode(packet);
} if (_isOverWebSocket == false)
{
await _output.WriteAsync(buffer.Packet, cancellationToken).ConfigureAwait(false);
if (buffer.Payload.Count > 0)
{
await _output.WriteAsync(buffer.Payload, cancellationToken).ConfigureAwait(false);
}
}
else
{
var bufferSegment = buffer.Join();
await _output.WriteAsync(bufferSegment, cancellationToken).ConfigureAwait(false);
}
PacketFormatterAdapter.Cleanup(); BytesSent += buffer.Length;
}
finally
{
PacketFormatterAdapter.Cleanup();
}
} }
} }