From c758ae89f4bdff2a8b6b91412e7d5cbd80b56755 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 27 May 2018 08:24:15 +0200 Subject: [PATCH 01/10] 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(); } From 7a7fab090756d4390caf38c2bc6168edbdbf6d14 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Fri, 1 Jun 2018 21:13:55 +0200 Subject: [PATCH 02/10] fixed rebase --- .../MqttConnectionContext.cs | 8 ++- .../MQTTnet.AspnetCore/ReaderExtensions.cs | 62 +++++++++---------- 2 files changed, 37 insertions(+), 33 deletions(-) diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs index 0bd2b4e3..7683fd01 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -48,7 +48,8 @@ namespace MQTTnet.AspNetCore while (!cancellationToken.IsCancellationRequested) { ReadResult readResult; - + ReadingPacketStarted?.Invoke(this, EventArgs.Empty); + var readTask = input.ReadAsync(cancellationToken); if (readTask.IsCompleted) { @@ -84,6 +85,7 @@ namespace MQTTnet.AspNetCore // 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); + ReadingPacketCompleted?.Invoke(this, EventArgs.Empty); } } @@ -106,6 +108,10 @@ namespace MQTTnet.AspNetCore } private int messageId; + + public event EventHandler ReadingPacketStarted; + public event EventHandler ReadingPacketCompleted; + public Task PublishAsync(MqttPublishPacket packet) { if (!packet.PacketIdentifier.HasValue && packet.QualityOfServiceLevel > MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce) diff --git a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs index a4e5e021..718538b1 100644 --- a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -1,6 +1,7 @@ using System; using System.Buffers; using System.IO; +using MQTTnet.Adapter; using MQTTnet.Exceptions; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -10,37 +11,23 @@ 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) + private static bool TryReadBodyLength(ref ReadOnlySequence 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++; @@ -55,7 +42,8 @@ namespace MQTTnet.AspNetCore input = input.Slice(index); - return value; + result = value; + return true; } @@ -75,17 +63,22 @@ namespace MQTTnet.AspNetCore { packet = null; var copy = input; - var header = copy.ReadHeader(); - if (header == null || copy.Length < header.BodyLength) + if (copy.Length < 2) { return false; } - input = copy.Slice(header.BodyLength); - var bodySlice = copy.Slice(0, header.BodyLength); + var fixedheader = copy.First.Span[0]; + if (!TryReadBodyLength(ref copy, out var bodyLength)) + { + return false; + } + + input = copy.Slice(bodyLength); + var bodySlice = copy.Slice(0, bodyLength); using (var body = new MemoryStream(bodySlice.GetArray())) { - packet = serializer.Deserialize(header, body); + packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body)); return true; } } @@ -93,19 +86,24 @@ namespace MQTTnet.AspNetCore public static bool TryDeserialize(this IMqttPacketSerializer serializer, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) { packet = null; + consumed = input.Start; + observed = input.End; var copy = input; - var header = copy.ReadHeader(); - if (header == null || copy.Length < header.BodyLength) + if (copy.Length < 2) { - consumed = input.Start; - observed = input.End; return false; } - var bodySlice = copy.Slice(0, header.BodyLength); + var fixedheader = copy.First.Span[0]; + if (!TryReadBodyLength(ref copy, out var bodyLength)) + { + return false; + } + + var bodySlice = copy.Slice(0, bodyLength); using (var body = new MemoryStream(bodySlice.GetArray())) { - packet = serializer.Deserialize(header, body); + packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body)); consumed = bodySlice.End; observed = bodySlice.End; return true; From d5ac8c7183ff3af9313fc441afe858b3890012d9 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Fri, 1 Jun 2018 21:41:04 +0200 Subject: [PATCH 03/10] use final packages --- Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj | 8 +++----- .../MQTTnet.TestApp.AspNetCore2.csproj | 4 ++-- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index 661ff168..9754f40c 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -17,11 +17,9 @@ - - - - - + + + diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index c587ecc8..f6ead870 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -1,7 +1,7 @@  - netcoreapp2.0 + netcoreapp2.1 Latest @@ -10,7 +10,7 @@ - + From 7c2adf636fd26d6cbc4e32b331d93c4c285d7826 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 2 Jun 2018 08:35:41 +0200 Subject: [PATCH 04/10] removed dead code and make new connection context an opt in so kestrel 2.0 still works --- .../MqttConnectionContext.cs | 40 ++----------------- .../MQTTnet.AspnetCore/ReaderExtensions.cs | 25 ------------ .../ServiceCollectionExtensions.cs | 12 +++++- Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs | 8 +++- 4 files changed, 19 insertions(+), 66 deletions(-) diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs index 7683fd01..88bfcc91 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -97,46 +97,12 @@ namespace MQTTnet.AspNetCore { foreach (var packet in packets) { - await WriteAsync(packet); + var buffer = PacketSerializer.Serialize(packet); + await Connection.Transport.Output.WriteAsync(buffer.AsMemory()); } } - - public async Task WriteAsync(MqttBasePacket packet) - { - var buffer = PacketSerializer.Serialize(packet); - await Connection.Transport.Output.WriteAsync(buffer.AsMemory()); - } - - private int messageId; - + public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; - - 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/ReaderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs index 718538b1..5b832539 100644 --- a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -4,7 +4,6 @@ using System.IO; using MQTTnet.Adapter; using MQTTnet.Exceptions; using MQTTnet.Packets; -using MQTTnet.Protocol; using MQTTnet.Serializer; namespace MQTTnet.AspNetCore @@ -59,30 +58,6 @@ namespace MQTTnet.AspNetCore return input.ToArray(); } - public static bool TryDeserialize(this IMqttPacketSerializer serializer, ref ReadOnlySequence input, out MqttBasePacket packet) - { - packet = null; - var copy = input; - if (copy.Length < 2) - { - return false; - } - - var fixedheader = copy.First.Span[0]; - if (!TryReadBodyLength(ref copy, out var bodyLength)) - { - return false; - } - - input = copy.Slice(bodyLength); - var bodySlice = copy.Slice(0, bodyLength); - using (var body = new MemoryStream(bodySlice.GetArray())) - { - packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body)); - return true; - } - } - public static bool TryDeserialize(this IMqttPacketSerializer serializer, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) { packet = null; diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 9d2a6762..fa061eb7 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -30,11 +30,19 @@ namespace MQTTnet.AspNetCore if (options.DefaultEndpointOptions.IsEnabled) { - services.AddSingleton(); - services.AddSingleton(s => s.GetService()); + services.AddSingleton(); + services.AddSingleton(s => s.GetService()); } return services; } + + public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services) + { + services.AddSingleton(); + services.AddSingleton(s => s.GetService()); + + return services; + } } } diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index 591cc87d..f52040a0 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -17,8 +17,12 @@ namespace MQTTnet.TestApp.AspNetCore2 public void ConfigureServices(IServiceCollection services) { - var mqttServerOptions = new MqttServerOptionsBuilder().Build(); - services.AddHostedMqttServer(mqttServerOptions); + var mqttServerOptions = new MqttServerOptionsBuilder() + .WithoutDefaultEndpoint() + .Build(); + services + .AddHostedMqttServer(mqttServerOptions) + .AddMqttConnectionHandler(); } // In class _Startup_ of the ASP.NET Core 2.0 project. From 78c605966883cba51fa594e7cb5500beec0b5897 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 2 Jun 2018 09:31:27 +0200 Subject: [PATCH 05/10] enabled clients based on tcp to benefit from new api and added Benchmark for EndtoEnd with new API --- .../MqttClientConnectionContextFactory.cs | 34 +++++++++ .../Client}/Tcp/BufferExtensions.cs | 2 +- .../Client}/Tcp/DuplexPipe.cs | 2 +- .../Client}/Tcp/SocketAwaitable.cs | 2 +- .../Client}/Tcp/SocketReceiver.cs | 2 +- .../Client}/Tcp/SocketSender.cs | 2 +- .../Client}/Tcp/TcpConnection.cs | 45 +++++++---- .../MqttConnectionContext.cs | 6 ++ .../MQTTnet.AspnetCore/ReaderExtensions.cs | 11 +-- Source/MQTTnet/MqttFactory.cs | 8 ++ .../MQTTnet.Benchmarks.csproj | 2 + ...rocessingMqttConnectionContextBenchmark.cs | 76 +++++++++++++++++++ Tests/MQTTnet.Benchmarks/Program.cs | 4 + Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs | 8 +- .../MQTTnet.TestApp.AspNetCore2.csproj | 1 + 15 files changed, 174 insertions(+), 31 deletions(-) create mode 100644 Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/BufferExtensions.cs (93%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/DuplexPipe.cs (96%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketAwaitable.cs (97%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketReceiver.cs (96%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketSender.cs (98%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/TcpConnection.cs (85%) create mode 100644 Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs new file mode 100644 index 00000000..e308ccd4 --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -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(); + } + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs similarity index 93% rename from Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs index 879306c1..5911a3a1 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs @@ -1,7 +1,7 @@ using System; using System.Runtime.InteropServices; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public static class BufferExtensions { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs similarity index 96% rename from Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs index f5f3316d..e234da55 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs @@ -1,6 +1,6 @@ using System.IO.Pipelines; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class DuplexPipe : IDuplexPipe { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs similarity index 97% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs index 2271bd76..96160d13 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs @@ -6,7 +6,7 @@ using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketAwaitable : ICriticalNotifyCompletion { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs similarity index 96% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs index bc8e5c03..219b7222 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs @@ -2,7 +2,7 @@ using System.IO.Pipelines; using System.Net.Sockets; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketReceiver { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs similarity index 98% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs index 7cb1bc1d..c8ba8326 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs @@ -5,7 +5,7 @@ using System.Diagnostics; using System.IO.Pipelines; using System.Net.Sockets; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketSender { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs similarity index 85% rename from Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs index 83cb98a8..40ba2959 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs @@ -4,27 +4,35 @@ 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.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { - public class TcpConnection + public class TcpConnection : ConnectionContext { - private readonly Socket _socket; private volatile bool _aborted; private readonly EndPoint _endPoint; + private SocketSender _sender; + private SocketReceiver _receiver; + + private Socket _socket; private IDuplexPipe _application; - private IDuplexPipe _transport; - private readonly SocketSender _sender; - private readonly SocketReceiver _receiver; + + + public bool IsConnected { get; private set; } + + public override string ConnectionId { get; set; } + + public override IFeatureCollection Features { get; } + + public override IDictionary Items { get; set; } + public override IDuplexPipe Transport { get; set; } public TcpConnection(EndPoint endPoint) { - _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); _endPoint = endPoint; - - _sender = new SocketSender(_socket, PipeScheduler.ThreadPool); - _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool); } public TcpConnection(Socket socket) @@ -38,29 +46,34 @@ namespace MQTTnet.Benchmarks.Tcp public Task DisposeAsync() { - _transport?.Output.Complete(); - _transport?.Input.Complete(); + IsConnected = false; + + Transport?.Output.Complete(); + Transport?.Input.Complete(); _socket?.Dispose(); return Task.CompletedTask; } - public async Task StartAsync() + public async Task StartAsync() { - if (!_socket.Connected) + 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; + Transport = pair.Transport; _application = pair.Application; _ = ExecuteAsync(); - return pair.Transport; + IsConnected = true; } private async Task ExecuteAsync() diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs index 88bfcc91..f7acfa4d 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore.Connections; using MQTTnet.Adapter; +using MQTTnet.AspNetCore.Client.Tcp; using MQTTnet.Packets; using MQTTnet.Serializer; using System; @@ -27,6 +28,10 @@ namespace MQTTnet.AspNetCore public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { + if (Connection is TcpConnection tcp && !tcp.IsConnected) + { + return tcp.StartAsync(); + } return Task.CompletedTask; } @@ -34,6 +39,7 @@ namespace MQTTnet.AspNetCore { Connection.Transport.Input.Complete(); Connection.Transport.Output.Complete(); + return Task.CompletedTask; } diff --git a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs index 5b832539..04117103 100644 --- a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -76,13 +76,10 @@ namespace MQTTnet.AspNetCore } var bodySlice = copy.Slice(0, bodyLength); - using (var body = new MemoryStream(bodySlice.GetArray())) - { - packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body)); - consumed = bodySlice.End; - observed = bodySlice.End; - return true; - } + packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray()))); + consumed = bodySlice.End; + observed = bodySlice.End; + return true; } } } diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index 00438fa9..e830ad20 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -22,6 +22,14 @@ namespace MQTTnet return new MqttClient(new MqttClientAdapterFactory(), logger); } + public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory mqttClientAdapterFactory) + { + if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (mqttClientAdapterFactory == null) throw new ArgumentNullException(nameof(mqttClientAdapterFactory)); + + return new MqttClient(mqttClientAdapterFactory, logger); + } + public IMqttServer CreateMqttServer() { var logger = new MqttNetLogger(); diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 5e612fb9..2a6b91a5 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -10,9 +10,11 @@ + + diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs new file mode 100644 index 00000000..1e2fbb2b --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs @@ -0,0 +1,76 @@ +using BenchmarkDotNet.Attributes; +using MQTTnet.Client; + + +using MQTTnet.AspNetCore; + +using Microsoft.AspNetCore; +using Microsoft.AspNetCore.Hosting; +using MQTTnet.Server; +using MQTTnet.Diagnostics; +using MQTTnet.AspNetCore.Client; + +namespace MQTTnet.Benchmarks +{ + [MemoryDiagnoser] + public class MessageProcessingMqttConnectionContextBenchmark + { + private IWebHost _host; + private IMqttClient _mqttClient; + private MqttApplicationMessage _message; + + [GlobalSetup] + public void Setup() + { + _host = WebHost.CreateDefaultBuilder() + .UseKestrel(o => o.ListenAnyIP(1883, l => l.UseMqtt())) + .ConfigureServices(services => { + var mqttServerOptions = new MqttServerOptionsBuilder() + .WithoutDefaultEndpoint() + .Build(); + services + .AddHostedMqttServer(mqttServerOptions) + .AddMqttConnectionHandler(); + }) + .Configure(app => { + app.UseMqttServer(s => { + + }); + }) + .Build(); + + var factory = new MqttFactory(); + _mqttClient = factory.CreateMqttClient(new MqttNetLogger(), new MqttClientConnectionContextFactory()); + + _host.StartAsync().GetAwaiter().GetResult(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost").Build(); + + _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult(); + + _message = new MqttApplicationMessageBuilder() + .WithTopic("A") + .Build(); + } + + [GlobalCleanup] + public void Cleanup() + { + _mqttClient.DisconnectAsync().GetAwaiter().GetResult(); + _mqttClient.Dispose(); + + _host.StopAsync().GetAwaiter().GetResult(); + _host.Dispose(); + } + + [Benchmark] + public void Send_10000_Messages() + { + for (var i = 0; i < 10000; i++) + { + _mqttClient.PublishAsync(_message).GetAwaiter().GetResult(); + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs index 53069163..9407ccfe 100644 --- a/Tests/MQTTnet.Benchmarks/Program.cs +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -16,6 +16,7 @@ namespace MQTTnet.Benchmarks Console.WriteLine("5 = ChannelAdapterBenchmark"); Console.WriteLine("6 = MqttTcpChannelBenchmark"); Console.WriteLine("7 = TcpPipesBenchmark"); + Console.WriteLine("8 = MessageProcessingMqttConnectionContextBenchmark"); var pressedKey = Console.ReadKey(true); switch (pressedKey.KeyChar) @@ -41,6 +42,9 @@ namespace MQTTnet.Benchmarks case '7': BenchmarkRunner.Run(); break; + case '8': + BenchmarkRunner.Run(new AllowNonOptimized()); + break; } Console.ReadLine(); diff --git a/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs b/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs index cb3ffbaf..4252ab36 100644 --- a/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs @@ -4,7 +4,7 @@ using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; using BenchmarkDotNet.Attributes; -using MQTTnet.Benchmarks.Tcp; +using MQTTnet.AspNetCore.Client.Tcp; namespace MQTTnet.Benchmarks { @@ -25,10 +25,12 @@ namespace MQTTnet.Benchmarks var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883)); - _client = clientConnection.StartAsync().GetAwaiter().GetResult(); + clientConnection.StartAsync().GetAwaiter().GetResult(); + _client = clientConnection.Transport; var serverConnection = new TcpConnection(task.GetAwaiter().GetResult()); - _server = serverConnection.StartAsync().GetAwaiter().GetResult(); + serverConnection.StartAsync().GetAwaiter().GetResult(); + _server = serverConnection.Transport; } diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index f6ead870..a216d728 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -10,6 +10,7 @@ + From c32a606a7064aef784f9042604b0af16d5d3e8b1 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 2 Jun 2018 11:53:58 +0200 Subject: [PATCH 06/10] add memory diagnoser --- Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs index 53b71fd6..f6ff61a3 100644 --- a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -9,6 +9,7 @@ namespace MQTTnet.Benchmarks { [ClrJob] [RPlotExporter, RankColumn] + [MemoryDiagnoser] public class MessageProcessingBenchmark { private IMqttServer _mqttServer; From cf947415e8c9b2e4bcd275642042ce030aaf5089 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 2 Jun 2018 11:54:22 +0200 Subject: [PATCH 07/10] enable abort exceptions --- .../MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs index 40ba2959..4bf74fbb 100644 --- a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs @@ -131,14 +131,14 @@ namespace MQTTnet.AspNetCore.Client.Tcp if (!_aborted) { // Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix. - //error = new MqttCommunicationException(); + error = ConnectionAborted(); } } catch (ObjectDisposedException) { if (!_aborted) { - //error = new MqttCommunicationException(); + error = ConnectionAborted(); } } catch (IOException ex) @@ -153,7 +153,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp { if (_aborted) { - //error = error ?? new MqttCommunicationException(); + error = error ?? ConnectionAborted(); } _application.Output.Complete(error); @@ -193,6 +193,11 @@ namespace MQTTnet.AspNetCore.Client.Tcp } } + private Exception ConnectionAborted() + { + return new MqttCommunicationException("Connection Aborted"); + } + private async Task DoSend() { Exception error = null; From e65c3f2f2b84b47952f15824cfe961f897a21d43 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 16 Jun 2018 17:30:08 +0200 Subject: [PATCH 08/10] moved files into the correct place --- .../MqttConnectionContext.cs | 114 ----------------- .../MqttClientConnectionContextFactory.cs | 0 .../Client/Tcp/BufferExtensions.cs | 0 .../Client/Tcp/DuplexPipe.cs | 0 .../Client/Tcp/SocketAwaitable.cs | 0 .../Client/Tcp/SocketReceiver.cs | 0 .../Client/Tcp/SocketSender.cs | 0 .../Client/Tcp/TcpConnection.cs | 1 + .../ConnectionBuilderExtensions.cs | 0 .../MqttConnectionContext.cs | 120 ++++++++++++++++++ .../MqttConnectionHandler.cs | 0 .../MQTTnet.AspnetCore/ReaderExtensions.cs | 0 12 files changed, 121 insertions(+), 114 deletions(-) delete mode 100644 Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs (99%) rename {Frameworks => Source}/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs (100%) create mode 100644 Source/MQTTnet.AspnetCore/MqttConnectionContext.cs rename {Frameworks => Source}/MQTTnet.AspnetCore/MqttConnectionHandler.cs (100%) rename {Frameworks => Source}/MQTTnet.AspnetCore/ReaderExtensions.cs (100%) diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs deleted file mode 100644 index f7acfa4d..00000000 --- a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ /dev/null @@ -1,114 +0,0 @@ -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 ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) - { - var input = Connection.Transport.Input; - - while (!cancellationToken.IsCancellationRequested) - { - ReadResult readResult; - ReadingPacketStarted?.Invoke(this, EventArgs.Empty); - - 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); - ReadingPacketCompleted?.Invoke(this, EventArgs.Empty); - } - } - - cancellationToken.ThrowIfCancellationRequested(); - return null; - } - - public async Task SendPacketsAsync(TimeSpan timeout, IEnumerable packets, CancellationToken cancellationToken) - { - foreach (var packet in packets) - { - var buffer = PacketSerializer.Serialize(packet); - await Connection.Transport.Output.WriteAsync(buffer.AsMemory()); - } - } - - public event EventHandler ReadingPacketStarted; - public event EventHandler ReadingPacketCompleted; - } -} diff --git a/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs rename to Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs similarity index 99% rename from Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs rename to Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs index 4bf74fbb..7417e744 100644 --- a/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs +++ b/Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.IO; using System.IO.Pipelines; using System.Net; diff --git a/Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs b/Source/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs rename to Source/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs new file mode 100644 index 00000000..3d42937f --- /dev/null +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -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 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; + } +} diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/MqttConnectionHandler.cs rename to Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs diff --git a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs similarity index 100% rename from Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs rename to Source/MQTTnet.AspnetCore/ReaderExtensions.cs From 410bc211a546f7d2e424b45270c0e13c2825ff22 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 16 Jun 2018 17:39:04 +0200 Subject: [PATCH 09/10] fixed packet reader --- Source/MQTTnet/Serializer/MqttPacketReader.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Serializer/MqttPacketReader.cs b/Source/MQTTnet/Serializer/MqttPacketReader.cs index d50fae5d..79329561 100644 --- a/Source/MQTTnet/Serializer/MqttPacketReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketReader.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Serializer while (totalBytesRead < buffer.Length) { - var bytesRead = await channel.ReadAsync(buffer, 0, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); + var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); if (bytesRead <= 0) { if (cancellationToken.IsCancellationRequested) From d573423372fa57ba2830c73ea866efdaad7d10fc Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 16 Jun 2018 17:39:37 +0200 Subject: [PATCH 10/10] downgrade dependencies to 2.0 --- Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj | 4 ++-- Tests/MQTTnet.TestApp.AspNetCore2/Program.cs | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index 9754f40c..8db250bf 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -18,8 +18,8 @@ - - + + diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs index a7164117..5248ce61 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Program.cs @@ -13,7 +13,10 @@ namespace MQTTnet.TestApp.AspNetCore2 private static IWebHost BuildWebHost(string[] args) => WebHost.CreateDefaultBuilder(args) - .UseKestrel(o => o.ListenAnyIP(1883, l => l.UseMqtt())) + .UseKestrel(o => { + o.ListenAnyIP(1883, l => l.UseMqtt()); + o.ListenAnyIP(5000); // default http pipeline + }) .UseStartup() .Build(); }