span based reader

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -7,6 +7,7 @@
     <Authors />
     <PackageId />
     <LangVersion>7.2</LangVersion>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -54,6 +54,7 @@ namespace MQTTnet.AspNetCore
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             var input = Connection.Transport.Input;
+            var reader = new SpanBasedMqttPacketBodyReader();
 
             try
             {

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -35,18 +35,22 @@ namespace MQTTnet.AspNetCore
             }
 
             var bodySlice = copy.Slice(0, bodyLength);
-            var buffer = bodySlice.GetArray();
+            var buffer = bodySlice.GetMemory();
-            packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length), buffer.Length + 2));
+
+            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<byte> input)
+        private static ReadOnlyMemory<byte> GetMemory(this in ReadOnlySequence<byte> 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;

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<byte> _buffer;
+        private int _offset;
+        
+        public void SetBuffer(ReadOnlyMemory<byte> 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<byte> 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();
+        }
+    }
+}

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; }
     }

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);
+    }
+}

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<byte> ReadRemainingData()
+        public byte[] ReadRemainingData()
         {
-            return new ArraySegment<byte>(_buffer, (int)Offset, (int)(_length - Offset));
+            return new ArraySegment<byte>(_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<byte> ReadWithLengthPrefix()
+        public byte[] ReadWithLengthPrefix()
+        {
+            return ReadSegmentWithLengthPrefix().ToArray();
+        }
+
+        private ArraySegment<byte> 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);
         }
     }

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)
             {

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);
 

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)
             {

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()