Merge pull request #36 from JanEggers/Buffering

added perf test app

Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs

@@ -10,6 +10,7 @@ using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Serializer;
 using MQTTnet.Core.Server;
+using MQTTnet.Core.Channel;
 
 namespace MQTTnet.Implementations
 {
@@ -86,7 +87,7 @@ namespace MQTTnet.Implementations
                 try
                 {
                     var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new BufferedCommunicationChannel(new MqttTcpChannel(clientSocket, null)), new MqttPacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception) when (!(exception is ObjectDisposedException))

Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs

@@ -93,17 +93,15 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public async Task ReadAsync(byte[] buffer)
+        public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
         {
-            if (buffer == null) throw new ArgumentNullException(nameof(buffer));
-
             try
             {
                 var totalBytes = 0;
 
                 do
                 {
-                    var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false);
+                    var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false);
                     if (read == 0)
                     {
                         throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting));
@@ -111,7 +109,8 @@ namespace MQTTnet.Implementations
 
                     totalBytes += read;
                 }
-                while (totalBytes < buffer.Length);
+                while (totalBytes < length);
+                return new ArraySegment<byte>(buffer, 0, length);
             }
             catch (SocketException exception)
             {
@@ -143,5 +142,10 @@ namespace MQTTnet.Implementations
 
             return certificates;
         }
+
+        public int Peek()
+        {
+            return _socket.Available;
+        }
     }
 }

Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs

@@ -11,9 +11,6 @@ namespace MQTTnet.Implementations
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
         private ClientWebSocket _webSocket = new ClientWebSocket();
-        private const int BufferSize = 4096;
-        private const int BufferAmplifier = 20;
-        private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier];
         private int WebSocketBufferSize;
         private int WebSocketBufferOffset;
 
@@ -42,51 +39,40 @@ namespace MQTTnet.Implementations
             _webSocket?.Dispose();
         }
 
-        public Task ReadAsync(byte[] buffer)
+        public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
         {
-            return Task.WhenAll(ReadToBufferAsync(buffer));
+            await ReadToBufferAsync(length, buffer).ConfigureAwait(false);
+            
+            var result = new ArraySegment<byte>(buffer, WebSocketBufferOffset, length);
+            WebSocketBufferSize -= length;
+            WebSocketBufferOffset += length;
+            
+            return result;
         }
 
-        private async Task ReadToBufferAsync(byte[] buffer)
+        private async Task ReadToBufferAsync(int length, byte[] buffer)
         {
-            var temporaryBuffer = new byte[BufferSize];
+            if (WebSocketBufferSize > 0)
-            var offset = 0;
-
-            while (_webSocket.State == WebSocketState.Open)
             {
-                if (WebSocketBufferSize == 0)
-                {
-                    WebSocketBufferOffset = 0;
-
-                    WebSocketReceiveResult response;
-                    do
-                    {
-                        response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(temporaryBuffer), CancellationToken.None).ConfigureAwait(false);
-
-                        temporaryBuffer.CopyTo(WebSocketBuffer, offset);
-                        offset += response.Count;
-                        temporaryBuffer = new byte[BufferSize];
-                    } while (!response.EndOfMessage);
-
-                    WebSocketBufferSize = response.Count;
-                    if (response.MessageType == WebSocketMessageType.Close)
-                    {
-                        await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
-                    }
-
-                    Buffer.BlockCopy(WebSocketBuffer, 0, buffer, 0, buffer.Length);
-                    WebSocketBufferSize -= buffer.Length;
-                    WebSocketBufferOffset += buffer.Length;
-                }
-                else
-                {
-                    Buffer.BlockCopy(WebSocketBuffer, WebSocketBufferOffset, buffer, 0, buffer.Length);
-                    WebSocketBufferSize -= buffer.Length;
-                    WebSocketBufferOffset += buffer.Length;
-                }
-
                 return;
             }
+
+            var offset = 0;
+            while (_webSocket.State == WebSocketState.Open && WebSocketBufferSize < length)
+            {
+                WebSocketReceiveResult response;
+                do
+                {
+                    response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false);
+                    offset += response.Count;
+                } while (!response.EndOfMessage);
+
+                WebSocketBufferSize = response.Count;
+                if (response.MessageType == WebSocketMessageType.Close)
+                {
+                    await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
+                }
+            }
         }
 
         public Task WriteAsync(byte[] buffer)
@@ -105,5 +91,10 @@ namespace MQTTnet.Implementations
                 throw new MqttCommunicationException(exception);
             }
         }
+
+        public int Peek()
+        {
+            return WebSocketBufferSize;
+        }
     }
 }

Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs

@@ -22,7 +22,7 @@ namespace MQTTnet
             {
                 case MqttConnectionType.Tcp:
                 case MqttConnectionType.Tls:
-                    return new MqttTcpChannel();
+                    return new BufferedCommunicationChannel( new MqttTcpChannel() );
                 case MqttConnectionType.Ws:
                 case MqttConnectionType.Wss:
                     return new MqttWebSocketChannel();

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs

@@ -92,17 +92,15 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public async Task ReadAsync(byte[] buffer)
+        public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
         {
-            if (buffer == null) throw new ArgumentNullException(nameof(buffer));
-
             try
             {
                 var totalBytes = 0;
 
                 do
                 {
-                    var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false);
+                    var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false);
                     if (read == 0)
                     {
                         throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting));
@@ -110,7 +108,8 @@ namespace MQTTnet.Implementations
 
                     totalBytes += read;
                 }
-                while (totalBytes < buffer.Length);
+                while (totalBytes < length);
+                return new ArraySegment<byte>(buffer, 0, length);
             }
             catch (SocketException exception)
             {
@@ -142,5 +141,10 @@ namespace MQTTnet.Implementations
 
             return certificates;
         }
+
+        public int Peek()
+        {
+            return _socket.Available;
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs

@@ -11,9 +11,6 @@ namespace MQTTnet.Implementations
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
         private ClientWebSocket _webSocket = new ClientWebSocket();
-        private const int BufferSize = 4096;
-        private const int BufferAmplifier = 20;
-        private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier];
         private int WebSocketBufferSize;
         private int WebSocketBufferOffset;
 
@@ -42,51 +39,40 @@ namespace MQTTnet.Implementations
             _webSocket?.Dispose();
         }
 
