Merge pull request #592 from JanEggers/spanV2

performance improvements span based serializers

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

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

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -1,12 +1,13 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
+    <TargetFrameworks>netstandard2.0;netcoreapp2.1</TargetFrameworks>
     <Product />
     <Company />
     <Authors />
     <PackageId />
     <LangVersion>7.2</LangVersion>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -17,8 +17,21 @@ 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;
+            }
+
+
+            _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).
 
@@ -33,20 +46,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();
+            _input?.Complete();
-            Connection.Transport.Output.Complete();
+            _output?.Complete();
 
             return Task.CompletedTask;
         }
@@ -79,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;
                             }
@@ -114,13 +128,19 @@ namespace MQTTnet.AspNetCore
 
         public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
         {
-            var buffer = PacketFormatterAdapter.Encode(packet).AsMemory();
+            var formatter = PacketFormatterAdapter;
-            var output = Connection.Transport.Output;
+           
 
             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
             {

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

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, _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
 
             var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter);
             ClientAcceptedHandler?.Invoke(eventArgs);

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<byte> input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
+        public static bool TryDecode(this MqttPacketFormatterAdapter formatter, SpanBasedMqttPacketBodyReader reader, in ReadOnlySequence<byte> input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
         {
             if (formatter == null) throw new ArgumentNullException(nameof(formatter));
 
@@ -35,18 +35,27 @@ 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));
+            reader.SetBuffer(buffer);
+
+            var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2);
+
+            if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
+            {
+                formatter.DetectProtocolVersion(receivedMqttPacket);
+            }
+
+            packet = formatter.Decode(receivedMqttPacket);
             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 +71,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 +79,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,124 @@
+using MQTTnet.Exceptions;
+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()
+        {
+            return _buffer.Span[_offset++];
+        }
+
+        public byte[] ReadRemainingData()
+        {
+            return _buffer.Slice(_offset).ToArray();
+        }
+
+        public byte[] ReadWithLengthPrefix()
+        {
+            return ReadSegmentWithLengthPrefix().ToArray();
+        }
+
+        private ReadOnlySpan<byte> ReadSegmentWithLengthPrefix()
+        {
+            var span = _buffer.Span;
+            var length = BinaryPrimitives.ReadUInt16BigEndian(span.Slice(_offset));
+
+            var result = span.Slice(_offset+2, length);
+            _offset += 2 + length;
+            return result;
+        }
+        
+        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);
+                return result;
+            }
+        }
+
+        public ushort ReadTwoByteInteger()
+        {
+            var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset));
+            _offset += 2;
+            return result;
+        }
+
+        public uint ReadFourByteInteger()
+        {
+            var result = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_offset));
+            _offset += 4;
+            return result;
+        }
+
+        public uint ReadVariableLengthInteger()
+        {
+            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()
+        {
+            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)
+        {
+            _offset = (int)position;
+        }
+    }
+}

Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs

@@ -0,0 +1,146 @@
+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<byte> _pool;
+
+        public SpanBasedMqttPacketWriter()
+        {
+            _pool = ArrayPool<byte>.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 == null) throw new ArgumentNullException(nameof(propertyWriter));
+
+            GrowIfNeeded(propertyWriter.Length);
+            Write(propertyWriter.GetBuffer(), 0, propertyWriter.Length);
+            Commit(propertyWriter.Length);
+        }
+
+        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;
+        }
+    }
+}

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

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/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<byte> Encode(MqttBasePacket packet)
         {
@@ -75,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();
+                        return new MqttV500PacketFormatter(writer);
-                        break;
                     }
-
                 case MqttProtocolVersion.V311:
                     {
-                        _formatter = new MqttV311PacketFormatter();
+                        return new MqttV311PacketFormatter(writer);
-                        break;
                     }
-
                 case MqttProtocolVersion.V310:
                     {
-
+                        return new MqttV310PacketFormatter(writer);
-                        _formatter = new MqttV310PacketFormatter();
-                        break;
                     }
-                    
                 default:
                     {
                         throw new NotSupportedException();

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).
     /// </summary>
-    public class MqttPacketWriter
+    public class MqttPacketWriter : IMqttPacketWriter
     {
         private static readonly ArraySegment<byte> ZeroVariableLengthIntegerArray = new ArraySegment<byte>(new byte[1], 0, 1);
         private static readonly ArraySegment<byte> ZeroTwoByteIntegerArray = new ArraySegment<byte>(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<byte> EncodeVariableLengthInteger(uint value)
         {
             if (value == 0)
@@ -129,16 +142,22 @@ 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)
             {
+                if (writer.Length == 0)
+                {
+                    return;
+                }
+
+                Write(writer._buffer, 0, writer.Length);
                 return;
             }
 
-            Write(propertyWriter._buffer, 0, propertyWriter.Length);
+            throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(MqttPacketWriter).Name}");
         }
 
         public void Reset(int length)

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;
 
