Performance optimizations

MQTTnet.Core/Serializer/ByteReader.cs

@@ -4,36 +4,42 @@ namespace MQTTnet.Core.Serializer
 {
     public class ByteReader
     {
+        private readonly int _source;
         private int _index;
-        private readonly int _byte;
 
-        public ByteReader(byte @byte)
+        public ByteReader(int source)
         {
-            _byte = @byte;
+            _source = source;
         }
 
         public bool Read()
         {
             if (_index >= 8)
             {
-                throw new InvalidOperationException("End of the byte reached.");
+                throw new InvalidOperationException("End of byte reached.");
             }
 
-            var result = ((1 << _index) & _byte) > 0;
+            var result = ((1 << _index) & _source) > 0;
             _index++;
-
             return result;
         }
 
         public byte Read(int count)
         {
+            if (_index + count > 8)
+            {
+                throw new InvalidOperationException("End of byte will be reached.");
+            }
+
             var result = 0;
             for (var i = 0; i < count; i++)
             {
-                if (Read())
+                if (((1 << _index) & _source) > 0)
                 {
                     result |= 1 << i;
                 }
+
+                _index++;
             }
 
             return (byte)result;

MQTTnet.Core/Serializer/ByteWriter.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.Core.Serializer
 
         public byte Value => (byte)_byte;
 
-        public void Write(byte @byte, int count)
+        public void Write(int @byte, int count)
         {
             for (var i = 0; i < count; i++)
             {

MQTTnet.Core/Serializer/DefaultMqttV311PacketSerializer.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Serializer
 {
     public class DefaultMqttV311PacketSerializer : IMqttPacketSerializer
     {
-        public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
+        public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
             if (destination == null) throw new ArgumentNullException(nameof(destination));
@@ -20,99 +20,85 @@ namespace MQTTnet.Core.Serializer
             var connectPacket = packet as MqttConnectPacket;
             if (connectPacket != null)
             {
-                await SerializeAsync(connectPacket, destination);
-                return;
+                return SerializeAsync(connectPacket, destination);
             }
 
             var connAckPacket = packet as MqttConnAckPacket;
             if (connAckPacket != null)
             {
-                await SerializeAsync(connAckPacket, destination);
-                return;
+                return SerializeAsync(connAckPacket, destination);
             }
 
             var disconnectPacket = packet as MqttDisconnectPacket;
             if (disconnectPacket != null)
             {
-                await SerializeAsync(disconnectPacket, destination);
-                return;
+                return SerializeAsync(disconnectPacket, destination);
             }
 
             var pingReqPacket = packet as MqttPingReqPacket;
             if (pingReqPacket != null)
             {
-                await SerializeAsync(pingReqPacket, destination);
-                return;
+                return SerializeAsync(pingReqPacket, destination);
             }
 
             var pingRespPacket = packet as MqttPingRespPacket;
             if (pingRespPacket != null)
             {
-                await SerializeAsync(pingRespPacket, destination);
-                return;
+                return SerializeAsync(pingRespPacket, destination);
             }
 
             var publishPacket = packet as MqttPublishPacket;
             if (publishPacket != null)
             {
-                await SerializeAsync(publishPacket, destination);
-                return;
+                return SerializeAsync(publishPacket, destination);
             }
 
             var pubAckPacket = packet as MqttPubAckPacket;
             if (pubAckPacket != null)
             {
-                await SerializeAsync(pubAckPacket, destination);
-                return;
+                return SerializeAsync(pubAckPacket, destination);
             }
 
             var pubRecPacket = packet as MqttPubRecPacket;
             if (pubRecPacket != null)
             {
-                await SerializeAsync(pubRecPacket, destination);
-                return;
+                return SerializeAsync(pubRecPacket, destination);
             }
 
             var pubRelPacket = packet as MqttPubRelPacket;
             if (pubRelPacket != null)
             {
-                await SerializeAsync(pubRelPacket, destination);
-                return;
+                return SerializeAsync(pubRelPacket, destination);
             }
 
             var pubCompPacket = packet as MqttPubCompPacket;
             if (pubCompPacket != null)
             {
-                await SerializeAsync(pubCompPacket, destination);
-                return;
+                return SerializeAsync(pubCompPacket, destination);
             }
 
             var subscribePacket = packet as MqttSubscribePacket;
             if (subscribePacket != null)
             {
-                await SerializeAsync(subscribePacket, destination);
-                return;
+                return SerializeAsync(subscribePacket, destination);
             }
 
             var subAckPacket = packet as MqttSubAckPacket;
             if (subAckPacket != null)
             {
-                await SerializeAsync(subAckPacket, destination);
-                return;
+                return SerializeAsync(subAckPacket, destination);
             }
 
             var unsubscribePacket = packet as MqttUnsubscribePacket;
             if (unsubscribePacket != null)
             {
-                await SerializeAsync(unsubscribePacket, destination);
-                return;
+                return SerializeAsync(unsubscribePacket, destination);
             }
 
             var unsubAckPacket = packet as MqttUnsubAckPacket;
             if (unsubAckPacket != null)
             {
-                await SerializeAsync(unsubAckPacket, destination);
-                return;
+                return SerializeAsync(unsubAckPacket, destination);
             }
 
             throw new MqttProtocolViolationException("Packet type invalid.");
@@ -287,6 +273,7 @@ namespace MQTTnet.Core.Serializer
 
             await reader.ReadRemainingDataByteAsync();
             await reader.ReadRemainingDataByteAsync();
+
             var protocolName = await reader.ReadRemainingDataAsync(4);
 
             if (Encoding.UTF8.GetString(protocolName, 0, protocolName.Length) != "MQTT")
@@ -382,19 +369,17 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private async Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
+        private static readonly byte[] MqttPrefix = Encoding.UTF8.GetBytes("MQTT");
+
+        private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
         {
             ValidateConnectPacket(packet);
 
             using (var output = new MqttPacketWriter())
             {
                 // Write variable header
-                output.Write(0x00); // 3.1.2.1 Protocol Name
-                output.Write(0x04); // ""
-                output.Write('M');
-                output.Write('Q');
-                output.Write('T');
-                output.Write('T');
+                output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
+                output.Write(MqttPrefix);
                 output.Write(0x04); // 3.1.2.2 Protocol Level
 
                 var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
@@ -404,7 +389,7 @@ namespace MQTTnet.Core.Serializer
 
                 if (packet.WillMessage != null)
                 {
-                    connectFlags.Write((byte)packet.WillMessage.QualityOfServiceLevel, 2);
+                    connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2);
                     connectFlags.Write(packet.WillMessage.Retain);
                 }
                 else
@@ -412,7 +397,7 @@ namespace MQTTnet.Core.Serializer
                     connectFlags.Write(0, 2);
                     connectFlags.Write(false);
                 }
-
+                
                 connectFlags.Write(packet.Password != null);
                 connectFlags.Write(packet.Username != null);
 
@@ -437,11 +422,11 @@ namespace MQTTnet.Core.Serializer
                 }
 
                 output.InjectFixedHeader(MqttControlPacketType.Connect);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -452,26 +437,26 @@ namespace MQTTnet.Core.Serializer
                 output.Write((byte)packet.ConnectReturnCode);
 
                 output.InjectFixedHeader(MqttControlPacketType.ConnAck);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
         {
-            await SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination);
+            return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination);
         }
 
-        private async Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination)
         {
-            await SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, destination);
+            return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, destination);
         }
 
-        private async Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination)
         {
-            await SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, destination);
+            return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, destination);
         }
 
