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

MqttConnectionContext: Adapting mqtt over tcp (#1616)

Browse Source 
Adapting mqtt over tcp.
This commit is contained in:
老九
2022-12-07 19:36:13 +01:00
committed by GitHub
parent f8aa686557
commit c1c17ba162
1 changed files with 67 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+67 -28
View File
@@ -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<ITlsConnectionFeature>();
if (tlsFeature != null)
{
return tlsFeature.ClientCertificate;
}
// mqtt over websocket
var httpFeature = _connection.Features.Get<IHttpContextFeature>();
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<IHttpConnectionFeature>();
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<ITlsConnectionFeature>();
if (tlsFeature != null)
{
return true;
}
public bool IsSecureConnection => Http?.HttpContext?.Request.IsHttps ?? false;
// mqtt over websocket
var httpFeature = _connection.Features.Get<IHttpContextFeature>();
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<IHttpContextFeature>();
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<MqttPacket> 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);
}
}
}