-        public Task ReadAsync(byte[] buffer)
+        public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
         {
-            return Task.WhenAll(ReadToBufferAsync(buffer));
+            await ReadToBufferAsync(length, buffer).ConfigureAwait(false);
+
+            var result = new ArraySegment<byte>(buffer, WebSocketBufferOffset, length);
+            WebSocketBufferSize -= length;
+            WebSocketBufferOffset += length;
+
+            return result;
         }
 
-        private async Task ReadToBufferAsync(byte[] buffer)
+        private async Task ReadToBufferAsync(int length, byte[] buffer)
         {
-            var temporaryBuffer = new byte[BufferSize];
+            if (WebSocketBufferSize > 0)
-            var offset = 0;
-
-            while (_webSocket.State == WebSocketState.Open)
             {
-                if (WebSocketBufferSize == 0)
-                {
-                    WebSocketBufferOffset = 0;
-
-                    WebSocketReceiveResult response;
-                    do
-                    {
-                        response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(temporaryBuffer), CancellationToken.None).ConfigureAwait(false);
-
-                        temporaryBuffer.CopyTo(WebSocketBuffer, offset);
-                        offset += response.Count;
-                        temporaryBuffer = new byte[BufferSize];
-                    } while (!response.EndOfMessage);
-
-                    WebSocketBufferSize = response.Count;
-                    if (response.MessageType == WebSocketMessageType.Close)
-                    {
-                        await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
-                    }
-
-                    Buffer.BlockCopy(WebSocketBuffer, 0, buffer, 0, buffer.Length);
-                    WebSocketBufferSize -= buffer.Length;
-                    WebSocketBufferOffset += buffer.Length;
-                }
-                else
-                {
-                    Buffer.BlockCopy(WebSocketBuffer, WebSocketBufferOffset, buffer, 0, buffer.Length);
-                    WebSocketBufferSize -= buffer.Length;
-                    WebSocketBufferOffset += buffer.Length;
-                }
-
                 return;
             }
+
+            var offset = 0;
+            while (_webSocket.State == WebSocketState.Open && WebSocketBufferSize < length)
+            {
+                WebSocketReceiveResult response;
+                do
+                {
+                    response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false);
+                    offset += response.Count;
+                } while (!response.EndOfMessage);
+
+                WebSocketBufferSize = response.Count;
+                if (response.MessageType == WebSocketMessageType.Close)
+                {
+                    await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
+                }
+            }
         }
 
         public Task WriteAsync(byte[] buffer)
@@ -105,5 +91,10 @@ namespace MQTTnet.Implementations
                 throw new MqttCommunicationException(exception);
             }
         }
+
+        public int Peek()
+        {
+            return WebSocketBufferSize;
+        }
     }
 }

MQTTnet.Core/Channel/BufferConstants.cs

@@ -0,0 +1,7 @@
+namespace MQTTnet.Core.Channel
+{
+    public static class BufferConstants
+    {
+        public const int Size = 4096 * 20;
+    }
+}

MQTTnet.Core/Channel/BufferedCommunicationChannel.cs

@@ -0,0 +1,77 @@
+using System.Threading.Tasks;
+using MQTTnet.Core.Client;
+using System;
+
+namespace MQTTnet.Core.Channel
+{
+    public class BufferedCommunicationChannel : IMqttCommunicationChannel
+    {
+        private IMqttCommunicationChannel _inner { get; }
+        private int _bufferSize = 0;
+        private int _bufferOffset = 0;
+
+        public BufferedCommunicationChannel(IMqttCommunicationChannel inner)
+        {
+            _inner = inner;
+        }
+
+        public Task ConnectAsync(MqttClientOptions options)
+        {
+            return _inner.ConnectAsync(options);
+        }
+
+        public Task DisconnectAsync()
+        {
+            return _inner.DisconnectAsync();
+        }
+
+        public int Peek()
+        {
+            return _inner.Peek();
+        }
+
+        public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
+        {
+            //read from buffer
+            if (_bufferSize > 0)
+            {
+                return ReadFomBuffer(length, buffer);
+            }
+
+            var available = _inner.Peek();
+            // if there are less or equal bytes available then requested then just read em
+            if (available <= length)
+            {
+                return await _inner.ReadAsync(length, buffer);
+            }
+
+            //if more bytes are available than requested do buffer them to reduce calls to network buffers
+            await WriteToBuffer(available, buffer).ConfigureAwait(false);
+            return ReadFomBuffer(length, buffer);
+        }
+
+        private async Task WriteToBuffer(int available, byte[] buffer)
+        {
+            await _inner.ReadAsync(available, buffer).ConfigureAwait(false);
+            _bufferSize = available;
+            _bufferOffset = 0;
+        }
+
+        private ArraySegment<byte> ReadFomBuffer(int length, byte[] buffer)
+        {
+            var result = new ArraySegment<byte>(buffer, _bufferOffset, length);
+            _bufferSize -= length;
+            _bufferOffset += length;
+
+            if (_bufferSize < 0)
+            {
+            }
+            return result;
+        }
+
+        public Task WriteAsync(byte[] buffer)
+        {
+            return _inner.WriteAsync(buffer);
+        }
+    }
+}

MQTTnet.Core/Channel/IMqttCommunicationChannel.cs

@@ -1,5 +1,6 @@
 using System.Threading.Tasks;
 using MQTTnet.Core.Client;
+using System;
 
 namespace MQTTnet.Core.Channel
 {
@@ -11,6 +12,11 @@ namespace MQTTnet.Core.Channel
 
         Task WriteAsync(byte[] buffer);
 
-        Task ReadAsync(byte[] buffer);
+        /// <summary>
+        /// get the currently available number of bytes without reading them
+        /// </summary>
+        int Peek();
+
+        Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer);
     }
 }

MQTTnet.Core/Packets/MqttPacketHeader.cs

@@ -0,0 +1,13 @@
+using MQTTnet.Core.Protocol;
+
+namespace MQTTnet.Core.Packets
+{
+    public class MqttPacketHeader
+    {
+        public MqttControlPacketType ControlPacketType { get; set; }
+
+        public byte FixedHeader { get; set; }
+
+        public int BodyLength { get; set; }
+    }
+}

MQTTnet.Core/Serializer/MqttPacketReader.cs

@@ -2,55 +2,28 @@
 using System.IO;
 using System.Text;
 using System.Threading.Tasks;
-using MQTTnet.Core.Channel;
 using MQTTnet.Core.Exceptions;
 using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Channel;
