From c758ae89f4bdff2a8b6b91412e7d5cbd80b56755 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 27 May 2018 08:24:15 +0200 Subject: [PATCH] first iteration --- .../ConnectionBuilderExtensions.cs | 12 ++ .../MqttConnectionContext.cs | 136 ++++++++++++++++++ .../MqttConnectionHandler.cs | 40 ++++++ .../MQTTnet.AspnetCore/ReaderExtensions.cs | 115 +++++++++++++++ .../MQTTnet.AspNetCore.csproj | 9 +- .../ServiceCollectionExtensions.cs | 3 +- Tests/MQTTnet.TestApp.AspNetCore2/Program.cs | 2 + 7 files changed, 313 insertions(+), 4 deletions(-) create mode 100644 Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs create mode 100644 Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs create mode 100644 Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs create mode 100644 Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs diff --git a/Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs new file mode 100644 index 00000000..262a333f --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs @@ -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(); + } + } +} diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs new file mode 100644 index 00000000..0bd2b4e3 --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -0,0 +1,136 @@ +using Microsoft.AspNetCore.Connections; +using MQTTnet.Adapter; +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) + { + 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 ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + var input = Connection.Transport.Input; + + 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 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); + } + } + + cancellationToken.ThrowIfCancellationRequested(); + return null; + } + + public async Task SendPacketsAsync(TimeSpan timeout, IEnumerable packets, CancellationToken cancellationToken) + { + foreach (var packet in packets) + { + await WriteAsync(packet); + } + } + + public async Task WriteAsync(MqttBasePacket packet) + { + var buffer = PacketSerializer.Serialize(packet); + await Connection.Transport.Output.WriteAsync(buffer.AsMemory()); + } + + private int messageId; + public Task PublishAsync(MqttPublishPacket packet) + { + if (!packet.PacketIdentifier.HasValue && packet.QualityOfServiceLevel > MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce) + { + packet.PacketIdentifier = (ushort)Interlocked.Increment(ref messageId); + } + return WriteAsync(packet); + } + + public Task SubscribeAsync(MqttSubscribePacket packet) + { + if (!packet.PacketIdentifier.HasValue) + { + packet.PacketIdentifier = (ushort)Interlocked.Increment(ref messageId); + } + return WriteAsync(packet); + } + + public Task ConnectAsync(MqttConnectPacket packet) + { + if (string.IsNullOrEmpty(packet.ClientId)) + { + packet.ClientId = Guid.NewGuid().ToString(); + } + return WriteAsync(packet); + } + } +} diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs new file mode 100644 index 00000000..49a5c098 --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs @@ -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 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() + { + } + } +} diff --git a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs new file mode 100644 index 00000000..a4e5e021 --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -0,0 +1,115 @@ +using System; +using System.Buffers; +using System.IO; +using MQTTnet.Exceptions; +using MQTTnet.Packets; +using MQTTnet.Protocol; +using MQTTnet.Serializer; + +namespace MQTTnet.AspNetCore +{ + public static class ReaderExtensions + { + public static MqttPacketHeader ReadHeader(this ref ReadOnlySequence input) + { + if (input.Length < 2) + { + return null; + } + + var fixedHeader = input.First.Span[0]; + var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4); + var bodyLength = ReadBodyLength(ref input); + + return new MqttPacketHeader + { + FixedHeader = fixedHeader, + ControlPacketType = controlPacketType, + BodyLength = bodyLength + }; + } + + private static int ReadBodyLength(ref ReadOnlySequence input) + { + // 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; + + var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray(); + + do + { + 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); + + return value; + } + + + + public static byte[] GetArray(this in ReadOnlySequence input) + { + if (input.IsSingleSegment) + { + return input.First.Span.ToArray(); + } + + // Should be rare + return input.ToArray(); + } + + public static bool TryDeserialize(this IMqttPacketSerializer serializer, ref ReadOnlySequence input, out MqttBasePacket packet) + { + packet = null; + var copy = input; + var header = copy.ReadHeader(); + if (header == null || copy.Length < header.BodyLength) + { + return false; + } + + input = copy.Slice(header.BodyLength); + var bodySlice = copy.Slice(0, header.BodyLength); + using (var body = new MemoryStream(bodySlice.GetArray())) + { + packet = serializer.Deserialize(header, body); + return true; + } + } + + public static bool TryDeserialize(this IMqttPacketSerializer serializer, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) + { + packet = null; + var copy = input; + var header = copy.ReadHeader(); + if (header == null || copy.Length < header.BodyLength) + { + consumed = input.Start; + observed = input.End; + return false; + } + + var bodySlice = copy.Slice(0, header.BodyLength); + using (var body = new MemoryStream(bodySlice.GetArray())) + { + packet = serializer.Deserialize(header, body); + consumed = bodySlice.End; + observed = bodySlice.End; + return true; + } + } + } +} diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index 6b858954..661ff168 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -9,6 +9,7 @@ + 7.2 @@ -16,9 +17,11 @@ - - - + + + + + diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 5b8f1612..9d2a6762 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -30,7 +30,8 @@ namespace MQTTnet.AspNetCore if (options.DefaultEndpointOptions.IsEnabled) { - services.AddSingleton(s => s.GetService()); + services.AddSingleton(); + services.AddSingleton(s => s.GetService()); } return services; diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs index 38b8c12d..a7164117 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore; using Microsoft.AspNetCore.Hosting; +using MQTTnet.AspNetCore; namespace MQTTnet.TestApp.AspNetCore2 { @@ -12,6 +13,7 @@ namespace MQTTnet.TestApp.AspNetCore2 private static IWebHost BuildWebHost(string[] args) => WebHost.CreateDefaultBuilder(args) + .UseKestrel(o => o.ListenAnyIP(1883, l => l.UseMqtt())) .UseStartup() .Build(); }