From c1c17ba162265d77eeb5f43b6444174a65c442d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=80=81=E4=B9=9D?= <366193849@qq.com> Date: Thu, 8 Dec 2022 02:36:13 +0800 Subject: [PATCH] MqttConnectionContext: Adapting mqtt over tcp (#1616) Adapting mqtt over tcp. --- .../MqttConnectionContext.cs | 95 +++++++++++++------ 1 file changed, 67 insertions(+), 28 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index a57df2d9..137ed1c9 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -2,82 +2,123 @@ // 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; -using System.IO.Pipelines; -using System.Security.Cryptography.X509Certificates; -using System.Threading; -using System.Threading.Tasks; using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Http.Connections.Features; +using Microsoft.AspNetCore.Http.Features; using MQTTnet.Adapter; using MQTTnet.AspNetCore.Client.Tcp; using MQTTnet.Exceptions; using MQTTnet.Formatter; using MQTTnet.Internal; using MQTTnet.Packets; +using System; +using System.IO.Pipelines; +using System.Net; +using System.Security.Cryptography.X509Certificates; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.AspNetCore { public sealed class MqttConnectionContext : IMqttChannelAdapter { readonly AsyncLock _writerLock = new AsyncLock(); + readonly ConnectionContext _connection; PipeReader _input; PipeWriter _output; public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection) { PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); - Connection = connection ?? throw new ArgumentNullException(nameof(connection)); - - _input = Connection.Transport.Input; - _output = Connection.Transport.Output; + _connection = connection ?? throw new ArgumentNullException(nameof(connection)); + _input = connection.Transport.Input; + _output = connection.Transport.Output; } public long BytesReceived { get; private set; } public long BytesSent { get; private set; } - public X509Certificate2 ClientCertificate => Http?.HttpContext?.Connection?.ClientCertificate; + public X509Certificate2 ClientCertificate + { + get + { + // mqtt over tcp + var tlsFeature = _connection.Features.Get(); + if (tlsFeature != null) + { + return tlsFeature.ClientCertificate; + } + + // mqtt over websocket + var httpFeature = _connection.Features.Get(); + if (httpFeature != null && httpFeature.HttpContext != null) + { + return httpFeature.HttpContext.Connection.ClientCertificate; + } + return null; + } + } - public ConnectionContext Connection { get; } public string Endpoint { get { -#if NETCOREAPP3_1 - if (Connection?.RemoteEndPoint != null) + // mqtt over tcp +#if NETCOREAPP3_1_OR_GREATER + if (_connection.RemoteEndPoint != null) { - return Connection.RemoteEndPoint.ToString(); + return _connection.RemoteEndPoint.ToString(); } #endif - var connection = Http?.HttpContext?.Connection; - if (connection == null) + // mqtt over websocket + var httpFeature = _connection.Features.Get(); + if (httpFeature != null && httpFeature.RemoteIpAddress != null) { - return Connection.ConnectionId; + return new IPEndPoint(httpFeature.RemoteIpAddress, httpFeature.RemotePort).ToString(); } - return $"{connection.RemoteIpAddress}:{connection.RemotePort}"; + return null; } } - public bool IsReadingPacket { get; private set; } + public bool IsSecureConnection + { + get + { + // mqtt over tcp + var tlsFeature = _connection.Features.Get(); + if (tlsFeature != null) + { + return true; + } - public bool IsSecureConnection => Http?.HttpContext?.Request.IsHttps ?? false; + // mqtt over websocket + var httpFeature = _connection.Features.Get(); + if (httpFeature != null && httpFeature.HttpContext != null) + { + return httpFeature.HttpContext.Request.IsHttps; + } + return false; + } + } + + + public bool IsReadingPacket { get; private set; } public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } - IHttpContextFeature Http => Connection.Features.Get(); public async Task ConnectAsync(CancellationToken cancellationToken) { - if (Connection is TcpConnection tcp && !tcp.IsConnected) + if (_connection is TcpConnection tcp && !tcp.IsConnected) { await tcp.StartAsync().ConfigureAwait(false); } - _input = Connection.Transport.Input; - _output = Connection.Transport.Output; + _input = _connection.Transport.Input; + _output = _connection.Transport.Output; } public Task DisconnectAsync(CancellationToken cancellationToken) @@ -94,14 +135,12 @@ namespace MQTTnet.AspNetCore public async Task ReceivePacketAsync(CancellationToken cancellationToken) { - var input = Connection.Transport.Input; - try { while (!cancellationToken.IsCancellationRequested) { ReadResult readResult; - var readTask = input.ReadAsync(cancellationToken); + var readTask = _input.ReadAsync(cancellationToken); if (readTask.IsCompleted) { readResult = readTask.Result; @@ -141,7 +180,7 @@ namespace MQTTnet.AspNetCore // The buffer was sliced up to where it was consumed, so we can just advance to the start. // We mark examined as buffer.End so that if we didn't receive a full frame, we'll wait for more data // before yielding the read again. - input.AdvanceTo(consumed, observed); + _input.AdvanceTo(consumed, observed); } } }