+using MQTTnet.Core.Packets;
 
 namespace MQTTnet.Core.Serializer
 {
-    public sealed class MqttPacketReader : IDisposable
+    public sealed class MqttPacketReader : BinaryReader
     {
-        private readonly MemoryStream _remainingData = new MemoryStream(1024);
+        private readonly MqttPacketHeader _header;
-        private readonly IMqttCommunicationChannel _source;
 
-        private int _remainingLength;
+        public MqttPacketReader(Stream stream, MqttPacketHeader header)
-
+            : base(stream)
-        public MqttPacketReader(IMqttCommunicationChannel source)
         {
-            _source = source ?? throw new ArgumentNullException(nameof(source));
+            _header = header;
         }
-
+        
-        public MqttControlPacketType ControlPacketType { get; private set; }
+        public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
-
+        
-        public byte FixedHeader { get; private set; }
+        public override ushort ReadUInt16()
-
-        public bool EndOfRemainingData => _remainingData.Position == _remainingData.Length;
-
-        public async Task ReadToEndAsync()
         {
-            await ReadFixedHeaderAsync().ConfigureAwait(false);
+            var buffer = ReadBytes(2);
-            await ReadRemainingLengthAsync().ConfigureAwait(false);
-
-            if (_remainingLength == 0)
-            {
-                return;
-            }
-
-            var buffer = new byte[_remainingLength];
-            await ReadFromSourceAsync(buffer).ConfigureAwait(false);
-            
-            _remainingData.Write(buffer, 0, buffer.Length);
-            _remainingData.Position = 0;
-        }
-
-        public byte ReadRemainingDataByte()
-        {
-            return ReadRemainingData(1)[0];
-        }
-
-        public ushort ReadRemainingDataUShort()
-        {
-            var buffer = ReadRemainingData(2);
 
             var temp = buffer[0];
             buffer[0] = buffer[1];
@@ -59,31 +32,58 @@ namespace MQTTnet.Core.Serializer
             return BitConverter.ToUInt16(buffer, 0);
         }
 
-        public string ReadRemainingDataStringWithLengthPrefix()
+        public string ReadStringWithLengthPrefix()
         {
-            var buffer = ReadRemainingDataWithLengthPrefix();
+            var buffer = ReadWithLengthPrefix();
             return Encoding.UTF8.GetString(buffer, 0, buffer.Length);
         }
 
-        public byte[] ReadRemainingDataWithLengthPrefix()
+        public byte[] ReadWithLengthPrefix()
         {
-            var length = ReadRemainingDataUShort();
+            var length = ReadUInt16();
-            return ReadRemainingData(length);
+            return ReadBytes(length);
         }
 
         public byte[] ReadRemainingData()
         {
-            return ReadRemainingData(_remainingLength - (int)_remainingData.Position);
+            return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
         }
 
-        public byte[] ReadRemainingData(int length)
+        public static async Task<MqttPacketHeader> ReadHeaderFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
         {
-            var buffer = new byte[length];
+            var fixedHeader = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
-            _remainingData.Read(buffer, 0, buffer.Length);
+            var byteReader = new ByteReader(fixedHeader);
-            return buffer;
+            byteReader.Read(4);
+            var controlPacketType = (MqttControlPacketType)byteReader.Read(4);
+            var bodyLength = await ReadBodyLengthFromSourceAsync(source, buffer).ConfigureAwait(false);
+
+            return new MqttPacketHeader()
+            {
+                FixedHeader = fixedHeader,
+                ControlPacketType = controlPacketType,
+                BodyLength = bodyLength
+            };
         }
 
-        private async Task ReadRemainingLengthAsync()
+        private static async Task<byte> ReadStreamByteAsync(IMqttCommunicationChannel source, byte[] readBuffer)
+        {
+            var result = await ReadFromSourceAsync(source, 1, readBuffer).ConfigureAwait(false);
+            return result.Array[result.Offset];
+        }
+
+        public static async Task<ArraySegment<byte>> ReadFromSourceAsync(IMqttCommunicationChannel source, int length, byte[] buffer)
+        {
+            try
+            {
+                return await source.ReadAsync(length, buffer);
+            }
+            catch (Exception exception)
+            {
+                throw new MqttCommunicationException(exception);
+            }
+        }
+        
+        private static async Task<int> ReadBodyLengthFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
         {
             // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
             var multiplier = 1;
@@ -91,7 +91,7 @@ namespace MQTTnet.Core.Serializer
             byte encodedByte;
             do
             {
-                encodedByte = await ReadStreamByteAsync().ConfigureAwait(false);
+                encodedByte = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
                 value += (encodedByte & 127) * multiplier;
                 multiplier *= 128;
                 if (multiplier > 128 * 128 * 128)
@@ -99,41 +99,7 @@ namespace MQTTnet.Core.Serializer
                     throw new MqttProtocolViolationException("Remaining length is ivalid.");
                 }
             } while ((encodedByte & 128) != 0);
-
+            return value;
-            _remainingLength = value;
-        }
-
-        private Task ReadFromSourceAsync(byte[] buffer)
-        {
-            try
-            {
-                return _source.ReadAsync(buffer);
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
-        }
-
-        private async Task<byte> ReadStreamByteAsync()
-        {
-            var buffer = new byte[1];
-            await ReadFromSourceAsync(buffer).ConfigureAwait(false);
-            return buffer[0];
-        }
-
-        private async Task ReadFixedHeaderAsync()
-        {
-            FixedHeader = await ReadStreamByteAsync().ConfigureAwait(false);
-
-            var byteReader = new ByteReader(FixedHeader);
-            byteReader.Read(4);
-            ControlPacketType = (MqttControlPacketType)byteReader.Read(4);
-        }
-
-        public void Dispose()
-        {
-            _remainingData?.Dispose();
         }
     }
 }

MQTTnet.Core/Serializer/MqttPacketSerializer.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
@@ -15,80 +17,97 @@ namespace MQTTnet.Core.Serializer
         private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");
 
         public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
+        private byte[] _readBuffer = new byte[BufferConstants.Size]; 
 
-        public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
+        public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
             if (destination == null) throw new ArgumentNullException(nameof(destination));
 