@@ -11,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();
 
@@ -27,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<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
         }
 
@@ -77,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)
             {
@@ -100,7 +112,7 @@ namespace MQTTnet.Formatter.V3
             }
         }
 
-        private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -110,7 +122,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -120,7 +132,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -130,7 +142,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -140,7 +152,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -167,7 +179,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -223,7 +235,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body)
+        private MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -264,7 +276,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 +296,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -301,7 +313,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        protected virtual MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
+        protected virtual MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -332,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);
 
@@ -394,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);
@@ -402,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)
             {
@@ -414,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);
 
@@ -459,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)
             {
@@ -471,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)
             {
@@ -483,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)
             {
@@ -495,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].");
 
@@ -518,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)
             {
@@ -538,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].");
 
@@ -560,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)
             {
@@ -577,7 +589,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

@@ -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)
@@ -82,7 +92,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/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<byte> 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 remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
-
+            
-            var headerSize = 1 + remainingLengthBuffer.Count;
+            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<byte>(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);
 

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<byte> Encode(MqttBasePacket mqttPacket)

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

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

Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj

@@ -14,7 +14,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
-    <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
+    <ProjectReference Include="..\MQTTnet.Core.Tests\MQTTnet.Tests.csproj" />
   </ItemGroup>
 
 </Project>

Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs

@@ -0,0 +1,22 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Formatter;
+using MQTTnet.Tests;
+
+namespace MQTTnet.AspNetCore.Tests
+{
+    [TestClass]
+    public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializer_Tests
+    {
+        protected override IMqttPacketBodyReader ReaderFactory(byte[] data)
+        {
+            var result = new SpanBasedMqttPacketBodyReader();
+            result.SetBuffer(data);
+            return result;
+        }
+
+        protected override IMqttPacketWriter WriterFactory()
+        {
+            return new SpanBasedMqttPacketWriter();
+        }
+    }
+}

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

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
 {
     /// <summary>
     /// this options may be used to run benchmarks in debugmode and attach a performance profiler
     /// https://benchmarkdotnet.org/Configs/Configs.htm
     /// </summary>
-    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);
         }
     }

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

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

Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <OutputType>Exe</OutputType>
     <DebugType>Full</DebugType>
-    <TargetFramework Condition=" '$(OS)' == 'Windows_NT' ">net461</TargetFramework>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">net461;netcoreapp2.1</TargetFrameworks>
     <TargetFramework Condition=" '$(OS)' != 'Windows_NT' ">netcoreapp2.1</TargetFramework>
     <LangVersion>7.2</LangVersion>
   </PropertyGroup>

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<TcpPipesBenchmark>();
                     break;
                 case '8':
-                    BenchmarkRunner.Run<MessageProcessingMqttConnectionContextBenchmark>(new AllowNonOptimized());
+                    BenchmarkRunner.Run<MessageProcessingMqttConnectionContextBenchmark>(new RuntimeCompareConfig()/*new AllowNonOptimized()*/);
                     break;
             }
 

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()
         {
@@ -528,21 +554,21 @@ namespace MQTTnet.Tests
             Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data)));
         }
 
-        private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+        protected virtual IMqttPacketWriter WriterFactory()
         {
-            IMqttPacketFormatter serializer;
+            return new MqttPacketWriter();
-            if (protocolVersion == MqttProtocolVersion.V311)
+        }
-            {
+
-                serializer = new MqttV311PacketFormatter();
+        protected virtual IMqttPacketBodyReader ReaderFactory(byte[] data)
-            }
+        {
-            else if (protocolVersion == MqttProtocolVersion.V310)
+            return new MqttPacketBodyReader(data, 0, data.Length);
-            {
+        }
-                serializer = new MqttV310PacketFormatter();
+
-            }
+        private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
-            else
+        {
-            {
+            var writer = WriterFactory();
-                throw new NotSupportedException();
+
-            }
+            var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
 
             var buffer1 = serializer.Encode(packet);
 
@@ -554,7 +580,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)));
@@ -562,23 +589,13 @@ namespace MQTTnet.Tests
             }
         }
 
-        private static T Roundtrip<T>(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+        private T Roundtrip<T>(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
             where T : MqttBasePacket
         {
-            IMqttPacketFormatter serializer;
+            var writer = WriterFactory();
-            if (protocolVersion == MqttProtocolVersion.V311)
-            {
-                serializer = new MqttV311PacketFormatter();
-            }
-            else if (protocolVersion == MqttProtocolVersion.V310)
-            {
-                serializer = new MqttV310PacketFormatter();
-            }
-            else
-            {
-                throw new NotSupportedException();
-            }
 
+            var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
+            
             var buffer1 = serializer.Encode(packet);
 
             using (var headerStream = new MemoryStream(Join(buffer1)))
@@ -590,7 +607,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));
                 }
             }
         }