Refactor package deserialization.

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -157,10 +157,6 @@ namespace MQTTnet.Adapter
         private async Task<ReceivedMqttPacket> ReceiveAsync(IMqttChannel channel, CancellationToken cancellationToken)
         {
             var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, cancellationToken).ConfigureAwait(false);
-            if (fixedHeader == null)
-            {
-                return null;
-            }
 
             try
             {

Source/MQTTnet/Serializer/Extensions.cs

@@ -12,7 +12,10 @@ namespace MQTTnet.Serializer
             }
 
             var buffer = new byte[source.Count];
-            Buffer.BlockCopy(source.Array, source.Offset, buffer, 0, buffer.Length);
+            if (buffer.Length > 0)
+            {
+                Array.Copy(source.Array, source.Offset, buffer, 0, buffer.Length);
+            }
 
             return buffer;
         }

Source/MQTTnet/Serializer/MqttFixedHeader.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Serializer
 {
-    public class MqttFixedHeader
+    public struct MqttFixedHeader
     {
         public MqttFixedHeader(byte flags, int remainingLength)
         {
@@ -10,6 +10,6 @@
 
         public byte Flags { get; }
 
-        public int RemainingLength { get; set; }
+        public int RemainingLength { get; }
     }
 }

Source/MQTTnet/Serializer/MqttPacketReader.cs

@@ -1,4 +1,5 @@
-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Exceptions;
@@ -8,23 +9,29 @@ namespace MQTTnet.Serializer
 {
     public static class MqttPacketReader
     {
+        [ThreadStatic]
+        private static byte[] _fixedHeaderBuffer;
+
+        [ThreadStatic]
+        private static byte[] _singleByteBuffer;
+
         public static async Task<MqttFixedHeader> ReadFixedHeaderAsync(IMqttChannel channel, CancellationToken cancellationToken)
         {
             // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length.
             // So in all cases at least 2 bytes must be read for a complete MQTT packet.
             // async/await is used here because the next packet is received in a couple of minutes so the performance
             // impact is acceptable according to a useless waiting thread.
-            var buffer = new byte[2];
+            var buffer = InitializeFixedHeaderBuffer();
             var totalBytesRead = 0;
 
             while (totalBytesRead < buffer.Length)
             {
-                var bytesRead = await channel.ReadAsync(buffer, 0, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
+                var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
                 if (bytesRead <= 0)
                 {
                     if (cancellationToken.IsCancellationRequested)
                     {
-                        return null;
+                        throw new TaskCanceledException();
                     }
 
                     ExceptionHelper.ThrowGracefulSocketClose();
@@ -56,7 +63,7 @@ namespace MQTTnet.Serializer
                 // is too big for reading 1 byte in a row. We expect that the remaining data was sent
                 // directly after the initial bytes. If the client disconnects just in this moment we
                 // will get an exception anyway.
-                encodedByte = ReadByteAsync(channel, cancellationToken).GetAwaiter().GetResult();
+                encodedByte = ReadByte(channel, cancellationToken);
 
                 value += (byte)(encodedByte & 127) * multiplier;
                 if (multiplier > 128 * 128 * 128)
@@ -70,10 +77,10 @@ namespace MQTTnet.Serializer
             return value;
         }
 
-        private static async Task<byte> ReadByteAsync(IMqttChannel channel, CancellationToken cancellationToken)
+        private static byte ReadByte(IMqttChannel channel, CancellationToken cancellationToken)
         {
-            var buffer = new byte[1];
+            var buffer = InitializeSingleByteBuffer();
-            var readCount = await channel.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+            var readCount = channel.ReadAsync(buffer, 0, 1, cancellationToken).GetAwaiter().GetResult();
             if (readCount <= 0)
             {
                 ExceptionHelper.ThrowGracefulSocketClose();
@@ -81,5 +88,15 @@ namespace MQTTnet.Serializer
 
             return buffer[0];
         }
+
+        private static byte[] InitializeFixedHeaderBuffer()
+        {
+            return _fixedHeaderBuffer ?? (_fixedHeaderBuffer = new byte[2]);
+        }
+
+        private static byte[] InitializeSingleByteBuffer()
+        {
+            return _singleByteBuffer ?? (_singleByteBuffer = new byte[1]);
+        }
     }
 }

Source/MQTTnet/Serializer/MqttPacketSerializer.cs

@@ -199,19 +199,18 @@ namespace MQTTnet.Serializer
 
         private static MqttBasePacket DeserializePublish(ReceivedMqttPacket receivedMqttPacket)
         {
-            var body = receivedMqttPacket.Body;
+            ThrowIfBodyIsEmpty(receivedMqttPacket.Body);
-            ThrowIfBodyIsEmpty(body);
 
             var retain = (receivedMqttPacket.FixedHeader & 0x1) > 0;
             var qualityOfServiceLevel = (MqttQualityOfServiceLevel)(receivedMqttPacket.FixedHeader >> 1 & 0x3);
             var dup = (receivedMqttPacket.FixedHeader & 0x3) > 0;
 
-            var topic = body.ReadStringWithLengthPrefix();
+            var topic = receivedMqttPacket.Body.ReadStringWithLengthPrefix();
 
             ushort? packetIdentifier = null;
             if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
-                packetIdentifier = body.ReadUInt16();
+                packetIdentifier = receivedMqttPacket.Body.ReadUInt16();
             }
 
             var packet = new MqttPublishPacket
@@ -219,7 +218,7 @@ namespace MQTTnet.Serializer
                 PacketIdentifier = packetIdentifier,
                 Retain = retain,
                 Topic = topic,
-                Payload = body.ReadRemainingData().ToArray(),
+                Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray(),
                 QualityOfServiceLevel = qualityOfServiceLevel,
                 Dup = dup
             };

Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs

@@ -9,6 +9,7 @@ namespace MQTTnet.Benchmarks
 {
     [ClrJob]
     [RPlotExporter, RankColumn]
+    [MemoryDiagnoser]
     public class MessageProcessingBenchmark
     {
         private IMqttServer _mqttServer;

Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -46,12 +46,19 @@ namespace MQTTnet.Benchmarks
         [Benchmark]
         public void Deserialize_10000_Messages()
         {
+            var channel = new BenchmarkMqttChannel(_serializedPacket);
+
             for (var i = 0; i < 10000; i++)
             {
-                var channel = new BenchmarkMqttChannel(_serializedPacket);
+                channel.Reset();
 
                 var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult();
-                _serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength)));
+
+                var receivedPacket = new ReceivedMqttPacket(
+                    header.Flags,
+                    new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength));
+
+                _serializer.Deserialize(receivedPacket);
             }
         }
 
@@ -68,6 +75,11 @@ namespace MQTTnet.Benchmarks
 
             public string Endpoint { get; }
 
+            public void Reset()
+            {
+                _position = _buffer.Offset;
+            }
+
             public Task ConnectAsync(CancellationToken cancellationToken)
             {
                 throw new NotImplementedException();