+            using (var stream = new MemoryStream())
+            using (var writer = new MqttPacketWriter(stream))
+            {
+                var header = new List<byte>();
+                header.Add(SerializePacket(packet, writer));
+
+                var body = stream.ToArray();
+                MqttPacketWriter.BuildLengthHeader(body.Length, header);
+                
+                await destination.WriteAsync(header.ToArray()).ConfigureAwait(false);
+                await destination.WriteAsync(body).ConfigureAwait(false);
+            }
+        }
+
+        private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter writer)
+        {
             if (packet is MqttConnectPacket connectPacket)
             {
-                return SerializeAsync(connectPacket, destination);
+                return Serialize(connectPacket, writer);
             }
 
             if (packet is MqttConnAckPacket connAckPacket)
             {
-                return SerializeAsync(connAckPacket, destination);
+                return Serialize(connAckPacket, writer);
             }
 
             if (packet is MqttDisconnectPacket disconnectPacket)
             {
-                return SerializeAsync(disconnectPacket, destination);
+                return Serialize(disconnectPacket, writer);
             }
 
             if (packet is MqttPingReqPacket pingReqPacket)
             {
-                return SerializeAsync(pingReqPacket, destination);
+                return Serialize(pingReqPacket, writer);
             }
 
             if (packet is MqttPingRespPacket pingRespPacket)
             {
-                return SerializeAsync(pingRespPacket, destination);
+                return Serialize(pingRespPacket, writer);
             }
 
             if (packet is MqttPublishPacket publishPacket)
             {
-                return SerializeAsync(publishPacket, destination);
+                return Serialize(publishPacket, writer);
             }
 
             if (packet is MqttPubAckPacket pubAckPacket)
             {
-                return SerializeAsync(pubAckPacket, destination);
+                return Serialize(pubAckPacket, writer);
             }
 
             if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return SerializeAsync(pubRecPacket, destination);
+                return Serialize(pubRecPacket, writer);
             }
 
             if (packet is MqttPubRelPacket pubRelPacket)
             {
-                return SerializeAsync(pubRelPacket, destination);
+                return Serialize(pubRelPacket, writer);
             }
 
             if (packet is MqttPubCompPacket pubCompPacket)
             {
-                return SerializeAsync(pubCompPacket, destination);
+                return Serialize(pubCompPacket, writer);
             }
 
             if (packet is MqttSubscribePacket subscribePacket)
             {
-                return SerializeAsync(subscribePacket, destination);
+                return Serialize(subscribePacket, writer);
             }
 
             if (packet is MqttSubAckPacket subAckPacket)
             {
-                return SerializeAsync(subAckPacket, destination);
+                return Serialize(subAckPacket, writer);
             }
 
             if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
-                return SerializeAsync(unsubscribePacket, destination);
+                return Serialize(unsubscribePacket, writer);
             }
 
             if (packet is MqttUnsubAckPacket unsubAckPacket)
             {
-                return SerializeAsync(unsubAckPacket, destination);
+                return Serialize(unsubAckPacket, writer);
             }
 
             throw new MqttProtocolViolationException("Packet type invalid.");
@@ -98,11 +117,13 @@ namespace MQTTnet.Core.Serializer
         {
             if (source == null) throw new ArgumentNullException(nameof(source));
 
-            using (var mqttPacketReader = new MqttPacketReader(source))
+            var header = await MqttPacketReader.ReadHeaderFromSourceAsync(source, _readBuffer).ConfigureAwait(false);
-            {
-                await mqttPacketReader.ReadToEndAsync().ConfigureAwait(false);
 
-                switch (mqttPacketReader.ControlPacketType)
+            var body = await GetBody(source, header).ConfigureAwait(false);
+
+            using (var mqttPacketReader = new MqttPacketReader(body, header))
+            {
+                switch (header.ControlPacketType)
                 {
                     case MqttControlPacketType.Connect:
                         {
@@ -121,14 +142,14 @@ namespace MQTTnet.Core.Serializer
 
                     case MqttControlPacketType.Publish:
                         {
-                            return DeserializePublish(mqttPacketReader);
+                            return DeserializePublish(mqttPacketReader, header);
                         }
 
                     case MqttControlPacketType.PubAck:
                         {
                             return new MqttPubAckPacket
                             {
-                                PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort()
+                                PacketIdentifier = mqttPacketReader.ReadUInt16()
                             };
                         }
 
@@ -136,7 +157,7 @@ namespace MQTTnet.Core.Serializer
                         {
                             return new MqttPubRecPacket
                             {
-                                PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort()
+                                PacketIdentifier = mqttPacketReader.ReadUInt16()
                             };
                         }
 
@@ -144,7 +165,7 @@ namespace MQTTnet.Core.Serializer
                         {
                             return new MqttPubRelPacket
                             {
-                                PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort()
+                                PacketIdentifier = mqttPacketReader.ReadUInt16()
                             };
                         }
 
@@ -152,7 +173,7 @@ namespace MQTTnet.Core.Serializer
                         {
                             return new MqttPubCompPacket
                             {
-                                PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort()
+                                PacketIdentifier = mqttPacketReader.ReadUInt16()
                             };
                         }
 
@@ -185,28 +206,39 @@ namespace MQTTnet.Core.Serializer
                         {
                             return new MqttUnsubAckPacket
                             {
-                                PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort()
+                                PacketIdentifier = mqttPacketReader.ReadUInt16()
                             };
                         }
 
                     default:
                         {
-                            throw new MqttProtocolViolationException($"Packet type ({(int)mqttPacketReader.ControlPacketType}) not supported.");
+                            throw new MqttProtocolViolationException($"Packet type ({(int)header.ControlPacketType}) not supported.");
                         }
                 }
             }
         }
 
+        private async Task<MemoryStream> GetBody(IMqttCommunicationChannel source, MqttPacketHeader header)
+        {
+            if (header.BodyLength > 0)
+            {
+                var segment = await MqttPacketReader.ReadFromSourceAsync(source, header.BodyLength, _readBuffer).ConfigureAwait(false);
+                return new MemoryStream(segment.Array, segment.Offset, segment.Count);
+            }
+
+            return new MemoryStream();
+        }
+
         private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader)
         {
             var packet = new MqttUnsubscribePacket
             {
-                PacketIdentifier = reader.ReadRemainingDataUShort(),
+                PacketIdentifier = reader.ReadUInt16(),
             };
 
             while (!reader.EndOfRemainingData)
             {
-                packet.TopicFilters.Add(reader.ReadRemainingDataStringWithLengthPrefix());
+                packet.TopicFilters.Add(reader.ReadStringWithLengthPrefix());
             }
 
             return packet;
@@ -216,32 +248,32 @@ namespace MQTTnet.Core.Serializer
         {
             var packet = new MqttSubscribePacket
             {
-                PacketIdentifier = reader.ReadRemainingDataUShort()
+                PacketIdentifier = reader.ReadUInt16()
             };
 
             while (!reader.EndOfRemainingData)
             {
                 packet.TopicFilters.Add(new TopicFilter(
-                    reader.ReadRemainingDataStringWithLengthPrefix(),
+                    reader.ReadStringWithLengthPrefix(),
-                    (MqttQualityOfServiceLevel)reader.ReadRemainingDataByte()));
+                    (MqttQualityOfServiceLevel)reader.ReadByte()));
             }
 
             return packet;
         }
 
