mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 10:26:11 +08:00
moved files into the correct place
This commit is contained in:
JanEggers
@@ -0,0 +1,34 @@
|
||||
using System;
|
||||
using System.Net;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.AspNetCore.Client.Tcp;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client
|
||||
{
|
||||
public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
|
||||
{
|
||||
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion };
|
||||
|
||||
switch (options.ChannelOptions)
|
||||
{
|
||||
case MqttClientTcpOptions tcpOptions:
|
||||
{
|
||||
var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
|
||||
var tcpConnection = new TcpConnection(endpoint);
|
||||
return new MqttConnectionContext(serializer, tcpConnection);
|
||||
}
|
||||
default:
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
using System;
|
||||
using System.Runtime.InteropServices;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public static class BufferExtensions
|
||||
{
|
||||
public static ArraySegment<byte> GetArray(this Memory<byte> memory)
|
||||
{
|
||||
return ((ReadOnlyMemory<byte>)memory).GetArray();
|
||||
}
|
||||
|
||||
public static ArraySegment<byte> GetArray(this ReadOnlyMemory<byte> memory)
|
||||
{
|
||||
if (!MemoryMarshal.TryGetArray(memory, out var result))
|
||||
{
|
||||
throw new InvalidOperationException("Buffer backed by array was expected");
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
using System.IO.Pipelines;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public class DuplexPipe : IDuplexPipe
|
||||
{
|
||||
public DuplexPipe(PipeReader reader, PipeWriter writer)
|
||||
{
|
||||
Input = reader;
|
||||
Output = writer;
|
||||
}
|
||||
|
||||
public PipeReader Input { get; }
|
||||
|
||||
public PipeWriter Output { get; }
|
||||
|
||||
public static DuplexPipePair CreateConnectionPair(PipeOptions inputOptions, PipeOptions outputOptions)
|
||||
{
|
||||
var input = new Pipe(inputOptions);
|
||||
var output = new Pipe(outputOptions);
|
||||
|
||||
var transportToApplication = new DuplexPipe(output.Reader, input.Writer);
|
||||
var applicationToTransport = new DuplexPipe(input.Reader, output.Writer);
|
||||
|
||||
return new DuplexPipePair(applicationToTransport, transportToApplication);
|
||||
}
|
||||
|
||||
// This class exists to work around issues with value tuple on .NET Framework
|
||||
public readonly struct DuplexPipePair
|
||||
{
|
||||
public IDuplexPipe Transport { get; }
|
||||
public IDuplexPipe Application { get; }
|
||||
|
||||
public DuplexPipePair(IDuplexPipe transport, IDuplexPipe application)
|
||||
{
|
||||
Transport = transport;
|
||||
Application = application;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
using System;
|
||||
using System.Diagnostics;
|
||||
using System.IO.Pipelines;
|
||||
using System.Net.Sockets;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public class SocketAwaitable : ICriticalNotifyCompletion
|
||||
{
|
||||
private static readonly Action _callbackCompleted = () => { };
|
||||
|
||||
private readonly PipeScheduler _ioScheduler;
|
||||
|
||||
private Action _callback;
|
||||
private int _bytesTransferred;
|
||||
private SocketError _error;
|
||||
|
||||
public SocketAwaitable(PipeScheduler ioScheduler)
|
||||
{
|
||||
_ioScheduler = ioScheduler;
|
||||
}
|
||||
|
||||
public SocketAwaitable GetAwaiter() => this;
|
||||
public bool IsCompleted => ReferenceEquals(_callback, _callbackCompleted);
|
||||
|
||||
public int GetResult()
|
||||
{
|
||||
Debug.Assert(ReferenceEquals(_callback, _callbackCompleted));
|
||||
|
||||
_callback = null;
|
||||
|
||||
if (_error != SocketError.Success)
|
||||
{
|
||||
throw new SocketException((int)_error);
|
||||
}
|
||||
|
||||
return _bytesTransferred;
|
||||
}
|
||||
|
||||
public void OnCompleted(Action continuation)
|
||||
{
|
||||
if (ReferenceEquals(_callback, _callbackCompleted) ||
|
||||
ReferenceEquals(Interlocked.CompareExchange(ref _callback, continuation, null), _callbackCompleted))
|
||||
{
|
||||
Task.Run(continuation);
|
||||
}
|
||||
}
|
||||
|
||||
public void UnsafeOnCompleted(Action continuation)
|
||||
{
|
||||
OnCompleted(continuation);
|
||||
}
|
||||
|
||||
public void Complete(int bytesTransferred, SocketError socketError)
|
||||
{
|
||||
_error = socketError;
|
||||
_bytesTransferred = bytesTransferred;
|
||||
var continuation = Interlocked.Exchange(ref _callback, _callbackCompleted);
|
||||
|
||||
if (continuation != null)
|
||||
{
|
||||
_ioScheduler.Schedule(state => ((Action)state)(), continuation);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
using System;
|
||||
using System.IO.Pipelines;
|
||||
using System.Net.Sockets;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public class SocketReceiver
|
||||
{
|
||||
private readonly Socket _socket;
|
||||
private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
|
||||
private readonly SocketAwaitable _awaitable;
|
||||
|
||||
public SocketReceiver(Socket socket, PipeScheduler scheduler)
|
||||
{
|
||||
_socket = socket;
|
||||
_awaitable = new SocketAwaitable(scheduler);
|
||||
_eventArgs.UserToken = _awaitable;
|
||||
_eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
|
||||
}
|
||||
|
||||
public SocketAwaitable ReceiveAsync(Memory<byte> buffer)
|
||||
{
|
||||
#if NETCOREAPP2_1
|
||||
_eventArgs.SetBuffer(buffer);
|
||||
#else
|
||||
var segment = buffer.GetArray();
|
||||
|
||||
_eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
|
||||
#endif
|
||||
if (!_socket.ReceiveAsync(_eventArgs))
|
||||
{
|
||||
_awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
|
||||
}
|
||||
|
||||
return _awaitable;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.IO.Pipelines;
|
||||
using System.Net.Sockets;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public class SocketSender
|
||||
{
|
||||
private readonly Socket _socket;
|
||||
private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
|
||||
private readonly SocketAwaitable _awaitable;
|
||||
|
||||
private List<ArraySegment<byte>> _bufferList;
|
||||
|
||||
public SocketSender(Socket socket, PipeScheduler scheduler)
|
||||
{
|
||||
_socket = socket;
|
||||
_awaitable = new SocketAwaitable(scheduler);
|
||||
_eventArgs.UserToken = _awaitable;
|
||||
_eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
|
||||
}
|
||||
|
||||
public SocketAwaitable SendAsync(in ReadOnlySequence<byte> buffers)
|
||||
{
|
||||
if (buffers.IsSingleSegment)
|
||||
{
|
||||
return SendAsync(buffers.First);
|
||||
}
|
||||
|
||||
#if NETCOREAPP2_1
|
||||
if (!_eventArgs.MemoryBuffer.Equals(Memory<byte>.Empty))
|
||||
#else
|
||||
if (_eventArgs.Buffer != null)
|
||||
#endif
|
||||
{
|
||||
_eventArgs.SetBuffer(null, 0, 0);
|
||||
}
|
||||
|
||||
_eventArgs.BufferList = GetBufferList(buffers);
|
||||
|
||||
if (!_socket.SendAsync(_eventArgs))
|
||||
{
|
||||
_awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
|
||||
}
|
||||
|
||||
return _awaitable;
|
||||
}
|
||||
|
||||
private SocketAwaitable SendAsync(ReadOnlyMemory<byte> memory)
|
||||
{
|
||||
// The BufferList getter is much less expensive then the setter.
|
||||
if (_eventArgs.BufferList != null)
|
||||
{
|
||||
_eventArgs.BufferList = null;
|
||||
}
|
||||
|
||||
#if NETCOREAPP2_1
|
||||
_eventArgs.SetBuffer(MemoryMarshal.AsMemory(memory));
|
||||
#else
|
||||
var segment = memory.GetArray();
|
||||
|
||||
_eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
|
||||
#endif
|
||||
if (!_socket.SendAsync(_eventArgs))
|
||||
{
|
||||
_awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
|
||||
}
|
||||
|
||||
return _awaitable;
|
||||
}
|
||||
|
||||
private List<ArraySegment<byte>> GetBufferList(in ReadOnlySequence<byte> buffer)
|
||||
{
|
||||
Debug.Assert(!buffer.IsEmpty);
|
||||
Debug.Assert(!buffer.IsSingleSegment);
|
||||
|
||||
if (_bufferList == null)
|
||||
{
|
||||
_bufferList = new List<ArraySegment<byte>>();
|
||||
}
|
||||
else
|
||||
{
|
||||
// Buffers are pooled, so it's OK to root them until the next multi-buffer write.
|
||||
_bufferList.Clear();
|
||||
}
|
||||
|
||||
foreach (var b in buffer)
|
||||
{
|
||||
_bufferList.Add(b.GetArray());
|
||||
}
|
||||
|
||||
return _bufferList;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.IO.Pipelines;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.AspNetCore.Connections;
|
||||
using Microsoft.AspNetCore.Http.Features;
|
||||
using MQTTnet.Exceptions;
|
||||
|
||||
namespace MQTTnet.AspNetCore.Client.Tcp
|
||||
{
|
||||
public class TcpConnection : ConnectionContext
|
||||
{
|
||||
private volatile bool _aborted;
|
||||
private readonly EndPoint _endPoint;
|
||||
private SocketSender _sender;
|
||||
private SocketReceiver _receiver;
|
||||
|
||||
private Socket _socket;
|
||||
private IDuplexPipe _application;
|
||||
|
||||
|
||||
public bool IsConnected { get; private set; }
|
||||
|
||||
public override string ConnectionId { get; set; }
|
||||
|
||||
public override IFeatureCollection Features { get; }
|
||||
|
||||
public override IDictionary<object, object> Items { get; set; }
|
||||
public override IDuplexPipe Transport { get; set; }
|
||||
|
||||
public TcpConnection(EndPoint endPoint)
|
||||
{
|
||||
_endPoint = endPoint;
|
||||
}
|
||||
|
||||
public TcpConnection(Socket socket)
|
||||
{
|
||||
_socket = socket;
|
||||
_endPoint = socket.RemoteEndPoint;
|
||||
|
||||
_sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
|
||||
_receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
|
||||
}
|
||||
|
||||
public Task DisposeAsync()
|
||||
{
|
||||
IsConnected = false;
|
||||
|
||||
Transport?.Output.Complete();
|
||||
Transport?.Input.Complete();
|
||||
|
||||
_socket?.Dispose();
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task StartAsync()
|
||||
{
|
||||
if (_socket == null)
|
||||
{
|
||||
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
_sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
|
||||
_receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
|
||||
await _socket.ConnectAsync(_endPoint);
|
||||
}
|
||||
|
||||
var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
|
||||
|
||||
Transport = pair.Transport;
|
||||
_application = pair.Application;
|
||||
|
||||
_ = ExecuteAsync();
|
||||
|
||||
IsConnected = true;
|
||||
}
|
||||
|
||||
private async Task ExecuteAsync()
|
||||
{
|
||||
Exception sendError = null;
|
||||
try
|
||||
{
|
||||
// Spawn send and receive logic
|
||||
var receiveTask = DoReceive();
|
||||
var sendTask = DoSend();
|
||||
|
||||
// If the sending task completes then close the receive
|
||||
// We don't need to do this in the other direction because the kestrel
|
||||
// will trigger the output closing once the input is complete.
|
||||
if (await Task.WhenAny(receiveTask, sendTask) == sendTask)
|
||||
{
|
||||
// Tell the reader it's being aborted
|
||||
_socket.Dispose();
|
||||
}
|
||||
|
||||
// Now wait for both to complete
|
||||
await receiveTask;
|
||||
sendError = await sendTask;
|
||||
|
||||
// Dispose the socket(should noop if already called)
|
||||
_socket.Dispose();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"Unexpected exception in {nameof(TcpConnection)}.{nameof(StartAsync)}: " + ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Complete the output after disposing the socket
|
||||
_application.Input.Complete(sendError);
|
||||
}
|
||||
}
|
||||
private async Task DoReceive()
|
||||
{
|
||||
Exception error = null;
|
||||
|
||||
try
|
||||
{
|
||||
await ProcessReceives();
|
||||
}
|
||||
catch (SocketException ex) when (ex.SocketErrorCode == SocketError.ConnectionReset)
|
||||
{
|
||||
error = new MqttCommunicationException(ex);
|
||||
}
|
||||
catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted ||
|
||||
ex.SocketErrorCode == SocketError.ConnectionAborted ||
|
||||
ex.SocketErrorCode == SocketError.Interrupted ||
|
||||
ex.SocketErrorCode == SocketError.InvalidArgument)
|
||||
{
|
||||
if (!_aborted)
|
||||
{
|
||||
// Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix.
|
||||
error = ConnectionAborted();
|
||||
}
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
if (!_aborted)
|
||||
{
|
||||
error = ConnectionAborted();
|
||||
}
|
||||
}
|
||||
catch (IOException ex)
|
||||
{
|
||||
error = ex;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
error = new IOException(ex.Message, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (_aborted)
|
||||
{
|
||||
error = error ?? ConnectionAborted();
|
||||
}
|
||||
|
||||
_application.Output.Complete(error);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessReceives()
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
// Ensure we have some reasonable amount of buffer space
|
||||
var buffer = _application.Output.GetMemory();
|
||||
|
||||
var bytesReceived = await _receiver.ReceiveAsync(buffer);
|
||||
|
||||
if (bytesReceived == 0)
|
||||
{
|
||||
// FIN
|
||||
break;
|
||||
}
|
||||
|
||||
_application.Output.Advance(bytesReceived);
|
||||
|
||||
var flushTask = _application.Output.FlushAsync();
|
||||
|
||||
if (!flushTask.IsCompleted)
|
||||
{
|
||||
await flushTask;
|
||||
}
|
||||
|
||||
var result = flushTask.GetAwaiter().GetResult();
|
||||
if (result.IsCompleted)
|
||||
{
|
||||
// Pipe consumer is shut down, do we stop writing
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Exception ConnectionAborted()
|
||||
{
|
||||
return new MqttCommunicationException("Connection Aborted");
|
||||
}
|
||||
|
||||
private async Task<Exception> DoSend()
|
||||
{
|
||||
Exception error = null;
|
||||
|
||||
try
|
||||
{
|
||||
await ProcessSends();
|
||||
}
|
||||
catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted)
|
||||
{
|
||||
error = null;
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
error = null;
|
||||
}
|
||||
catch (IOException ex)
|
||||
{
|
||||
error = ex;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
error = new IOException(ex.Message, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_aborted = true;
|
||||
_socket.Shutdown(SocketShutdown.Both);
|
||||
}
|
||||
|
||||
return error;
|
||||
}
|
||||
|
||||
private async Task ProcessSends()
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
// Wait for data to write from the pipe producer
|
||||
var result = await _application.Input.ReadAsync();
|
||||
var buffer = result.Buffer;
|
||||
|
||||
if (result.IsCanceled)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
var end = buffer.End;
|
||||
var isCompleted = result.IsCompleted;
|
||||
if (!buffer.IsEmpty)
|
||||
{
|
||||
await _sender.SendAsync(buffer);
|
||||
}
|
||||
|
||||
_application.Input.AdvanceTo(end);
|
||||
|
||||
if (isCompleted)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
using Microsoft.AspNetCore.Connections;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public static class ConnectionBuilderExtensions
|
||||
{
|
||||
public static IConnectionBuilder UseMqtt(this IConnectionBuilder builder)
|
||||
{
|
||||
return builder.UseConnectionHandler<MqttConnectionHandler>();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
using Microsoft.AspNetCore.Connections;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.AspNetCore.Client.Tcp;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Serializer;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO.Pipelines;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public class MqttConnectionContext : IMqttChannelAdapter
|
||||
{
|
||||
public IMqttPacketSerializer PacketSerializer { get; }
|
||||
public ConnectionContext Connection { get; }
|
||||
|
||||
public string Endpoint => Connection.ConnectionId;
|
||||
|
||||
public MqttConnectionContext(
|
||||
IMqttPacketSerializer packetSerializer,
|
||||
ConnectionContext connection)
|
||||
{
|
||||
PacketSerializer = packetSerializer;
|
||||
Connection = connection;
|
||||
}
|
||||
|
||||
public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
||||
{
|
||||
if (Connection is TcpConnection tcp && !tcp.IsConnected)
|
||||
{
|
||||
return tcp.StartAsync();
|
||||
}
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
||||
{
|
||||
Connection.Transport.Input.Complete();
|
||||
Connection.Transport.Output.Complete();
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
}
|
||||
|
||||
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
||||
{
|
||||
var input = Connection.Transport.Input;
|
||||
|
||||
try
|
||||
{
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
ReadResult readResult;
|
||||
var readTask = input.ReadAsync(cancellationToken);
|
||||
if (readTask.IsCompleted)
|
||||
{
|
||||
readResult = readTask.Result;
|
||||
}
|
||||
else
|
||||
{
|
||||
readResult = await readTask;
|
||||
}
|
||||
|
||||
var buffer = readResult.Buffer;
|
||||
|
||||
var consumed = buffer.Start;
|
||||
var observed = buffer.Start;
|
||||
|
||||
try
|
||||
{
|
||||
if (!buffer.IsEmpty)
|
||||
{
|
||||
if (PacketSerializer.TryDeserialize(buffer, out var packet, out consumed, out observed))
|
||||
{
|
||||
return packet;
|
||||
}
|
||||
else
|
||||
{
|
||||
// we did receive something but the message is not yet complete
|
||||
ReadingPacketStarted?.Invoke(this, EventArgs.Empty);
|
||||
}
|
||||
}
|
||||
else if (readResult.IsCompleted)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
ReadingPacketCompleted?.Invoke(this, EventArgs.Empty);
|
||||
}
|
||||
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
return null;
|
||||
}
|
||||
|
||||
public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
|
||||
{
|
||||
var buffer = PacketSerializer.Serialize(packet);
|
||||
return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask();
|
||||
}
|
||||
|
||||
public event EventHandler ReadingPacketStarted;
|
||||
public event EventHandler ReadingPacketCompleted;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
using Microsoft.AspNetCore.Connections;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Serializer;
|
||||
using MQTTnet.Server;
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter
|
||||
{
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
public override async Task OnConnectedAsync(ConnectionContext connection)
|
||||
{
|
||||
var serializer = new MqttPacketSerializer();
|
||||
using (var adapter = new MqttConnectionContext(serializer, connection))
|
||||
{
|
||||
var args = new MqttServerAdapterClientAcceptedEventArgs(adapter);
|
||||
ClientAccepted?.Invoke(this, args);
|
||||
|
||||
await args.SessionTask;
|
||||
}
|
||||
}
|
||||
|
||||
public Task StartAsync(IMqttServerOptions options)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.IO;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Serializer;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public static class ReaderExtensions
|
||||
{
|
||||
private static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int result)
|
||||
{
|
||||
// Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
|
||||
var multiplier = 1;
|
||||
var value = 0;
|
||||
byte encodedByte;
|
||||
var index = 1;
|
||||
result = 0;
|
||||
|
||||
var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray();
|
||||
|
||||
do
|
||||
{
|
||||
if (index == temp.Length)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
encodedByte = temp[index];
|
||||
index++;
|
||||
|
||||
value += (byte)(encodedByte & 127) * multiplier;
|
||||
if (multiplier > 128 * 128 * 128)
|
||||
{
|
||||
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.AsSpan(1, index).ToArray())}).");
|
||||
}
|
||||
|
||||
multiplier *= 128;
|
||||
} while ((encodedByte & 128) != 0);
|
||||
|
||||
input = input.Slice(index);
|
||||
|
||||
result = value;
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
|
||||
public static byte[] GetArray(this in ReadOnlySequence<byte> input)
|
||||
{
|
||||
if (input.IsSingleSegment)
|
||||
{
|
||||
return input.First.Span.ToArray();
|
||||
}
|
||||
|
||||
// Should be rare
|
||||
return input.ToArray();
|
||||
}
|
||||
|
||||
public static bool TryDeserialize(this IMqttPacketSerializer serializer, in ReadOnlySequence<byte> input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
|
||||
{
|
||||
packet = null;
|
||||
consumed = input.Start;
|
||||
observed = input.End;
|
||||
var copy = input;
|
||||
if (copy.Length < 2)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var fixedheader = copy.First.Span[0];
|
||||
if (!TryReadBodyLength(ref copy, out var bodyLength))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var bodySlice = copy.Slice(0, bodyLength);
|
||||
packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray())));
|
||||
consumed = bodySlice.End;
|
||||
observed = bodySlice.End;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user