-        private async Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination)
         {
             ValidatePublishPacket(packet);
 
@@ -502,29 +487,29 @@ namespace MQTTnet.Core.Serializer
                 fixedHeader.Write(packet.Dup);
 
                 output.InjectFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.Write(packet.PacketIdentifier);
 
                 output.InjectFixedHeader(MqttControlPacketType.PubAck);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.Write(packet.PacketIdentifier);
 
                 output.InjectFixedHeader(MqttControlPacketType.PubRec);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
@@ -539,24 +524,24 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private async Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.Write(packet.PacketIdentifier);
 
                 output.InjectFixedHeader(MqttControlPacketType.PubComp);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.Write(packet.PacketIdentifier);
 
-                if (packet.TopicFilters?.Any() == true)
+                if (packet.TopicFilters?.Count > 0)
                 {
                     foreach (var topicFilter in packet.TopicFilters)
                     {
@@ -566,11 +551,11 @@ namespace MQTTnet.Core.Serializer
                 }
 
                 output.InjectFixedHeader(MqttControlPacketType.Subscribe, 0x02);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -585,11 +570,11 @@ namespace MQTTnet.Core.Serializer
                 }
 
                 output.InjectFixedHeader(MqttControlPacketType.SubAck);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -604,27 +589,27 @@ namespace MQTTnet.Core.Serializer
                 }
 
                 output.InjectFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.Write(packet.PacketIdentifier);
 
                 output.InjectFixedHeader(MqttControlPacketType.UnsubAck);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
 