-        private static MqttBasePacket DeserializePublish(MqttPacketReader reader)
+        private static MqttBasePacket DeserializePublish(MqttPacketReader reader, MqttPacketHeader mqttPacketHeader)
         {
-            var fixedHeader = new ByteReader(reader.FixedHeader);
+            var fixedHeader = new ByteReader(mqttPacketHeader.FixedHeader);
             var retain = fixedHeader.Read();
             var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2);
             var dup = fixedHeader.Read();
 
-            var topic = reader.ReadRemainingDataStringWithLengthPrefix();
+            var topic = reader.ReadStringWithLengthPrefix();
 
             ushort packetIdentifier = 0;
             if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
-                packetIdentifier = reader.ReadRemainingDataUShort();
+                packetIdentifier = reader.ReadUInt16();
             }
 
             var packet = new MqttPublishPacket
@@ -259,13 +291,13 @@ namespace MQTTnet.Core.Serializer
 
         private static MqttBasePacket DeserializeConnect(MqttPacketReader reader)
         {
-            reader.ReadRemainingData(2); // Skip 2 bytes
+            reader.ReadBytes(2); // Skip 2 bytes
 
             MqttProtocolVersion protocolVersion;
-            var protocolName = reader.ReadRemainingData(4);
+            var protocolName = reader.ReadBytes(4);
             if (protocolName.SequenceEqual(ProtocolVersionV310Name))
             {
-                reader.ReadRemainingData(2);
+                reader.ReadBytes(2);
                 protocolVersion = MqttProtocolVersion.V310;
             }
             else if (protocolName.SequenceEqual(ProtocolVersionV311Name))
@@ -277,8 +309,8 @@ namespace MQTTnet.Core.Serializer
                 throw new MqttProtocolViolationException("Protocol name is not supported.");
             }
 
-            var protocolLevel = reader.ReadRemainingDataByte();
+            var protocolLevel = reader.ReadByte();
-            var connectFlags = reader.ReadRemainingDataByte();
+            var connectFlags = reader.ReadByte();
 
             var connectFlagsReader = new ByteReader(connectFlags);
             connectFlagsReader.Read(); // Reserved.
@@ -295,26 +327,26 @@ namespace MQTTnet.Core.Serializer
             var passwordFlag = connectFlagsReader.Read();
             var usernameFlag = connectFlagsReader.Read();
 
-            packet.KeepAlivePeriod = reader.ReadRemainingDataUShort();
+            packet.KeepAlivePeriod = reader.ReadUInt16();
-            packet.ClientId = reader.ReadRemainingDataStringWithLengthPrefix();
+            packet.ClientId = reader.ReadStringWithLengthPrefix();
 
             if (willFlag)
             {
                 packet.WillMessage = new MqttApplicationMessage(
-                    reader.ReadRemainingDataStringWithLengthPrefix(),
+                    reader.ReadStringWithLengthPrefix(),
-                    reader.ReadRemainingDataWithLengthPrefix(),
+                    reader.ReadWithLengthPrefix(),
                     (MqttQualityOfServiceLevel)willQoS,
                     willRetain);
             }
 
             if (usernameFlag)
             {
-                packet.Username = reader.ReadRemainingDataStringWithLengthPrefix();
+                packet.Username = reader.ReadStringWithLengthPrefix();
             }
 
             if (passwordFlag)
             {
-                packet.Password = reader.ReadRemainingDataStringWithLengthPrefix();
+                packet.Password = reader.ReadStringWithLengthPrefix();
             }
 
             ValidateConnectPacket(packet);
@@ -325,12 +357,12 @@ namespace MQTTnet.Core.Serializer
         {
             var packet = new MqttSubAckPacket
             {
-                PacketIdentifier = reader.ReadRemainingDataUShort()
+                PacketIdentifier = reader.ReadUInt16()
             };
 
             while (!reader.EndOfRemainingData)
             {
-                packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)reader.ReadRemainingDataByte());
+                packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)reader.ReadByte());
             }
 
             return packet;
