From fbca8a6dc12aa87730a67cf012aa86f55f7c1b88 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Fri, 3 Aug 2018 22:09:51 +0200 Subject: [PATCH] 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()