-        private async Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination)
+        private Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 output.InjectFixedHeader(type);
-                await output.WriteToAsync(destination);
+                return output.WriteToAsync(destination);
             }
         }
     }

MQTTnet.Core/Serializer/MqttPacketReader.cs

@@ -107,15 +107,16 @@ namespace MQTTnet.Core.Serializer
             return await ReadRemainingDataAsync(length);
         }
 
-        public async Task<byte[]> ReadRemainingDataAsync()
+        public Task<byte[]> ReadRemainingDataAsync()
         {
-            return await ReadRemainingDataAsync(RemainingLength - (int)_remainingData.Position);
+            return ReadRemainingDataAsync(RemainingLength - (int)_remainingData.Position);
         }
 
         public async Task<byte[]> ReadRemainingDataAsync(int length)
         {
             var buffer = new byte[length];
             await _remainingData.ReadAsync(buffer, 0, buffer.Length);
+
             return buffer;
         }
 

MQTTnet.Core/Serializer/MqttPacketWriter.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.Core.Serializer
 {
     public sealed class MqttPacketWriter : IDisposable
     {
-        private readonly MemoryStream _buffer = new MemoryStream();
+        private readonly MemoryStream _buffer = new MemoryStream(512);
 
         public void InjectFixedHeader(byte fixedHeader)
         {
@@ -20,31 +20,28 @@ namespace MQTTnet.Core.Serializer
                 return;
             }
 
+            var backupBuffer = _buffer.ToArray();
             var remainingLength = (int)_buffer.Length;
-            using (var buffer = new MemoryStream())
+
+            _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;
+            do
             {
-                _buffer.WriteTo(buffer);
-                _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;
-                do
+                var encodedByte = x % 128;
+                x = x / 128;
+                if (x > 0)
                 {
-                    var encodedByte = (byte)(x % 128);
-                    x = x / 128;
-                    if (x > 0)
-                    {
-                        encodedByte = (byte)(encodedByte | 128);
-                    }
+                    encodedByte = encodedByte | 128;
+                }
 
-                    _buffer.WriteByte(encodedByte);
-                } while (x > 0);
+                _buffer.WriteByte((byte)encodedByte);
+            } while (x > 0);
 
-                buffer.Position = 0;
-                buffer.WriteTo(_buffer);
-            }
+            _buffer.Write(backupBuffer, 0, backupBuffer.Length);
         }
 
         public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0)
@@ -59,11 +56,6 @@ namespace MQTTnet.Core.Serializer
             _buffer.WriteByte(value);
         }
 
-        public void Write(char value)
-        {
-            _buffer.WriteByte((byte)value);
-        }
-
         public void Write(ushort value)
         {
             var buffer = BitConverter.GetBytes(value);
@@ -73,11 +65,15 @@ namespace MQTTnet.Core.Serializer
 
         public void Write(ByteWriter value)
         {
+            if (value == null) throw new ArgumentNullException(nameof(value));
+
             _buffer.WriteByte(value.Value);
         }
 
-        public void Write(byte[] value)
+        public void Write(params byte[] value)
         {
+            if (value == null) throw new ArgumentNullException(nameof(value));
+
             _buffer.Write(value, 0, value.Length);
         }
 
@@ -94,14 +90,16 @@ namespace MQTTnet.Core.Serializer
             Write(value);
         }
 
+        public Task WriteToAsync(IMqttCommunicationChannel destination)
+        {
+            if (destination == null) throw new ArgumentNullException(nameof(destination));
+
+            return destination.WriteAsync(_buffer.ToArray());
+        }
+
         public void Dispose()
         {
             _buffer?.Dispose();
         }
-
-        public async Task WriteToAsync(IMqttCommunicationChannel destination)
-        {
-            await destination.WriteAsync(_buffer.ToArray());
-        }
     }
 }