@@ -338,8 +370,8 @@ namespace MQTTnet.Core.Serializer
 
         private static MqttBasePacket DeserializeConnAck(MqttPacketReader reader)
         {
-            var variableHeader1 = reader.ReadRemainingDataByte();
+            var variableHeader1 = reader.ReadByte();
-            var variableHeader2 = reader.ReadRemainingDataByte();
+            var variableHeader2 = reader.ReadByte();
 
             var packet = new MqttConnAckPacket
             {
@@ -366,260 +398,212 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
+        private byte Serialize(MqttConnectPacket packet, MqttPacketWriter writer)
         {
             ValidateConnectPacket(packet);
 
-            using (var output = new MqttPacketWriter())
+            // Write variable header
+            writer.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
+            if (ProtocolVersion == MqttProtocolVersion.V311)
             {
-                // Write variable header
+                writer.Write(ProtocolVersionV311Name);
-                output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
+                writer.Write(0x04); // 3.1.2.2 Protocol Level (4)
-                if (ProtocolVersion == MqttProtocolVersion.V311)
-                {
-                    output.Write(ProtocolVersionV311Name);
-                    output.Write(0x04); // 3.1.2.2 Protocol Level (4)
-                }
-                else
-                {
-                    output.Write(ProtocolVersionV310Name);
-                    output.Write(0x64);
-                    output.Write(0x70);
-                    output.Write(0x03); // Protocol Level (3)
-                }
-                
-                var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
-                connectFlags.Write(false); // Reserved
-                connectFlags.Write(packet.CleanSession);
-                connectFlags.Write(packet.WillMessage != null);
-
-                if (packet.WillMessage != null)
-                {
-                    connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2);
-                    connectFlags.Write(packet.WillMessage.Retain);
-                }
-                else
-                {
-                    connectFlags.Write(0, 2);
-                    connectFlags.Write(false);
-                }
-                
-                connectFlags.Write(packet.Password != null);
-                connectFlags.Write(packet.Username != null);
-
-                output.Write(connectFlags);
-                output.Write(packet.KeepAlivePeriod);
-                output.WriteWithLengthPrefix(packet.ClientId);
-
-                if (packet.WillMessage != null)
-                {
-                    output.WriteWithLengthPrefix(packet.WillMessage.Topic);
-                    output.WriteWithLengthPrefix(packet.WillMessage.Payload);
-                }
-
-                if (packet.Username != null)
-                {
-                    output.WriteWithLengthPrefix(packet.Username);
-                }
-
-                if (packet.Password != null)
-                {
-                    output.WriteWithLengthPrefix(packet.Password);
-                }
-
-                output.InjectFixedHeader(MqttControlPacketType.Connect);
-                return output.WriteToAsync(destination);
             }
-        }
+            else
-
-        private Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
             {
-                var connectAcknowledgeFlags = new ByteWriter();
+                writer.Write(ProtocolVersionV310Name);
-
+                writer.Write(0x64);
-                if (ProtocolVersion == MqttProtocolVersion.V311)
+                writer.Write(0x70);
-                {
+                writer.Write(0x03); // Protocol Level (3)
-                    connectAcknowledgeFlags.Write(packet.IsSessionPresent);
-                }
-                
-                output.Write(connectAcknowledgeFlags);
-                output.Write((byte)packet.ConnectReturnCode);
-
-                output.InjectFixedHeader(MqttControlPacketType.ConnAck);
-                return output.WriteToAsync(destination);
             }
-        }
 
-        private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
+            var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
-        {
+            connectFlags.Write(false); // Reserved
-            using (var output = new MqttPacketWriter())
+            connectFlags.Write(packet.CleanSession);
+            connectFlags.Write(packet.WillMessage != null);
+
+            if (packet.WillMessage != null)
             {
-                output.Write(packet.PacketIdentifier);
+                connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2);
-
+                connectFlags.Write(packet.WillMessage.Retain);
-                output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02);
-                await output.WriteToAsync(destination).ConfigureAwait(false);
             }
+            else
+            {
+                connectFlags.Write(0, 2);
+                connectFlags.Write(false);
+            }
+
+            connectFlags.Write(packet.Password != null);
+            connectFlags.Write(packet.Username != null);
+
+            writer.Write(connectFlags);
+            writer.Write(packet.KeepAlivePeriod);
+            writer.WriteWithLengthPrefix(packet.ClientId);
+
+            if (packet.WillMessage != null)
+            {
+                writer.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                writer.WriteWithLengthPrefix(packet.WillMessage.Payload);
+            }
+
+            if (packet.Username != null)
+            {
+                writer.WriteWithLengthPrefix(packet.Username);
+            }
+
+            if (packet.Password != null)
+            {
+                writer.WriteWithLengthPrefix(packet.Password);
+            }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
         }
 
-        private static Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
+        private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter writer)
         {
-            return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination);
+            var connectAcknowledgeFlags = new ByteWriter();
+
+            if (ProtocolVersion == MqttProtocolVersion.V311)
+            {
+                connectAcknowledgeFlags.Write(packet.IsSessionPresent);
+            }
+
+            writer.Write(connectAcknowledgeFlags);
+            writer.Write((byte)packet.ConnectReturnCode);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
         }
 
-        private static Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination)
+        private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer)
         {
-            return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, destination);
+            writer.Write(packet.PacketIdentifier);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
         }
 
-        private static Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination)
+        private static byte Serialize(MqttDisconnectPacket packet, MqttPacketWriter writer)
         {
-            return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, destination);
+            return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, writer);
         }
 
-        private static Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination)
+        private static byte Serialize(MqttPingReqPacket packet, MqttPacketWriter writer)
+        {
+            return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, writer);
+        }
+
+        private static byte Serialize(MqttPingRespPacket packet, MqttPacketWriter writer)
+        {
+            return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, writer);
+        }
+
+        private static byte Serialize(MqttPublishPacket packet, MqttPacketWriter writer)
         {
             ValidatePublishPacket(packet);
 
-            using (var output = new MqttPacketWriter())
+            writer.WriteWithLengthPrefix(packet.Topic);
-            {
-                output.WriteWithLengthPrefix(packet.Topic);
 
-                if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+            if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                writer.Write(packet.PacketIdentifier);
+            }
+            else
+            {
+                if (packet.PacketIdentifier > 0)
                 {
-                    output.Write(packet.PacketIdentifier);
+                    throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5].");
                 }
-                else
+            }
+
+            if (packet.Payload?.Length > 0)
+            {
+                writer.Write(packet.Payload);
+            }
+
+            var fixedHeader = new ByteWriter();
+            fixedHeader.Write(packet.Retain);
+            fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2);
+            fixedHeader.Write(packet.Dup);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value);
+        }
+
+        private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer)
+        {
+            writer.Write(packet.PacketIdentifier);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
+        }
+
+        private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer)
+        {
+            writer.Write(packet.PacketIdentifier);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
+        }
+
+        private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer)
+        {
+            writer.Write(packet.PacketIdentifier);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
+        }
+
+        private static byte Serialize(MqttSubscribePacket packet, MqttPacketWriter writer)
+        {
+            writer.Write(packet.PacketIdentifier);
+
+            if (packet.TopicFilters?.Count > 0)
+            {
+                foreach (var topicFilter in packet.TopicFilters)
                 {
-                    if (packet.PacketIdentifier > 0)
+                    writer.WriteWithLengthPrefix(topicFilter.Topic);
-                    {
+                    writer.Write((byte)topicFilter.QualityOfServiceLevel);
-                        throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5].");
-                    }
                 }
+            }
 
-                if (packet.Payload?.Length > 0)
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
+        }
+
+        private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer)
+        {
+            writer.Write(packet.PacketIdentifier);
+
+            if (packet.SubscribeReturnCodes?.Any() == true)
+            {
+                foreach (var packetSubscribeReturnCode in packet.SubscribeReturnCodes)
                 {
-                    output.Write(packet.Payload);
+                    writer.Write((byte)packetSubscribeReturnCode);
                 }
-
-                var fixedHeader = new ByteWriter();
-                fixedHeader.Write(packet.Retain);
-                fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2);
-                fixedHeader.Write(packet.Dup);
-
-                output.InjectFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value);
-                return output.WriteToAsync(destination);
             }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
         }
 
