From fbca8a6dc12aa87730a67cf012aa86f55f7c1b88 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Fri, 3 Aug 2018 22:09:51 +0200 Subject: [PATCH 01/12] span based reader --- .../MQTTnet.AspNetCore.csproj | 1 + .../MqttConnectionContext.cs | 1 + Source/MQTTnet.AspnetCore/ReaderExtensions.cs | 18 +-- .../SpanBasedMqttPacketBodyReader.cs | 104 ++++++++++++++++++ Source/MQTTnet/Adapter/ReceivedMqttPacket.cs | 6 +- .../Formatter/IMqttPacketBodyReader.cs | 31 ++++++ .../MQTTnet/Formatter/MqttPacketBodyReader.cs | 16 ++- .../Formatter/V3/MqttV310PacketFormatter.cs | 25 ++--- .../Formatter/V3/MqttV311PacketFormatter.cs | 2 +- .../Formatter/V5/MqttV500PacketDecoder.cs | 30 ++--- .../Formatter/V5/MqttV500PropertiesReader.cs | 8 +- 11 files changed, 194 insertions(+), 48 deletions(-) create mode 100644 Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs create mode 100644 Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index dc477f76..32943b5f 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -7,6 +7,7 @@ 7.2 + true diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index a0ad1fe0..d798be81 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -54,6 +54,7 @@ namespace MQTTnet.AspNetCore public async Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) { var input = Connection.Transport.Input; + var reader = new SpanBasedMqttPacketBodyReader(); try { diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index 81d24c0e..3455d028 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -35,18 +35,22 @@ namespace MQTTnet.AspNetCore } var bodySlice = copy.Slice(0, bodyLength); - var buffer = bodySlice.GetArray(); - packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length), buffer.Length + 2)); + var buffer = bodySlice.GetMemory(); + + var reader = new SpanBasedMqttPacketBodyReader(); + reader.SetBuffer(buffer); + + packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2)); consumed = bodySlice.End; observed = bodySlice.End; return true; } - private static byte[] GetArray(this in ReadOnlySequence input) + private static ReadOnlyMemory GetMemory(this in ReadOnlySequence input) { if (input.IsSingleSegment) { - return input.First.Span.ToArray(); + return input.First; } // Should be rare @@ -62,7 +66,7 @@ namespace MQTTnet.AspNetCore var index = 1; result = 0; - var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray(); + var temp = input.Slice(0, Math.Min(5, input.Length)).GetMemory(); do { @@ -70,13 +74,13 @@ namespace MQTTnet.AspNetCore { return false; } - encodedByte = temp[index]; + encodedByte = temp.Span[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())})."); + throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())})."); } multiplier *= 128; diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs new file mode 100644 index 00000000..01ccb3a8 --- /dev/null +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs @@ -0,0 +1,104 @@ +using MQTTnet.Formatter; +using System; +using System.Buffers.Binary; +using System.Text; + +namespace MQTTnet.AspNetCore +{ + public class SpanBasedMqttPacketBodyReader : IMqttPacketBodyReader + { + private ReadOnlyMemory _buffer; + private int _offset; + + public void SetBuffer(ReadOnlyMemory buffer) + { + _buffer = buffer; + _offset = 0; + } + + public ulong Length => (ulong)_buffer.Length; + + public bool EndOfStream => _buffer.Length.Equals(_offset); + + public ulong Offset => (ulong)_offset; + + public byte ReadByte() + { + ValidateReceiveBuffer(1); + return _buffer.Span[_offset++]; + } + + public byte[] ReadRemainingData() + { + return _buffer.ToArray(); + } + + public ushort ReadUInt16() + { + ValidateReceiveBuffer(2); + + var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span); + _offset += 2; + return result; + } + + public byte[] ReadWithLengthPrefix() + { + return ReadSegmentWithLengthPrefix().ToArray(); + } + + private ReadOnlyMemory ReadSegmentWithLengthPrefix() + { + var length = ReadUInt16(); + + ValidateReceiveBuffer(length); + + var result = _buffer.Slice(_offset, length); + _offset += length; + return result; + } + + private void ValidateReceiveBuffer(ushort length) + { + if (_buffer.Length < _offset + length) + { + throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_buffer.Length} bytes"); + } + } + + public unsafe string ReadStringWithLengthPrefix() + { + var buffer = ReadSegmentWithLengthPrefix(); + fixed (byte* bytes = &buffer.Span.GetPinnableReference()) + { + var result = Encoding.UTF8.GetString(bytes, buffer.Length); + return result; + } + } + + public ushort ReadTwoByteInteger() + { + throw new NotImplementedException(); + } + + public uint ReadFourByteInteger() + { + throw new NotImplementedException(); + } + + public uint ReadVariableLengthInteger() + { + throw new NotImplementedException(); + } + + public bool ReadBoolean() + { + throw new NotImplementedException(); + } + + public void Seek(ulong position) + { + throw new NotImplementedException(); + } + } +} diff --git a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs index 683d6b64..355ad96f 100644 --- a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs +++ b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs @@ -4,16 +4,16 @@ namespace MQTTnet.Adapter { public class ReceivedMqttPacket { - public ReceivedMqttPacket(byte fixedHeader, MqttPacketBodyReader body, int totalLength) + public ReceivedMqttPacket(byte fixedHeader, IMqttPacketBodyReader body, int totalLength) { FixedHeader = fixedHeader; Body = body; TotalLength = totalLength; } - public byte FixedHeader { get; } + public byte FixedHeader { get; set; } - public MqttPacketBodyReader Body { get; } + public IMqttPacketBodyReader Body { get; } public int TotalLength { get; } } diff --git a/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs new file mode 100644 index 00000000..632c4931 --- /dev/null +++ b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs @@ -0,0 +1,31 @@ +using System; + +namespace MQTTnet.Formatter +{ + public interface IMqttPacketBodyReader + { + ulong Length { get; } + + ulong Offset { get; } + + bool EndOfStream { get; } + + byte ReadByte(); + + byte[] ReadRemainingData(); + + ushort ReadTwoByteInteger(); + + string ReadStringWithLengthPrefix(); + + byte[] ReadWithLengthPrefix(); + + uint ReadFourByteInteger(); + + uint ReadVariableLengthInteger(); + + bool ReadBoolean(); + + void Seek(ulong position); + } +} diff --git a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs index db971f09..c45fb3d9 100644 --- a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs +++ b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs @@ -1,11 +1,12 @@ using System; +using System.Linq; using System.Runtime.CompilerServices; using System.Text; using MQTTnet.Exceptions; namespace MQTTnet.Formatter { - public class MqttPacketBodyReader + public class MqttPacketBodyReader : IMqttPacketBodyReader { private readonly byte[] _buffer; private readonly ulong _initialOffset; @@ -68,9 +69,9 @@ namespace MQTTnet.Formatter throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only."); } - public ArraySegment ReadRemainingData() + public byte[] ReadRemainingData() { - return new ArraySegment(_buffer, (int)Offset, (int)(_length - Offset)); + return new ArraySegment(_buffer, (int)Offset, (int)(_length - Offset)).ToArray(); } public ushort ReadTwoByteInteger() @@ -95,7 +96,12 @@ namespace MQTTnet.Formatter return (uint)(byte0 << 24 | byte1 << 16 | byte2 << 8 | byte3); } - public ArraySegment ReadWithLengthPrefix() + public byte[] ReadWithLengthPrefix() + { + return ReadSegmentWithLengthPrefix().ToArray(); + } + + private ArraySegment ReadSegmentWithLengthPrefix() { var length = ReadTwoByteInteger(); @@ -140,7 +146,7 @@ namespace MQTTnet.Formatter public string ReadStringWithLengthPrefix() { - var buffer = ReadWithLengthPrefix(); + var buffer = ReadSegmentWithLengthPrefix(); return Encoding.UTF8.GetString(buffer.Array, buffer.Offset, buffer.Count); } } diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs index 02a28a12..d3e6b416 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs @@ -2,7 +2,6 @@ using System.Linq; using MQTTnet.Adapter; using MQTTnet.Exceptions; -using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -100,7 +99,7 @@ namespace MQTTnet.Formatter.V3 } } - private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -110,7 +109,7 @@ namespace MQTTnet.Formatter.V3 }; } - private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -120,7 +119,7 @@ namespace MQTTnet.Formatter.V3 }; } - private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -130,7 +129,7 @@ namespace MQTTnet.Formatter.V3 }; } - private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -140,7 +139,7 @@ namespace MQTTnet.Formatter.V3 }; } - private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -150,7 +149,7 @@ namespace MQTTnet.Formatter.V3 }; } - private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -167,7 +166,7 @@ namespace MQTTnet.Formatter.V3 return packet; } - private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -223,7 +222,7 @@ namespace MQTTnet.Formatter.V3 return packet; } - private MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body) + private MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -264,7 +263,7 @@ namespace MQTTnet.Formatter.V3 packet.WillMessage = new MqttApplicationMessage { Topic = body.ReadStringWithLengthPrefix(), - Payload = body.ReadWithLengthPrefix().ToArray(), + Payload = body.ReadWithLengthPrefix(), QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS, Retain = willRetain }; @@ -284,7 +283,7 @@ namespace MQTTnet.Formatter.V3 return packet; } - private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -301,7 +300,7 @@ namespace MQTTnet.Formatter.V3 return packet; } - protected virtual MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body) + protected virtual MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -577,7 +576,7 @@ namespace MQTTnet.Formatter.V3 } // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local - protected static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body) + protected static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body) { if (body == null || body.Length == 0) { diff --git a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs index 0bc60a99..44725d4e 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs @@ -82,7 +82,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } - protected override MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body) + protected override MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index 54588eea..46692cca 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -41,7 +41,7 @@ namespace MQTTnet.Formatter.V5 } } - private static MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -134,7 +134,7 @@ namespace MQTTnet.Formatter.V5 if (packet.WillMessage != null) { packet.WillMessage.Topic = body.ReadStringWithLengthPrefix(); - packet.WillMessage.Payload = body.ReadWithLengthPrefix().ToArray(); + packet.WillMessage.Payload = body.ReadWithLengthPrefix(); } if (usernameFlag) @@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -239,7 +239,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeDisconnectPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeDisconnectPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -277,7 +277,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -327,7 +327,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -363,7 +363,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -394,7 +394,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -440,7 +440,7 @@ namespace MQTTnet.Formatter.V5 return new MqttPingRespPacket(); } - private static MqttBasePacket DecodePublishPacket(byte header, MqttPacketBodyReader body) + private static MqttBasePacket DecodePublishPacket(byte header, IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -511,7 +511,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -549,7 +549,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -587,7 +587,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -625,7 +625,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -663,7 +663,7 @@ namespace MQTTnet.Formatter.V5 return packet; } - private static MqttBasePacket DecodeAuthPacket(MqttPacketBodyReader body) + private static MqttBasePacket DecodeAuthPacket(IMqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -709,7 +709,7 @@ namespace MQTTnet.Formatter.V5 } // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local - private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body) + private static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body) { if (body == null || body.Length == 0) { diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index c7f28dd4..e30cb3f2 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -9,11 +9,11 @@ namespace MQTTnet.Formatter.V5 { public class MqttV500PropertiesReader { - private readonly MqttPacketBodyReader _body; + private readonly IMqttPacketBodyReader _body; private readonly uint _length; private readonly ulong _targetOffset; - public MqttV500PropertiesReader(MqttPacketBodyReader body) + public MqttV500PropertiesReader(IMqttPacketBodyReader body) { _body = body ?? throw new ArgumentNullException(nameof(body)); @@ -75,7 +75,7 @@ namespace MQTTnet.Formatter.V5 public byte[] ReadAuthenticationData() { - return _body.ReadWithLengthPrefix().ToArray(); + return _body.ReadWithLengthPrefix(); } public bool? ReadRetainAvailable() @@ -165,7 +165,7 @@ namespace MQTTnet.Formatter.V5 public byte[] ReadCorrelationData() { - return _body.ReadWithLengthPrefix().ToArray(); + return _body.ReadWithLengthPrefix(); } public string ReadContentType() From a85792c392bd4e91d609ca7fab44988bf5b53d45 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 22 Dec 2018 17:16:09 +0100 Subject: [PATCH 02/12] fixed another issue --- Source/MQTTnet.AspnetCore/ReaderExtensions.cs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index 3455d028..cb347c4c 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -40,7 +40,14 @@ namespace MQTTnet.AspNetCore var reader = new SpanBasedMqttPacketBodyReader(); reader.SetBuffer(buffer); - packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2)); + var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2); + + if (!formatter.ProtocolVersion.HasValue) + { + formatter.DetectProtocolVersion(receivedMqttPacket); + } + + packet = formatter.Decode(receivedMqttPacket); consumed = bodySlice.End; observed = bodySlice.End; return true; From f8ade2daf47738af175756b4a45c360ae64e3c0c Mon Sep 17 00:00:00 2001 From: JanEggers Date: Fri, 3 Aug 2018 23:13:05 +0200 Subject: [PATCH 03/12] slight improvements --- Source/MQTTnet.AspnetCore/MqttConnectionContext.cs | 14 ++++++++------ .../SpanBasedMqttPacketBodyReader.cs | 8 ++++---- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index d798be81..d3e583f5 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -18,6 +18,8 @@ namespace MQTTnet.AspNetCore PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); Connection = connection ?? throw new ArgumentNullException(nameof(connection)); } + private PipeReader _input; + private PipeWriter _output; public string Endpoint => Connection.ConnectionId; public bool IsSecureConnection => false; // TODO: Fix detection (WS vs. WSS). @@ -33,20 +35,21 @@ namespace MQTTnet.AspNetCore private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1); - public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) + public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { if (Connection is TcpConnection tcp && !tcp.IsConnected) { - return tcp.StartAsync(); + await tcp.StartAsync().ConfigureAwait(false); } - return Task.CompletedTask; + _input = Connection.Transport.Input; + _output = Connection.Transport.Output; } public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { - Connection.Transport.Input.Complete(); - Connection.Transport.Output.Complete(); + _input?.Complete(); + _output?.Complete(); return Task.CompletedTask; } @@ -54,7 +57,6 @@ namespace MQTTnet.AspNetCore public async Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) { var input = Connection.Transport.Input; - var reader = new SpanBasedMqttPacketBodyReader(); try { diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs index 01ccb3a8..9e16dacf 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs @@ -30,7 +30,7 @@ namespace MQTTnet.AspNetCore public byte[] ReadRemainingData() { - return _buffer.ToArray(); + return _buffer.Slice(_offset).ToArray(); } public ushort ReadUInt16() @@ -47,13 +47,13 @@ namespace MQTTnet.AspNetCore return ReadSegmentWithLengthPrefix().ToArray(); } - private ReadOnlyMemory ReadSegmentWithLengthPrefix() + private ReadOnlySpan ReadSegmentWithLengthPrefix() { var length = ReadUInt16(); ValidateReceiveBuffer(length); - var result = _buffer.Slice(_offset, length); + var result = _buffer.Slice(_offset, length).Span; _offset += length; return result; } @@ -69,7 +69,7 @@ namespace MQTTnet.AspNetCore public unsafe string ReadStringWithLengthPrefix() { var buffer = ReadSegmentWithLengthPrefix(); - fixed (byte* bytes = &buffer.Span.GetPinnableReference()) + fixed (byte* bytes = &buffer.GetPinnableReference()) { var result = Encoding.UTF8.GetString(bytes, buffer.Length); return result; From 888e1797d72451b20ce03e8841438d25f2ae0d24 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 14:04:21 +0200 Subject: [PATCH 04/12] fixed cherry pick --- Source/MQTTnet.AspnetCore/ReaderExtensions.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index cb347c4c..e2cb1c44 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -42,7 +42,7 @@ namespace MQTTnet.AspNetCore var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2); - if (!formatter.ProtocolVersion.HasValue) + if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown) { formatter.DetectProtocolVersion(receivedMqttPacket); } From 149d01e173f4cba8d29a93811f01030f34f636d5 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 4 Aug 2018 16:02:46 +0200 Subject: [PATCH 05/12] added tests --- .../SpanBasedMqttPacketBodyReader.cs | 24 +++++-------------- .../MqttPacketSerializerTests.cs | 17 +++++++++++++ .../MqttPacketSerializer_Tests.cs | 7 +++++- 3 files changed, 29 insertions(+), 19 deletions(-) create mode 100644 Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs index 9e16dacf..63e0ab81 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs @@ -24,7 +24,6 @@ namespace MQTTnet.AspNetCore public byte ReadByte() { - ValidateReceiveBuffer(1); return _buffer.Span[_offset++]; } @@ -35,9 +34,7 @@ namespace MQTTnet.AspNetCore public ushort ReadUInt16() { - ValidateReceiveBuffer(2); - - var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span); + var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset)); _offset += 2; return result; } @@ -49,23 +46,14 @@ namespace MQTTnet.AspNetCore private ReadOnlySpan ReadSegmentWithLengthPrefix() { - var length = ReadUInt16(); + var span = _buffer.Span; + var length = BinaryPrimitives.ReadUInt16BigEndian(span.Slice(_offset)); - ValidateReceiveBuffer(length); - - var result = _buffer.Slice(_offset, length).Span; - _offset += length; + var result = span.Slice(_offset+2, length); + _offset += 2 + length; return result; } - - private void ValidateReceiveBuffer(ushort length) - { - if (_buffer.Length < _offset + length) - { - throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_buffer.Length} bytes"); - } - } - + public unsafe string ReadStringWithLengthPrefix() { var buffer = ReadSegmentWithLengthPrefix(); diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs new file mode 100644 index 00000000..f0920d54 --- /dev/null +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs @@ -0,0 +1,17 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Core.Tests; +using MQTTnet.Serializer; + +namespace MQTTnet.AspNetCore.Tests +{ + [TestClass] + public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializerTests + { + protected override IPacketBodyReader ReaderFactory(byte[] data) + { + var result = new SpanBasedMqttPacketBodyReader(); + result.SetBuffer(data); + return result; + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index a1aad718..cb1b0695 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -528,7 +528,12 @@ namespace MQTTnet.Tests Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data))); } - private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) + protected virtual IMqttPacketBodyReader ReaderFactory(byte[] data) + { + return new MqttPacketBodyReader(data, 0, data.Length); + } + + private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) { IMqttPacketFormatter serializer; if (protocolVersion == MqttProtocolVersion.V311) From cfc1b27950709ccb4ed84b4a86a89bb8291bb014 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 22 Dec 2018 17:00:57 +0100 Subject: [PATCH 06/12] fixed build and added implementations for new read functions --- .../MqttConnectionContext.cs | 11 +++- .../SpanBasedMqttPacketBodyReader.cs | 53 ++++++++++++++----- .../MQTTnet.AspNetCore.Tests.csproj | 2 +- .../MqttPacketSerializerTests.cs | 8 +-- 4 files changed, 54 insertions(+), 20 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index d3e583f5..df67eeb3 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -17,6 +17,12 @@ namespace MQTTnet.AspNetCore { PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); Connection = connection ?? throw new ArgumentNullException(nameof(connection)); + + if (Connection.Transport != null) + { + _input = Connection.Transport.Input; + _output = Connection.Transport.Output; + } } private PipeReader _input; private PipeWriter _output; @@ -117,8 +123,9 @@ namespace MQTTnet.AspNetCore public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken) { - var buffer = PacketFormatterAdapter.Encode(packet).AsMemory(); - var output = Connection.Transport.Output; + var buffer = PacketFormatterAdapter.Encode(packet); + var msg = buffer.AsMemory(); + var output = _output; await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs index 63e0ab81..dbfbb0fe 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs @@ -1,4 +1,5 @@ -using MQTTnet.Formatter; +using MQTTnet.Exceptions; +using MQTTnet.Formatter; using System; using System.Buffers.Binary; using System.Text; @@ -32,13 +33,6 @@ namespace MQTTnet.AspNetCore return _buffer.Slice(_offset).ToArray(); } - public ushort ReadUInt16() - { - var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset)); - _offset += 2; - return result; - } - public byte[] ReadWithLengthPrefix() { return ReadSegmentWithLengthPrefix().ToArray(); @@ -66,27 +60,60 @@ namespace MQTTnet.AspNetCore public ushort ReadTwoByteInteger() { - throw new NotImplementedException(); + var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset)); + _offset += 2; + return result; } public uint ReadFourByteInteger() { - throw new NotImplementedException(); + var result = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_offset)); + _offset += 4; + return result; } public uint ReadVariableLengthInteger() { - throw new NotImplementedException(); + var multiplier = 1; + var value = 0U; + byte encodedByte; + + do + { + encodedByte = ReadByte(); + value += (uint)((encodedByte & 127) * multiplier); + + if (multiplier > 2097152) + { + throw new MqttProtocolViolationException("Variable length integer is invalid."); + } + + multiplier *= 128; + } while ((encodedByte & 128) != 0); + + return value; } public bool ReadBoolean() { - throw new NotImplementedException(); + var buffer = ReadByte(); + + if (buffer == 0) + { + return false; + } + + if (buffer == 1) + { + return true; + } + + throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only."); } public void Seek(ulong position) { - throw new NotImplementedException(); + _offset = (int)position; } } } diff --git a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj index e523de5d..06f0e5f1 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj +++ b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj @@ -14,7 +14,7 @@ - + diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs index f0920d54..65625d34 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs @@ -1,13 +1,13 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Core.Tests; -using MQTTnet.Serializer; +using MQTTnet.Formatter; +using MQTTnet.Tests; namespace MQTTnet.AspNetCore.Tests { [TestClass] - public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializerTests + public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializer_Tests { - protected override IPacketBodyReader ReaderFactory(byte[] data) + protected override IMqttPacketBodyReader ReaderFactory(byte[] data) { var result = new SpanBasedMqttPacketBodyReader(); result.SetBuffer(data); From aee10ba6fa6be3ad0c48c5a9ef149704f36ae891 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 14:24:27 +0200 Subject: [PATCH 07/12] added netcore 2.1 build --- Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs | 4 ++++ Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj | 2 +- Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj | 2 +- 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs index 55192d6c..f5bb0b4c 100644 --- a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs +++ b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs @@ -5,6 +5,10 @@ using System.Diagnostics; using System.IO.Pipelines; using System.Net.Sockets; +#if NETCOREAPP2_1 +using System.Runtime.InteropServices; +#endif + namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketSender diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj index 32943b5f..a1abcc83 100644 --- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj +++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj @@ -1,7 +1,7 @@  - netstandard2.0 + netstandard2.0;netcoreapp2.1 diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 02818bf4..03aabc85 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -3,7 +3,7 @@ Exe Full - net461 + net461;netcoreapp2.1 netcoreapp2.1 7.2 From a9944def6896150dd50c345cbf9e466c2b882622 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 12:16:06 +0200 Subject: [PATCH 08/12] added span based writer --- .../MqttClientConnectionContextFactory.cs | 5 +- .../MqttConnectionContext.cs | 13 +- .../MqttConnectionHandler.cs | 4 +- .../MqttWebSocketServerAdapter.cs | 4 +- .../SpanBasedMqttPacketWriter.cs | 147 ++++++++++++++++++ Source/MQTTnet/Formatter/IMqttPacketWriter.cs | 20 +++ .../Formatter/MqttPacketFormatterAdapter.cs | 22 ++- Source/MQTTnet/Formatter/MqttPacketWriter.cs | 28 +++- .../Formatter/V3/MqttV310PacketFormatter.cs | 45 ++++-- .../Formatter/V3/MqttV311PacketFormatter.cs | 14 +- .../Formatter/V5/MqttV500PacketEncoder.cs | 50 +++--- .../Formatter/V5/MqttV500PacketFormatter.cs | 12 +- .../Formatter/V5/MqttV500PropertiesWriter.cs | 2 +- .../MqttPacketSerializerTests.cs | 5 + .../MqttPacketSerializer_Tests.cs | 25 ++- 15 files changed, 334 insertions(+), 62 deletions(-) create mode 100644 Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs create mode 100644 Source/MQTTnet/Formatter/IMqttPacketWriter.cs diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs index a83fcefd..e8f570c2 100644 --- a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs +++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -21,7 +21,10 @@ namespace MQTTnet.AspNetCore.Client { var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort()); var tcpConnection = new TcpConnection(endpoint); - return new MqttConnectionContext(new MqttPacketFormatterAdapter(options.ProtocolVersion), tcpConnection); + + var writer = new SpanBasedMqttPacketWriter(); + var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, writer); + return new MqttConnectionContext(formatter, tcpConnection); } default: { diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index df67eeb3..400aa070 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -123,14 +123,19 @@ namespace MQTTnet.AspNetCore public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken) { - var buffer = PacketFormatterAdapter.Encode(packet); - var msg = buffer.AsMemory(); - var output = _output; + var formatter = PacketFormatterAdapter; + await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try { - await output.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); + var buffer = formatter.Encode(packet); + var msg = buffer.AsMemory(); + var output = _output; + msg.CopyTo(output.GetMemory(msg.Length)); + PacketFormatterAdapter.FreeBuffer(); + output.Advance(msg.Length); + await output.FlushAsync().ConfigureAwait(false); } finally { diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs index ec2528f0..f3cb91a1 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs @@ -21,7 +21,9 @@ namespace MQTTnet.AspNetCore transferFormatFeature.ActiveFormat = TransferFormat.Binary; } - using (var adapter = new MqttConnectionContext(new MqttPacketFormatterAdapter(), connection)) + var writer = new SpanBasedMqttPacketWriter(); + var formatter = new MqttPacketFormatterAdapter(writer); + using (var adapter = new MqttConnectionContext(formatter, connection)) { var args = new MqttServerAdapterClientAcceptedEventArgs(adapter); ClientAcceptedHandler?.Invoke(args); diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index 3e5e6075..1cd512ba 100644 --- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -43,8 +43,10 @@ namespace MQTTnet.AspNetCore var isSecureConnection = clientCertificate != null; clientCertificate?.Dispose(); + var writer = new SpanBasedMqttPacketWriter(); + var formatter = new MqttPacketFormatterAdapter(writer); var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection); - var channelAdapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(), _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter))); + var channelAdapter = new MqttChannelAdapter(channel, formatter, new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter); ClientAcceptedHandler?.Invoke(eventArgs); diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs new file mode 100644 index 00000000..7ad6e005 --- /dev/null +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs @@ -0,0 +1,147 @@ +using MQTTnet.Formatter; +using System; +using System.Buffers; +using System.Buffers.Binary; +using System.Text; + +namespace MQTTnet.AspNetCore +{ + public class SpanBasedMqttPacketWriter : IMqttPacketWriter + { + private readonly ArrayPool _pool; + + public SpanBasedMqttPacketWriter() + { + _pool = ArrayPool.Create(); + + } + + private byte[] _buffer; + private int _position; + + public int Length { get; set; } + + public void FreeBuffer() + { + _pool.Return(_buffer); + } + + public byte[] GetBuffer() + { + return _buffer; + } + + public void Reset(int v) + { + _buffer = _pool.Rent(1500); + Length = v; + _position = v; + } + + public void Seek(int v) + { + _position = v; + } + + public void Write(byte value) + { + GrowIfNeeded(1); + _buffer[_position] = value; + Commit(1); + } + + public void Write(ushort value) + { + GrowIfNeeded(2); + + BinaryPrimitives.WriteUInt16BigEndian(_buffer.AsSpan(_position), value); + Commit(2); + } + + public void Write(IMqttPacketWriter propertyWriter) + { + if (propertyWriter is SpanBasedMqttPacketWriter writer) + { + GrowIfNeeded(1); + } + + throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(SpanBasedMqttPacketWriter).Name}"); + } + + public void Write(byte[] payload, int start, int length) + { + GrowIfNeeded(length); + + payload.AsSpan(start, length).CopyTo(_buffer.AsSpan(_position)); + Commit(length); + } + + public void WriteVariableLengthInteger(uint value) + { + GrowIfNeeded(4); + + var x = value; + do + { + var encodedByte = x % 128; + x = x / 128; + if (x > 0) + { + encodedByte = encodedByte | 128; + } + + _buffer[_position] = (byte)encodedByte; + Commit(1); + } while (x > 0); + } + + public void WriteWithLengthPrefix(string value) + { + var bytesLength = Encoding.UTF8.GetByteCount(value ?? string.Empty); + GrowIfNeeded(bytesLength + 2); + + Write((ushort)bytesLength); + Encoding.UTF8.GetBytes(value ?? string.Empty, 0, value?.Length ?? 0, _buffer, _position); + Commit(bytesLength); + } + + public void WriteWithLengthPrefix(byte[] payload) + { + GrowIfNeeded(payload.Length + 2); + + Write((ushort)payload.Length); + payload.CopyTo(_buffer, _position); + Commit(payload.Length); + } + + private void Commit(int count) + { + if (_position == Length) + { + Length += count; + } + + _position += count; + } + + private void GrowIfNeeded(int requiredAdditional) + { + var requiredTotal = _position + requiredAdditional; + if (_buffer.Length >= requiredTotal) + { + return; + } + + var newBufferLength = _buffer.Length; + while (newBufferLength < requiredTotal) + { + newBufferLength *= 2; + } + + var newBuffer = _pool.Rent(newBufferLength); + Array.Copy(_buffer, newBuffer, _buffer.Length); + _pool.Return(_buffer); + _buffer = newBuffer; + } + } +} diff --git a/Source/MQTTnet/Formatter/IMqttPacketWriter.cs b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs new file mode 100644 index 00000000..cce98cf5 --- /dev/null +++ b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs @@ -0,0 +1,20 @@ +namespace MQTTnet.Formatter +{ + public interface IMqttPacketWriter + { + int Length { get; } + + void WriteWithLengthPrefix(string value); + void Write(byte returnCode); + void WriteWithLengthPrefix(byte[] payload); + void Write(ushort keepAlivePeriod); + + void Write(IMqttPacketWriter propertyWriter); + void WriteVariableLengthInteger(uint length); + void Write(byte[] payload, int v, int length); + void Reset(int v); + void Seek(int v); + void FreeBuffer(); + byte[] GetBuffer(); + } +} diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 47f80db9..0b44228f 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -12,14 +12,26 @@ namespace MQTTnet.Formatter private IMqttPacketFormatter _formatter; public MqttPacketFormatterAdapter() + : this(new MqttPacketWriter()) + { + } + + public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion) + : this(protocolVersion, new MqttPacketWriter()) { } - public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion) + public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer) + : this(writer) { UseProtocolVersion(protocolVersion); } + public MqttPacketFormatterAdapter(IMqttPacketWriter writer) + { + Writer = writer; + } + public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown; public IMqttDataConverter DataConverter @@ -31,6 +43,8 @@ namespace MQTTnet.Formatter return _formatter.DataConverter; } } + + public IMqttPacketWriter Writer { get; } public ArraySegment Encode(MqttBasePacket packet) { @@ -80,20 +94,20 @@ namespace MQTTnet.Formatter { case MqttProtocolVersion.V500: { - _formatter = new MqttV500PacketFormatter(); + _formatter = new MqttV500PacketFormatter(Writer); break; } case MqttProtocolVersion.V311: { - _formatter = new MqttV311PacketFormatter(); + _formatter = new MqttV311PacketFormatter(Writer); break; } case MqttProtocolVersion.V310: { - _formatter = new MqttV310PacketFormatter(); + _formatter = new MqttV310PacketFormatter(Writer); break; } diff --git a/Source/MQTTnet/Formatter/MqttPacketWriter.cs b/Source/MQTTnet/Formatter/MqttPacketWriter.cs index 1cedd080..d11cc544 100644 --- a/Source/MQTTnet/Formatter/MqttPacketWriter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketWriter.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Formatter /// same as for the original MemoryStream in .net. Also this implementation allows accessing the internal /// buffer for all platforms and .net framework versions (which is not available at the regular MemoryStream). /// - public class MqttPacketWriter + public class MqttPacketWriter : IMqttPacketWriter { private static readonly ArraySegment ZeroVariableLengthIntegerArray = new ArraySegment(new byte[1], 0, 1); private static readonly ArraySegment ZeroTwoByteIntegerArray = new ArraySegment(new byte[2], 0, 2); @@ -33,6 +33,19 @@ namespace MQTTnet.Formatter return (byte)fixedHeader; } + public static int GetLengthOfVariableInteger(uint value) + { + var result = 0; + var x = value; + do + { + x = x / 128; + result++; + } while (x > 0); + + return result; + } + public static ArraySegment EncodeVariableLengthInteger(uint value) { if (value == 0) @@ -129,16 +142,21 @@ namespace MQTTnet.Formatter IncreasePosition(count); } - public void Write(MqttPacketWriter propertyWriter) + public void Write(IMqttPacketWriter propertyWriter) { if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); - if (propertyWriter.Length == 0) + if (propertyWriter is MqttPacketWriter writer) { - return; + if (writer.Length == 0) + { + return; + } + + Write(writer._buffer, 0, writer.Length); } - Write(propertyWriter._buffer, 0, propertyWriter.Length); + throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(MqttPacketWriter).Name}"); } public void Reset(int length) diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs index d3e6b416..70af10e6 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs @@ -10,8 +10,20 @@ namespace MQTTnet.Formatter.V3 public class MqttV310PacketFormatter : IMqttPacketFormatter { private const int FixedHeaderSize = 1; + + private readonly IMqttPacketWriter _packetWriter; + + public MqttV310PacketFormatter() + : this(new MqttPacketWriter()) + { + + } + + public MqttV310PacketFormatter(IMqttPacketWriter packetWriter) + { + _packetWriter = packetWriter; + } - private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter(); public IMqttDataConverter DataConverter { get; } = new MqttV310DataConverter(); @@ -26,17 +38,18 @@ namespace MQTTnet.Formatter.V3 var fixedHeader = EncodePacket(packet, _packetWriter); var remainingLength = (uint)(_packetWriter.Length - 5); - var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength); + var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength); - var headerSize = FixedHeaderSize + remainingLengthBuffer.Count; + var headerSize = FixedHeaderSize + remainingLengthSize; var headerOffset = 5 - headerSize; // Position cursor on correct offset on beginning of array (has leading 0x0) _packetWriter.Seek(headerOffset); _packetWriter.Write(fixedHeader); - _packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count); + _packetWriter.WriteVariableLengthInteger(remainingLength); var buffer = _packetWriter.GetBuffer(); + return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); } @@ -76,7 +89,7 @@ namespace MQTTnet.Formatter.V3 _packetWriter.FreeBuffer(); } - private byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter) + private byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter) { switch (packet) { @@ -331,7 +344,7 @@ namespace MQTTnet.Formatter.V3 } } - protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) + protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter) { ValidateConnectPacket(packet); @@ -393,7 +406,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } - protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter) + protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter) { packetWriter.Write(0); // Reserved. packetWriter.Write((byte)packet.ReturnCode.Value); @@ -401,7 +414,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } - private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -413,7 +426,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } - private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter) { ValidatePublishPacket(packet); @@ -458,7 +471,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); } - private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -470,7 +483,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); } - private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -482,7 +495,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); } - private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -494,7 +507,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); } - private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter) + private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter) { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); @@ -517,7 +530,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); } - private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -537,7 +550,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck); } - private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter) + private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter) { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); @@ -559,7 +572,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); } - private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { diff --git a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs index 44725d4e..a846cd09 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs @@ -6,7 +6,17 @@ namespace MQTTnet.Formatter.V3 { public class MqttV311PacketFormatter : MqttV310PacketFormatter { - protected override byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) + public MqttV311PacketFormatter() + : base() + { + } + + public MqttV311PacketFormatter(IMqttPacketWriter packetWriter) + : base(packetWriter) + { + } + + protected override byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter) { ValidateConnectPacket(packet); @@ -68,7 +78,7 @@ namespace MQTTnet.Formatter.V3 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } - protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter) + protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter) { byte connectAcknowledgeFlags = 0x0; if (packet.IsSessionPresent) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 83f003be..5ccd0301 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -8,7 +8,19 @@ namespace MQTTnet.Formatter.V5 { public class MqttV500PacketEncoder { - private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter(); + private readonly IMqttPacketWriter _packetWriter; + + public MqttV500PacketEncoder() + : this(new MqttPacketWriter()) + { + + } + + public MqttV500PacketEncoder(IMqttPacketWriter packetWriter) + { + _packetWriter = packetWriter; + } + public ArraySegment Encode(MqttBasePacket packet) { @@ -21,15 +33,15 @@ namespace MQTTnet.Formatter.V5 var fixedHeader = EncodePacket(packet, _packetWriter); var remainingLength = (uint)(_packetWriter.Length - 5); - var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength); - - var headerSize = 1 + remainingLengthBuffer.Count; + var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength); + + var headerSize = 1 + remainingLengthSize; var headerOffset = 5 - headerSize; // Position cursor on correct offset on beginning of array (has leading 0x0) _packetWriter.Seek(headerOffset); _packetWriter.Write(fixedHeader); - _packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count); + _packetWriter.WriteVariableLengthInteger(remainingLength); var buffer = _packetWriter.GetBuffer(); return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); @@ -40,7 +52,7 @@ namespace MQTTnet.Formatter.V5 _packetWriter.FreeBuffer(); } - private static byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter) + private static byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter) { switch (packet) { @@ -64,7 +76,7 @@ namespace MQTTnet.Formatter.V5 } } - private static byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter) { if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); @@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } - private static byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter) { if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); @@ -195,7 +207,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } - private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter) { if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); @@ -261,7 +273,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); } - private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter) { if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); @@ -294,7 +306,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); } - private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet); @@ -322,7 +334,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); } - private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet); @@ -350,7 +362,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } - private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter) + private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter) { ThrowIfPacketIdentifierIsInvalid(packet); @@ -378,7 +390,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); } - private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter) + private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter) { if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); @@ -425,7 +437,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); } - private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter) { if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3]."); @@ -450,7 +462,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck); } - private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter) + private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter) { if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); @@ -474,7 +486,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); } - private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter) { if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3]."); @@ -499,7 +511,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); } - private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, IMqttPacketWriter packetWriter) { if (!packet.ReasonCode.HasValue) { @@ -532,7 +544,7 @@ namespace MQTTnet.Formatter.V5 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PingResp); } - private static byte EncodeAuthPacket(MqttAuthPacket packet, MqttPacketWriter packetWriter) + private static byte EncodeAuthPacket(MqttAuthPacket packet, IMqttPacketWriter packetWriter) { packetWriter.Write((byte)packet.ReasonCode); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs index ab52885e..9a52a98d 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs @@ -6,9 +6,19 @@ namespace MQTTnet.Formatter.V5 { public class MqttV500PacketFormatter : IMqttPacketFormatter { - private readonly MqttV500PacketEncoder _encoder = new MqttV500PacketEncoder(); + private readonly MqttV500PacketEncoder _encoder; private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder(); + public MqttV500PacketFormatter() + { + _encoder = new MqttV500PacketEncoder(); + } + + public MqttV500PacketFormatter(IMqttPacketWriter writer) + { + _encoder = new MqttV500PacketEncoder(writer); + } + public IMqttDataConverter DataConverter { get; } = new MqttV500DataConverter(); public ArraySegment Encode(MqttBasePacket mqttPacket) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs index e10790ce..9e5fe39a 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs @@ -63,7 +63,7 @@ namespace MQTTnet.Formatter.V5 Write(MqttPropertyId.AuthenticationMethod, value); } - public void WriteToPacket(MqttPacketWriter packetWriter) + public void WriteToPacket(IMqttPacketWriter packetWriter) { if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs index 65625d34..2d389ab1 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs @@ -13,5 +13,10 @@ namespace MQTTnet.AspNetCore.Tests result.SetBuffer(data); return result; } + + protected override IMqttPacketWriter WriterFactory() + { + return new SpanBasedMqttPacketWriter(); + } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index cb1b0695..e4dd1359 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -528,6 +528,11 @@ namespace MQTTnet.Tests Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data))); } + protected virtual IMqttPacketWriter WriterFactory() + { + return new MqttPacketWriter(); + } + protected virtual IMqttPacketBodyReader ReaderFactory(byte[] data) { return new MqttPacketBodyReader(data, 0, data.Length); @@ -535,14 +540,16 @@ namespace MQTTnet.Tests private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) { + var writer = WriterFactory(); + IMqttPacketFormatter serializer; if (protocolVersion == MqttProtocolVersion.V311) { - serializer = new MqttV311PacketFormatter(); + serializer = new MqttV311PacketFormatter(writer); } else if (protocolVersion == MqttProtocolVersion.V310) { - serializer = new MqttV310PacketFormatter(); + serializer = new MqttV310PacketFormatter(writer); } else { @@ -559,7 +566,8 @@ namespace MQTTnet.Tests using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) { - var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0)); + var reader = ReaderFactory(bodyStream.ToArray()); + var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0)); var buffer2 = serializer.Encode(deserializedPacket); Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2))); @@ -567,17 +575,19 @@ namespace MQTTnet.Tests } } - private static T Roundtrip(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) + private T Roundtrip(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) where T : MqttBasePacket { + var writer = WriterFactory(); + IMqttPacketFormatter serializer; if (protocolVersion == MqttProtocolVersion.V311) { - serializer = new MqttV311PacketFormatter(); + serializer = new MqttV311PacketFormatter(writer); } else if (protocolVersion == MqttProtocolVersion.V310) { - serializer = new MqttV310PacketFormatter(); + serializer = new MqttV310PacketFormatter(writer); } else { @@ -595,7 +605,8 @@ namespace MQTTnet.Tests using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength)) { - return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0)); + var reader = ReaderFactory(bodyStream.ToArray()); + return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0)); } } } From 73b051d15986a58535f235accaffc8712e7d84e4 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 15:00:09 +0200 Subject: [PATCH 09/12] reuse reader --- Source/MQTTnet.AspnetCore/MqttConnectionContext.cs | 7 ++++++- Source/MQTTnet.AspnetCore/ReaderExtensions.cs | 4 +--- Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs | 10 ++++++---- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 400aa070..6d92bf54 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -23,9 +23,14 @@ namespace MQTTnet.AspNetCore _input = Connection.Transport.Input; _output = Connection.Transport.Output; } + + + _reader = new SpanBasedMqttPacketBodyReader(); } + private PipeReader _input; private PipeWriter _output; + private readonly SpanBasedMqttPacketBodyReader _reader; public string Endpoint => Connection.ConnectionId; public bool IsSecureConnection => false; // TODO: Fix detection (WS vs. WSS). @@ -88,7 +93,7 @@ namespace MQTTnet.AspNetCore { if (!buffer.IsEmpty) { - if (PacketFormatterAdapter.TryDecode(buffer, out var packet, out consumed, out observed)) + if (PacketFormatterAdapter.TryDecode(_reader, buffer, out var packet, out consumed, out observed)) { return packet; } diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index e2cb1c44..8af7ecbe 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -9,7 +9,7 @@ namespace MQTTnet.AspNetCore { public static class ReaderExtensions { - public static bool TryDecode(this MqttPacketFormatterAdapter formatter, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) + public static bool TryDecode(this MqttPacketFormatterAdapter formatter, SpanBasedMqttPacketBodyReader reader, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) { if (formatter == null) throw new ArgumentNullException(nameof(formatter)); @@ -36,8 +36,6 @@ namespace MQTTnet.AspNetCore var bodySlice = copy.Slice(0, bodyLength); var buffer = bodySlice.GetMemory(); - - var reader = new SpanBasedMqttPacketBodyReader(); reader.SetBuffer(buffer); var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2); diff --git a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index dedeff33..4d69d37c 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -24,21 +24,23 @@ namespace MQTTnet.AspNetCore.Tests var observed = part.Start; var result = false; + var reader = new SpanBasedMqttPacketBodyReader(); + part = sequence.Slice(sequence.Start, 0); // empty message should fail - result = serializer.TryDecode(part, out packet, out consumed, out observed); + result = serializer.TryDecode(reader, part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence.Slice(sequence.Start, 1); // partial fixed header should fail - result = serializer.TryDecode(part, out packet, out consumed, out observed); + result = serializer.TryDecode(reader, part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence.Slice(sequence.Start, 4); // partial body should fail - result = serializer.TryDecode(part, out packet, out consumed, out observed); + result = serializer.TryDecode(reader, part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence; // complete msg should work - result = serializer.TryDecode(part, out packet, out consumed, out observed); + result = serializer.TryDecode(reader, part, out packet, out consumed, out observed); Assert.IsTrue(result); } } From 9475ccd0d8b94542cd5dc2c7179036bab3c470fc Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 15:31:31 +0200 Subject: [PATCH 10/12] revert accidential change --- Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index 1cd512ba..8cb0d6fc 100644 --- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -46,7 +46,7 @@ namespace MQTTnet.AspNetCore var writer = new SpanBasedMqttPacketWriter(); var formatter = new MqttPacketFormatterAdapter(writer); var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection); - var channelAdapter = new MqttChannelAdapter(channel, formatter, new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); + var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter))); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter); ClientAcceptedHandler?.Invoke(eventArgs); From 03c61ad82c1970b3bb61a5ad9d31147402e7f226 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 15:37:15 +0200 Subject: [PATCH 11/12] fixed tests and added testcase for user property serialization --- .../SpanBasedMqttPacketBodyReader.cs | 5 ++ .../SpanBasedMqttPacketWriter.cs | 9 ++- .../Formatter/MqttPacketFormatterAdapter.cs | 22 ++++---- Source/MQTTnet/Formatter/MqttPacketWriter.cs | 1 + .../MqttPacketSerializer_Tests.cs | 56 ++++++++++--------- 5 files changed, 51 insertions(+), 42 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs index dbfbb0fe..2b118f7b 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs @@ -51,6 +51,11 @@ namespace MQTTnet.AspNetCore public unsafe string ReadStringWithLengthPrefix() { var buffer = ReadSegmentWithLengthPrefix(); + if (buffer.Length == 0) + { + return string.Empty; + } + fixed (byte* bytes = &buffer.GetPinnableReference()) { var result = Encoding.UTF8.GetString(bytes, buffer.Length); diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs index 7ad6e005..27e13c49 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs @@ -60,12 +60,11 @@ namespace MQTTnet.AspNetCore public void Write(IMqttPacketWriter propertyWriter) { - if (propertyWriter is SpanBasedMqttPacketWriter writer) - { - GrowIfNeeded(1); - } + if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); - throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(SpanBasedMqttPacketWriter).Name}"); + GrowIfNeeded(propertyWriter.Length); + Write(propertyWriter.GetBuffer(), 0, propertyWriter.Length); + Commit(propertyWriter.Length); } public void Write(byte[] payload, int start, int length) diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 0b44228f..723b6b8c 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -89,28 +89,30 @@ namespace MQTTnet.Formatter } ProtocolVersion = protocolVersion; + _formatter = GetMqttPacketFormatter(protocolVersion, Writer); + } + public static IMqttPacketFormatter GetMqttPacketFormatter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer) + { + if (protocolVersion == MqttProtocolVersion.Unknown) + { + throw new InvalidOperationException("MQTT protocol version is invalid."); + } + switch (protocolVersion) { case MqttProtocolVersion.V500: { - _formatter = new MqttV500PacketFormatter(Writer); - break; + return new MqttV500PacketFormatter(writer); } - case MqttProtocolVersion.V311: { - _formatter = new MqttV311PacketFormatter(Writer); - break; + return new MqttV311PacketFormatter(writer); } - case MqttProtocolVersion.V310: { - - _formatter = new MqttV310PacketFormatter(Writer); - break; + return new MqttV310PacketFormatter(writer); } - default: { throw new NotSupportedException(); diff --git a/Source/MQTTnet/Formatter/MqttPacketWriter.cs b/Source/MQTTnet/Formatter/MqttPacketWriter.cs index d11cc544..eb7253e0 100644 --- a/Source/MQTTnet/Formatter/MqttPacketWriter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketWriter.cs @@ -154,6 +154,7 @@ namespace MQTTnet.Formatter } Write(writer._buffer, 0, writer.Length); + return; } throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(MqttPacketWriter).Name}"); diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index e4dd1359..429a2312 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -231,6 +231,32 @@ namespace MQTTnet.Tests SerializeAndCompare(p, "Ow4ABUEvQi9DAHtIRUxMTw=="); } + [TestMethod] + public void SerializeV500_MqttPublishPacket() + { + var prop = new MqttPublishPacketProperties(); + + prop.ResponseTopic = "/Response"; + + prop.UserProperties.Add(new MqttUserProperty("Foo", "Bar")); + + var p = new MqttPublishPacket + { + PacketIdentifier = 123, + Dup = true, + Retain = true, + Payload = Encoding.ASCII.GetBytes("HELLO"), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, + Topic = "A/B/C", + Properties = prop + }; + + var deserialized = Roundtrip(p, MqttProtocolVersion.V500); + + Assert.AreEqual(prop.ResponseTopic, deserialized.Properties.ResponseTopic); + Assert.IsTrue(deserialized.Properties.UserProperties.Any(x => x.Name == "Foo")); + } + [TestMethod] public void DeserializeV311_MqttPublishPacket() { @@ -542,19 +568,7 @@ namespace MQTTnet.Tests { var writer = WriterFactory(); - IMqttPacketFormatter serializer; - if (protocolVersion == MqttProtocolVersion.V311) - { - serializer = new MqttV311PacketFormatter(writer); - } - else if (protocolVersion == MqttProtocolVersion.V310) - { - serializer = new MqttV310PacketFormatter(writer); - } - else - { - throw new NotSupportedException(); - } + var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer); var buffer1 = serializer.Encode(packet); @@ -580,20 +594,8 @@ namespace MQTTnet.Tests { var writer = WriterFactory(); - IMqttPacketFormatter serializer; - if (protocolVersion == MqttProtocolVersion.V311) - { - serializer = new MqttV311PacketFormatter(writer); - } - else if (protocolVersion == MqttProtocolVersion.V310) - { - serializer = new MqttV310PacketFormatter(writer); - } - else - { - throw new NotSupportedException(); - } - + var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer); + var buffer1 = serializer.Encode(packet); using (var headerStream = new MemoryStream(Join(buffer1))) From 477e2d23ab2a362ec9192b4ce1a1fa47a45eb48e Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 15 Apr 2019 15:52:51 +0200 Subject: [PATCH 12/12] added benchmark option to compare netfx and core runtime --- .../{ => Configurations}/AllowNonOptimized.cs | 7 ++----- .../Configurations/BaseConfig.cs | 15 +++++++++++++++ .../Configurations/RuntimeCompareConfig.cs | 17 +++++++++++++++++ Tests/MQTTnet.Benchmarks/Program.cs | 3 ++- 4 files changed, 36 insertions(+), 6 deletions(-) rename Tests/MQTTnet.Benchmarks/{ => Configurations}/AllowNonOptimized.cs (54%) create mode 100644 Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs create mode 100644 Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs diff --git a/Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs b/Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs similarity index 54% rename from Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs rename to Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs index ecb31a20..076d1e8d 100644 --- a/Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs +++ b/Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs @@ -3,20 +3,17 @@ using BenchmarkDotNet.Configs; using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Validators; -namespace MQTTnet.Benchmarks +namespace MQTTnet.Benchmarks.Configurations { /// /// this options may be used to run benchmarks in debugmode and attach a performance profiler /// https://benchmarkdotnet.org/Configs/Configs.htm /// - public class AllowNonOptimized : ManualConfig + public class AllowNonOptimized : BaseConfig { public AllowNonOptimized() { Add(JitOptimizationsValidator.DontFailOnError); // ALLOW NON-OPTIMIZED DLLS - Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default - Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default - Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default Add(Job.InProcess); } } diff --git a/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs b/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs new file mode 100644 index 00000000..418bb2f3 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs @@ -0,0 +1,15 @@ +using BenchmarkDotNet.Configs; +using System.Linq; + +namespace MQTTnet.Benchmarks.Configurations +{ + public class BaseConfig : ManualConfig + { + public BaseConfig() + { + Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default + Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default + Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs b/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs new file mode 100644 index 00000000..608271e2 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs @@ -0,0 +1,17 @@ +using BenchmarkDotNet.Configs; +using BenchmarkDotNet.Environments; +using BenchmarkDotNet.Jobs; +using BenchmarkDotNet.Toolchains.CsProj; + +namespace MQTTnet.Benchmarks.Configurations +{ + public class RuntimeCompareConfig : BaseConfig + { + public RuntimeCompareConfig() + { + Add(Job.Default.With(Runtime.Clr)); + Add(Job.Default.With(Runtime.Core).With(CsProjCoreToolchain.NetCoreApp21)); + } + + } +} diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs index 9407ccfe..280416d3 100644 --- a/Tests/MQTTnet.Benchmarks/Program.cs +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -1,5 +1,6 @@ using System; using BenchmarkDotNet.Running; +using MQTTnet.Benchmarks.Configurations; using MQTTnet.Diagnostics; namespace MQTTnet.Benchmarks @@ -43,7 +44,7 @@ namespace MQTTnet.Benchmarks BenchmarkRunner.Run(); break; case '8': - BenchmarkRunner.Run(new AllowNonOptimized()); + BenchmarkRunner.Run(new RuntimeCompareConfig()/*new AllowNonOptimized()*/); break; }