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

Avoid a split MqttPacketBuffer transfer caused by triggering PipeWriter.WriteAsync twice. (#1609)

Browse Source 
Avoid a split MqttPacketBuffer transfer caused by triggering PipeWriter.FlushAsync twice.
This commit is contained in:
老九
2022-12-05 19:19:31 +01:00
committed by GitHub
parent 099209fff2
commit 09e47fb5eb
1 changed files with 24 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+24 -26
View File
@@ -2,11 +2,6 @@
// 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.IO.Pipelines;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Connections.Features; using Microsoft.AspNetCore.Http.Connections.Features;
using MQTTnet.Adapter; using MQTTnet.Adapter;
@@ -15,12 +10,16 @@ using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using System;
using System.IO.Pipelines;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.AspNetCore namespace MQTTnet.AspNetCore
{ {
public sealed class MqttConnectionContext : IMqttChannelAdapter public sealed class MqttConnectionContext : IMqttChannelAdapter
{ {
readonly bool? _isOverWebSocket;
readonly AsyncLock _writerLock = new AsyncLock(); readonly AsyncLock _writerLock = new AsyncLock();
PipeReader _input; PipeReader _input;
PipeWriter _output; PipeWriter _output;
@@ -30,12 +29,6 @@ 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?.HttpContext != null)
{
_isOverWebSocket = feature.HttpContext.WebSockets.IsWebSocketRequest;
}
_input = Connection.Transport.Input; _input = Connection.Transport.Input;
_output = Connection.Transport.Output; _output = Connection.Transport.Output;
} }
@@ -181,20 +174,9 @@ namespace MQTTnet.AspNetCore
try try
{ {
var buffer = PacketFormatterAdapter.Encode(packet); var buffer = PacketFormatterAdapter.Encode(packet);
if (_isOverWebSocket == false) WritePacketBuffer(_output, buffer);
{ await _output.FlushAsync(cancellationToken).ConfigureAwait(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);
}
BytesSent += buffer.Length; BytesSent += buffer.Length;
} }
@@ -204,5 +186,21 @@ namespace MQTTnet.AspNetCore
} }
} }
} }
/// <summary>
/// copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter
/// MqttPacket will be transmitted within the bounds of a WebSocket frame after PipeWriter.FlushAsync
/// </summary>
/// <param name="output"></param>
/// <param name="buffer"></param>
private static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
{
var span = output.GetSpan(buffer.Length);
buffer.Packet.AsSpan().CopyTo(span);
buffer.Payload.AsSpan().CopyTo(span.Slice(buffer.Packet.Count));
output.Advance(buffer.Length);
}
} }
} }