-        private static Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination)
+        private static byte Serialize(MqttUnsubscribePacket packet, MqttPacketWriter writer)
         {
-            using (var output = new MqttPacketWriter())
+            writer.Write(packet.PacketIdentifier);
+
+            if (packet.TopicFilters?.Any() == true)
             {
-                output.Write(packet.PacketIdentifier);
+                foreach (var topicFilter in packet.TopicFilters)
-
-                output.InjectFixedHeader(MqttControlPacketType.PubAck);
-                return output.WriteToAsync(destination);
-            }
-        }
-
-        private static Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                output.InjectFixedHeader(MqttControlPacketType.PubRec);
-                return output.WriteToAsync(destination);
-            }
-        }
-
-        private static Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                output.InjectFixedHeader(MqttControlPacketType.PubComp);
-                return output.WriteToAsync(destination);
-            }
-        }
-
-        private static Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                if (packet.TopicFilters?.Count > 0)
                 {
-                    foreach (var topicFilter in packet.TopicFilters)
+                    writer.WriteWithLengthPrefix(topicFilter);
-                    {
-                        output.WriteWithLengthPrefix(topicFilter.Topic);
-                        output.Write((byte)topicFilter.QualityOfServiceLevel);
-                    }
                 }
-
-                output.InjectFixedHeader(MqttControlPacketType.Subscribe, 0x02);
-                return output.WriteToAsync(destination);
             }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
         }
 
-        private static Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination)
+        private static byte Serialize(MqttUnsubAckPacket packet, MqttPacketWriter writer)
         {
-            using (var output = new MqttPacketWriter())
+            writer.Write(packet.PacketIdentifier);
-            {
-                output.Write(packet.PacketIdentifier);
 
-                if (packet.SubscribeReturnCodes?.Any() == true)
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
-                {
-                    foreach (var packetSubscribeReturnCode in packet.SubscribeReturnCodes)
-                    {
-                        output.Write((byte)packetSubscribeReturnCode);
-                    }
-                }
-
-                output.InjectFixedHeader(MqttControlPacketType.SubAck);
-                return output.WriteToAsync(destination);
-            }
         }
 
-        private static Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination)
+        private static byte SerializeEmptyPacketAsync(MqttControlPacketType type, MqttPacketWriter writer)
         {
-            using (var output = new MqttPacketWriter())
+            return MqttPacketWriter.BuildFixedHeader(type);
-            {
-                output.Write(packet.PacketIdentifier);
-
-                if (packet.TopicFilters?.Any() == true)
-                {
-                    foreach (var topicFilter in packet.TopicFilters)
-                    {
-                        output.WriteWithLengthPrefix(topicFilter);
-                    }
-                }
-
-                output.InjectFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
-                return output.WriteToAsync(destination);
-            }
-        }
-
-        private static Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                output.InjectFixedHeader(MqttControlPacketType.UnsubAck);
-                return output.WriteToAsync(destination);
-            }
-        }
-
-        private static Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.InjectFixedHeader(type);
-                return output.WriteToAsync(destination);
-            }
         }
     }
 }

MQTTnet.Core/Serializer/MqttPacketWriter.cs

@@ -1,49 +1,45 @@
 using System;
 using System.IO;
 using System.Text;
-using System.Threading.Tasks;
-using MQTTnet.Core.Channel;
 using MQTTnet.Core.Protocol;
+using System.Collections.Generic;
 
 namespace MQTTnet.Core.Serializer
 {
-    public sealed class MqttPacketWriter : IDisposable
+    public sealed class MqttPacketWriter : BinaryWriter
     {
-        private readonly MemoryStream _buffer = new MemoryStream(1024);
+        public MqttPacketWriter( Stream stream )
+            : base(stream)
+        {
 
-        public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0)
+        }
+
+        public static byte BuildFixedHeader(MqttControlPacketType packetType, byte flags = 0)
         {
             var fixedHeader = (int)packetType << 4;
             fixedHeader |= flags;
-            InjectFixedHeader((byte)fixedHeader);
+            return (byte)fixedHeader;
         }
-
+        
-        public void Write(byte value)
+        public override void Write(ushort value)
-        {
-            _buffer.WriteByte(value);
-        }
-
-        public void Write(ushort value)
         {
             var buffer = BitConverter.GetBytes(value);
-            _buffer.WriteByte(buffer[1]);
+            Write(buffer[1]);
-            _buffer.WriteByte(buffer[0]);
+            Write(buffer[0]);
+        }
+
+        public new void Write(params byte[] values)
+        {
+            base.Write(values);
         }
 
         public void Write(ByteWriter value)
         {
             if (value == null) throw new ArgumentNullException(nameof(value));
 
-            _buffer.WriteByte(value.Value);
+            Write(value.Value);
         }
-
+        
-        public void Write(params byte[] value)
-        {
-            if (value == null) throw new ArgumentNullException(nameof(value));
-
-            _buffer.Write(value, 0, value.Length);
-        }
-
         public void WriteWithLengthPrefix(string value)
         {
             WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
@@ -57,36 +53,16 @@ namespace MQTTnet.Core.Serializer
             Write(value);
         }
 
-        public Task WriteToAsync(IMqttCommunicationChannel destination)
+        public static void BuildLengthHeader(int length, List<byte> header)
         {
-            if (destination == null) throw new ArgumentNullException(nameof(destination));
+            if (length == 0)
-
-            return destination.WriteAsync(_buffer.ToArray());
-        }
-
-        public void Dispose()
-        {
-            _buffer?.Dispose();
-        }
-
-        private void InjectFixedHeader(byte fixedHeader)
-        {
-            if (_buffer.Length == 0)
             {
-                Write(fixedHeader);
+                header.Add(0);
-                Write(0);
                 return;
             }
 
-            var backupBuffer = _buffer.ToArray();
-            var remainingLength = (int)_buffer.Length;
-
-            _buffer.SetLength(0);
-
-            _buffer.WriteByte(fixedHeader);
-
             // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
-            var x = remainingLength;
+            var x = length;
             do
             {
                 var encodedByte = x % 128;
@@ -96,10 +72,8 @@ namespace MQTTnet.Core.Serializer
                     encodedByte = encodedByte | 128;
                 }
 
-                _buffer.WriteByte((byte)encodedByte);
+                header.Add((byte)encodedByte);
             } while (x > 0);
-
-            _buffer.Write(backupBuffer, 0, backupBuffer.Length);
         }
     }
 }

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -418,15 +418,21 @@ namespace MQTTnet.Core.Tests
                 return _stream.WriteAsync(buffer, 0, buffer.Length);
             }
 
-            public Task ReadAsync(byte[] buffer)
+            public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
             {
-                return _stream.ReadAsync(buffer, 0, buffer.Length);
+                await _stream.ReadAsync(buffer, 0, length);
+                return new ArraySegment<byte>(buffer, 0, length);
             }
 
             public byte[] ToArray()
             {
                 return _stream.ToArray();
             }
+
+            public int Peek()
+            {
+                return (int)_stream.Length - (int)_stream.Position;
+            }
         }
 
         private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)

Tests/MQTTnet.TestApp.NetFramework/MQTTnet.TestApp.NetFramework.csproj

@@ -82,6 +82,7 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="PerformanceTest.cs" />
     <Compile Include="Program.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
   </ItemGroup>

Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs

@@ -0,0 +1,165 @@
+using MQTTnet.Core;
+using MQTTnet.Core.Client;
+using MQTTnet.Core.Packets;
+using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Server;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace MQTTnet.TestApp.NetFramework
+{
+    public static class PerformanceTest
+    {
+        public static async Task RunAsync()
+        {
+            var server = Task.Run(() => RunServerAsync());
+            var client = Task.Run(() => RunClientAsync(50, TimeSpan.FromMilliseconds(10)));
+
+            await Task.WhenAll(server, client).ConfigureAwait(false);
+        }
+
+        private static async Task RunClientAsync( int msgChunkSize, TimeSpan interval )
+        {
+            try
+            {
+                var options = new MqttClientOptions
+                {
+                    Server = "localhost",
+                    ClientId = "XYZ",
+                    CleanSession = true,
+                    DefaultCommunicationTimeout = TimeSpan.FromMinutes(10)
+                };
+
+                var client = new MqttClientFactory().CreateMqttClient(options);
+                client.ApplicationMessageReceived += (s, e) =>
+                {
+                };
+
+                client.Connected += async (s, e) =>
+                {
+                    Console.WriteLine("### CONNECTED WITH SERVER ###");
+
+                    await client.SubscribeAsync(new List<TopicFilter>
+                    {
+                        new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)
+                    });
+
+                    Console.WriteLine("### SUBSCRIBED ###");
+                };
+
+                client.Disconnected += async (s, e) =>
+                {
+                    Console.WriteLine("### DISCONNECTED FROM SERVER ###");
+                    await Task.Delay(TimeSpan.FromSeconds(5));
+
+                    try
+                    {
+                        await client.ConnectAsync();
+                    }
+                    catch
+                    {
+                        Console.WriteLine("### RECONNECTING FAILED ###");
+                    }
+                };
+
+                try
+                {
+                    await client.ConnectAsync();
+                }
+                catch (Exception exception)
+                {
+                    Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception);
+                }
+
+                Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
+
+                var last = DateTime.Now;
+                var msgs = 0;
+
+                while (true)
+                {
+                    for (int i = 0; i < msgChunkSize; i++)
+                    {
+                        var applicationMessage = new MqttApplicationMessage(
+                            "A/B/C",
+                            Encoding.UTF8.GetBytes("Hello World"),
+                            MqttQualityOfServiceLevel.AtLeastOnce,
+                            false
+                        );
+
+                        //do not await to send as much messages as possible 
+                        await client.PublishAsync(applicationMessage);
+                        msgs++;
+                    }
+
+                    var now = DateTime.Now;
+                    if (last < now - TimeSpan.FromSeconds(1))
+                    {
+                        Console.WriteLine( $"sending {msgs} inteded {msgChunkSize / interval.TotalSeconds}" );
+                        msgs = 0;
+                        last = now;
+                    }
+
+                    await Task.Delay(interval).ConfigureAwait(false);
+                }
+            }
+            catch (Exception exception)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+
+        private static void RunServerAsync()
+        {
+            try
+            {
+                var options = new MqttServerOptions
+                {
+                    ConnectionValidator = p =>
+                    {
+                        if (p.ClientId == "SpecialClient")
+                        {
+                            if (p.Username != "USER" || p.Password != "PASS")
+                            {
+                                return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
+                            }
+                        }
+
+                        return MqttConnectReturnCode.ConnectionAccepted;
+                    },
+                    DefaultCommunicationTimeout = TimeSpan.FromMinutes(10)
+                };
+                
+                var mqttServer = new MqttServerFactory().CreateMqttServer(options);
+                var last = DateTime.Now;
+                var msgs = 0;
+                mqttServer.ApplicationMessageReceived += (sender, args) => 
+                {
+                    msgs++;
+                    var now = DateTime.Now;
+                    if (last < now - TimeSpan.FromSeconds(1))
+                    {
+                        Console.WriteLine($"received {msgs}");
+                        msgs = 0;
+                        last = now;
+                    }
+                };
+                mqttServer.Start();
+
+                Console.WriteLine("Press any key to exit.");
+                Console.ReadLine();
+
+                mqttServer.Stop();
+            }
+            catch (Exception e)
+            {
+                Console.WriteLine(e);
+            }
+
+            Console.ReadLine();
+        }
+    }
+}

Tests/MQTTnet.TestApp.NetFramework/Program.cs

@@ -19,6 +19,7 @@ namespace MQTTnet.TestApp.NetFramework
             Console.WriteLine("MQTTnet - TestApp.NetFramework");
             Console.WriteLine("1 = Start client");
             Console.WriteLine("2 = Start server");
+            Console.WriteLine("3 = Start performance test");
             var pressedKey = Console.ReadKey(true);
             if (pressedKey.Key == ConsoleKey.D1)
             {
@@ -30,6 +31,11 @@ namespace MQTTnet.TestApp.NetFramework
                 Task.Run(() => RunServerAsync(args));
                 Thread.Sleep(Timeout.Infinite);
             }
+            else if (pressedKey.Key == ConsoleKey.D3)
+            {
+                Task.Run(() => PerformanceTest.RunAsync());
+                Thread.Sleep(Timeout.Infinite);
+            }
         }
 
         private static async Task RunClientAsync(string[] arguments)