1592 bug of mqttbufferreader (#1593)

Source/MQTTnet.Benchmarks/MqttBufferReaderBenchmark.cs

@@ -0,0 +1,43 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Text;
+using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using MQTTnet.Formatter;
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net60)]
+    [MemoryDiagnoser]
+    public class MqttBufferReaderBenchmark
+    {
+        byte[] _buffer;
+        int _bufferLength;
+
+        [GlobalSetup]
+        public void GlobalSetup()
+        {
+            var writer = new MqttBufferWriter(1024, 1024);
+            writer.WriteString("hgfjkdfkjlghfdjghdfljkdfhgdlkjfshgsldkfjghsdflkjghdsflkjhrstiuoghlkfjbhnfbutghjoiöjhklötnbhtroliöuhbjntluiobkjzbhtdrlskbhtruhjkfthgbkftgjhgfiklhotriuöhbjtrsioöbtrsötrhträhtrühjtriüoätrhjtsrölbktrbnhtrulöbionhströloubinströoliubhnsöotrunbtöroisntröointröioujhgötiohjgötorshjnbgtöorihbnjtröoihbjntröobntröoibntrjhötrohjbtröoihntröoibnrtoiöbtrjnboöitrhjtnriohötrhjtöroihjtroöihjtroösibntsroönbotöirsbntöoihjntröoihntroöbtrboöitrnhoöitrhjntröoishbnjtröosbhtröbntriohjtröoijtöoitbjöotibjnhöotirhbjntroöibhnjrtoöibnhtroöibnhtörsbnhtöoirbnhtöroibntoörhjnbträöbtrbträbtrbtirbätrsibohjntrsöiobthnjiohjsrtoib");
+
+            _buffer = writer.GetBuffer();
+            _bufferLength = writer.Length;
+        }
+
+        [Benchmark]
+        public void Use_Span()
+        {
+            var span = _buffer.AsSpan(0, _bufferLength);
+            Encoding.UTF8.GetString(span);
+        }
+        
+        [Benchmark]
+        public void Use_Encoding()
+        {
+            Encoding.UTF8.GetString(_buffer, 0, _bufferLength);
+        }
+    }
+}

Source/MQTTnet.Benchmarks/Program.cs

@@ -30,6 +30,7 @@ namespace MQTTnet.Benchmarks
             Console.WriteLine("d = UnsubscribeBenchmark");
             Console.WriteLine("e = MessageDeliveryBenchmark");
             Console.WriteLine("f = AsyncLockBenchmark");
+            Console.WriteLine("g = MqttBufferReaderBenchmark");
 
             var pressedKey = Console.ReadKey(true);
             switch (pressedKey.KeyChar)
@@ -79,9 +80,12 @@ namespace MQTTnet.Benchmarks
                 case 'f':
                     BenchmarkRunner.Run<AsyncLockBenchmark>();
                     break;
+                case 'g':
+                    BenchmarkRunner.Run<MqttBufferReaderBenchmark>();
+                    break;
             }
 
             Console.ReadLine();
         }
     }
-}
+}

Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs

@@ -47,10 +47,12 @@ namespace MQTTnet.Tests.Client
 
                 await managedClient.StartAsync(managedClientOptions);
                 await LongTestDelay();
+                await LongTestDelay();
                 
                 // Send test data.
                 await senderClient.PublishStringAsync("topic1");
                 await LongTestDelay();
+                await LongTestDelay();
                 
                 receivedMessages.AssertReceivedCountEquals(1);
 
@@ -329,13 +331,7 @@ namespace MQTTnet.Tests.Client
                 await testEnvironment.StartServer().ConfigureAwait(false);
 
                 var sendingClient = await testEnvironment.ConnectClient().ConfigureAwait(false);
-                await sendingClient.PublishAsync(
-                    new MqttApplicationMessage
-                    {
-                        Topic = "topic",
-                        Payload = new byte[] { 1 },
-                        Retain = true
-                    });
+                await sendingClient.PublishStringAsync("topic", "A", retain: true);
 
                 // Wait a bit for the retained message to be available
                 await LongTestDelay();
@@ -357,18 +353,15 @@ namespace MQTTnet.Tests.Client
 
                 await managedClient.SubscribeAsync("topic");
 
-                await managedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).WithAutoReconnectDelay(TimeSpan.FromSeconds(1)).Build());
-
+                await managedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).WithAutoReconnectDelay(TimeSpan.FromSeconds(0.5)).Build());
                 await LongTestDelay();
 
                 Assert.AreEqual(1, receivedManagedMessages.Count);
 
                 await managedClient.StopAsync();
-
                 await LongTestDelay();
 
-                await managedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).WithAutoReconnectDelay(TimeSpan.FromSeconds(1)).Build());
-
+                await managedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).WithAutoReconnectDelay(TimeSpan.FromSeconds(0.5)).Build());
                 await LongTestDelay();
 
                 // After reconnect and then some delay, the retained message must not be received,
@@ -377,7 +370,6 @@ namespace MQTTnet.Tests.Client
 
                 // Make sure that it gets received after subscribing again.
                 await managedClient.SubscribeAsync("topic");
-                
                 await LongTestDelay();
                 
                 Assert.AreEqual(2, receivedManagedMessages.Count);

Source/MQTTnet.Tests/Formatter/MqttBufferReader_Tests.cs

@@ -0,0 +1,110 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+
+namespace MQTTnet.Tests.Formatter
+{
+    [TestClass]
+    public sealed class MqttBufferReader_Tests
+    {
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException), "Expected at least 4 bytes but there are only 3 bytes")]
+        public void Fire_Exception_If_Not_Enough_Data()
+        {
+            var buffer = new byte[] { 0, 1, 2 };
+
+            var reader = new MqttBufferReader();
+
+            reader.SetBuffer(buffer, 0, 3);
+
+            // 1 byte is missing.
+            reader.ReadFourByteInteger();
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException), "Expected at least 4 bytes but there are only 3 bytes")]
+        public void Fire_Exception_If_Not_Enough_Data_With_Longer_Buffer()
+        {
+            var buffer = new byte[] { 0, 1, 2, 3, 4, 5, 6 };
+
+            var reader = new MqttBufferReader();
+
+            reader.SetBuffer(buffer, 0, 3);
+
+            // 1 byte is missing.
+            reader.ReadFourByteInteger();
+        }
+
+        [TestMethod]
+        public void Is_EndOfStream_Without_Buffer()
+        {
+            var reader = new MqttBufferReader();
+
+            Assert.IsTrue(reader.EndOfStream);
+            Assert.AreEqual(0, reader.BytesLeft);
+        }
+
+        [TestMethod]
+        public void Read_Remaining_Data_From_Larger_Buffer()
+        {
+            var buffer = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+
+            var reader = new MqttBufferReader();
+
+            // The used buffer contains more data than used!
+            reader.SetBuffer(buffer, 0, 5);
+
+            // This should only read 5 bytes even if more data is in the buffer
+            // due to custom bounds.
+            var remainingData = reader.ReadRemainingData();
+
+            Assert.IsTrue(reader.EndOfStream);
+            Assert.AreEqual(0, reader.BytesLeft);
+            Assert.AreEqual(5, remainingData.Length);
+        }
+
+        [TestMethod]
+        public void Report_Correct_Length_For_Full_Buffer()
+        {
+            var buffer = new byte[] { 5, 6, 7, 8, 9 };
+
+            var reader = new MqttBufferReader();
+            reader.SetBuffer(buffer, 0, 5);
+
+            Assert.IsFalse(reader.EndOfStream);
+            Assert.AreEqual(5, reader.BytesLeft);
+        }
+
+        [TestMethod]
+        public void Report_Correct_Length_For_Partial_End_Buffer()
+        {
+            var buffer = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+
+            var reader = new MqttBufferReader();
+
+            // The used buffer contains more data than used!
+            reader.SetBuffer(buffer, 5, 5);
+
+            Assert.IsFalse(reader.EndOfStream);
+            Assert.AreEqual(5, reader.BytesLeft);
+        }
+
+        [TestMethod]
+        public void Report_Correct_Length_For_Partial_Start_Buffer()
+        {
+            var buffer = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+
+            var reader = new MqttBufferReader();
+
+            // The used buffer contains more data than used!
+            reader.SetBuffer(buffer, 0, 5);
+
+            Assert.IsFalse(reader.EndOfStream);
+            Assert.AreEqual(5, reader.BytesLeft);
+        }
+    }
+}

Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs

@@ -37,9 +37,9 @@ namespace MQTTnet.Tests.MQTTv5
                 await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
 
                 var c2 = await testEnvironment.ConnectClient(clientOptions);
-                c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
+                c2.Dispose(); // Dispose will not send a DISCONNECT packet first so the will message must be sent.
 
-                await Task.Delay(1000);
+                await LongTestDelay();
 
                 Assert.AreEqual(1, receivedMessagesCount);
             }

Source/MQTTnet.Tests/Protocol_Tests.cs

@@ -8,27 +8,48 @@ using MQTTnet.Formatter;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class Protocol_Tests
+    public sealed class Protocol_Tests
     {
         [TestMethod]
         public void Encode_Four_Byte_Integer()
         {
             var writer = new MqttBufferWriter(4, 4);
-            
+
             for (uint value = 0; value < 268435455; value++)
             {
                 writer.WriteVariableByteInteger(value);
-                
+
                 var buffer = writer.GetBuffer();
-                
+
                 var reader = new MqttBufferReader();
                 reader.SetBuffer(buffer, 0, writer.Length);
                 var checkValue = reader.ReadVariableByteInteger();
 
                 Assert.AreEqual(value, checkValue);
-                
+
+                writer.Reset(0);
+            }
+        }
+
+        [TestMethod]
+        public void Encode_Two_Byte_Integer()
+        {
+            var writer = new MqttBufferWriter(2, 2);
+
+            for (ushort value = 0; value < ushort.MaxValue; value++)
+            {
+                writer.WriteTwoByteInteger(value);
+
+                var buffer = writer.GetBuffer();
+
+                var reader = new MqttBufferReader();
+                reader.SetBuffer(buffer, 0, writer.Length);
+                var checkValue = reader.ReadTwoByteInteger();
+
+                Assert.AreEqual(value, checkValue);
+
                 writer.Reset(0);
             }
         }
     }
-}
+}

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -26,6 +26,8 @@ namespace MQTTnet.Tests.Server
                 var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
                 await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
                 dummyClient.Dispose();
+
+                await LongTestDelay();
                 
                 // Now inject messages which are appended to the queue of the client.
                 await server.InjectApplicationMessage("T", "0", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);

Source/MQTTnet/Formatter/MqttBufferReader.cs

@@ -13,15 +13,14 @@ namespace MQTTnet.Formatter
     public sealed class MqttBufferReader
     {
         byte[] _buffer = EmptyBuffer.Array;
-        int _initialPosition;
-        int _length;
-        int _position;
+        int _maxPosition;
+        int _offset;
 
-        public bool EndOfStream => _position == _length;
-        
-        public int Position => _position;
-        
-        public int BytesLeft => _length - _position;
+        public int BytesLeft => _maxPosition - Position;
+
+        public bool EndOfStream => BytesLeft == 0;
+
+        public int Position { get; private set; }
 
         public byte[] ReadBinaryData()
         {
@@ -35,8 +34,8 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(length);
 
             var result = new byte[length];
-            Array.Copy(_buffer, _position, result, 0, length);
-            _position += length;
+            Array.Copy(_buffer, Position + _offset, result, 0, length);
+            Position += length;
 
             return result;
         }
@@ -45,32 +44,42 @@ namespace MQTTnet.Formatter
         {
             ValidateReceiveBuffer(1);
 
-            return _buffer[_position++];
+            return _buffer[Position++];
         }
 
         public uint ReadFourByteInteger()
         {
             ValidateReceiveBuffer(4);
 
-            var byte0 = _buffer[_position++];
-            var byte1 = _buffer[_position++];
-            var byte2 = _buffer[_position++];
-            var byte3 = _buffer[_position++];
+#if NETCOREAPP3_0_OR_GREATER
+            var value = System.Runtime.InteropServices.MemoryMarshal.Read<uint>(_buffer.AsSpan().Slice(Position));
+            value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+#else
+            var byte0 = _buffer[_offset + Position];
+            var byte1 = _buffer[_offset + Position + 1];
+            var byte2 = _buffer[_offset + Position + 2];
+            var byte3 = _buffer[_offset + Position + 3];
 
-            return (uint)((byte0 << 24) | (byte1 << 16) | (byte2 << 8) | byte3);
+            var value = (uint)((byte0 << 24) | (byte1 << 16) | (byte2 << 8) | byte3);
+#endif
+
+            Position += 4;
+            return value;
         }
 
         public byte[] ReadRemainingData()
         {
-            var bufferLength = _length - _position;
+            var bufferLength = _maxPosition - Position - _offset;
 
             if (bufferLength == 0)
             {
                 return EmptyBuffer.Array;
             }
-            
+
             var buffer = new byte[bufferLength];
-            Array.Copy(_buffer, _position, buffer, 0, bufferLength);
+            Array.Copy(_buffer, Position + _offset, buffer, 0, bufferLength);
+
+            Position += bufferLength;
 
             return buffer;
         }
@@ -83,11 +92,17 @@ namespace MQTTnet.Formatter
             {
                 return string.Empty;
             }
-            
+
             ValidateReceiveBuffer(length);
 
-            var result = Encoding.UTF8.GetString(_buffer, _position, length);
-            _position += length;
+#if NETCOREAPP3_0_OR_GREATER
+            // AsSpan() version is slightly faster. Not much but at least a little bit.
+            var result = Encoding.UTF8.GetString(_buffer.AsSpan(_offset + Position, length));
+#else
+            var result = Encoding.UTF8.GetString(_buffer, _offset + Position, length);
+#endif            
+
+            Position += length;
             return result;
         }
 
@@ -95,10 +110,18 @@ namespace MQTTnet.Formatter
         {
             ValidateReceiveBuffer(2);
 
-            var msb = _buffer[_position++];
-            var lsb = _buffer[_position++];
+#if NETCOREAPP3_0_OR_GREATER
+            var value = System.Runtime.InteropServices.MemoryMarshal.Read<ushort>(_buffer.AsSpan().Slice(Position));
+            value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+#else
+            var msb = _buffer[_offset + Position];
+            var lsb = _buffer[_offset + Position + 1];
 
-            return (ushort)((msb << 8) | lsb);
+            var value = (ushort)((msb << 8) | lsb);
+#endif
+            
+            Position += 2;
+            return value;
         }
 
         public uint ReadVariableByteInteger()
@@ -125,23 +148,26 @@ namespace MQTTnet.Formatter
 
         public void Seek(int position)
         {
-            _position = _initialPosition + position;
+            Position = position;
         }
 
-        public void SetBuffer(byte[] buffer, int position, int length)
+        public void SetBuffer(byte[] buffer, int offset, int length)
         {
-            _buffer = buffer;
-            _initialPosition = position;
-            _position = position;
-            _length = length;
+            _buffer = buffer ?? throw new ArgumentNullException(nameof(buffer));
+            _offset = offset;
+
+            Position = 0;
+            _maxPosition = length;
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         void ValidateReceiveBuffer(int length)
         {
-            if (_length < _position + length)
+            var newPosition = Position + _offset + length;
+
+            if (_maxPosition < newPosition)
             {
-                throw new MqttProtocolViolationException($"Expected at least {Position + length} bytes but there are only {_length} bytes");
+                throw new MqttProtocolViolationException($"Expected at least {Position + length} bytes but there are only {_maxPosition} bytes");
             }
         }
     }

Source/MQTTnet/Formatter/MqttBufferWriter.cs

@@ -63,17 +63,30 @@ namespace MQTTnet.Formatter
             return _buffer;
         }
 
-        public static int GetLengthOfVariableInteger(uint value)
+        public static int GetVariableByteIntegerSize(uint value)
         {
-            var result = 0;
-            var x = value;
-            do
+            // From RFC: Table 2.4 Size of Remaining Length field
+        
+            // 0 (0x00) to 127 (0x7F)
+            if (value <= 127)
             {
-                x /= 128;
-                result++;
-            } while (x > 0);
+                return 1;
+            }
 
-            return result;
+            // 128 (0x80, 0x01) to 16 383 (0xFF, 0x7F)
+            if (value <= 16383)
+            {
+                return 2;
+            }
+
+            // 16 384 (0x80, 0x80, 0x01) to 2 097 151 (0xFF, 0xFF, 0x7F)
+            if (value <= 2097151)
+            {
+                return 3;
+            }
+
+            // 2 097 152 (0x80, 0x80, 0x80, 0x01) to 268 435 455 (0xFF, 0xFF, 0xFF, 0x7F)
+            return 4;
         }
 
         public void Reset(int length)
@@ -180,7 +193,7 @@ namespace MQTTnet.Formatter
 
                 _buffer[_position] = (byte)(writtenBytes >> 8);
                 _buffer[_position + 1] = (byte)writtenBytes;
-                
+
                 IncreasePosition(writtenBytes + 2);
             }
         }

Source/MQTTnet/Formatter/MqttPacketBuffer.cs

@@ -37,7 +37,7 @@ namespace MQTTnet.Formatter
 
         public byte[] ToArray()
         {
-            if (Packet.Count == 0)
+            if (Payload.Count == 0)
             {
                 return Packet.ToArray();
             }
@@ -51,7 +51,7 @@ namespace MQTTnet.Formatter
         
         public ArraySegment<byte> Join()
         {
-            if (Packet.Count == 0)
+            if (Payload.Count == 0)
             {
                 return Packet;
             }

Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs

@@ -108,7 +108,7 @@ namespace MQTTnet.Formatter.V3
                 remainingLength += (uint)publishPacket.Payload.Length;
             }
 
-            var remainingLengthSize = MqttBufferWriter.GetLengthOfVariableInteger(remainingLength);
+            var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength);
 
             var headerSize = FixedHeaderSize + remainingLengthSize;
             var headerOffset = 5 - headerSize;

Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs

@@ -12,6 +12,8 @@ namespace MQTTnet.Formatter.V5
 {
     public sealed class MqttV5PacketEncoder
     {
+        const int FixedHeaderSize = 1;
+        
         readonly MqttBufferWriter _bufferWriter;
         readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096));
 
@@ -40,9 +42,9 @@ namespace MQTTnet.Formatter.V5
                 remainingLength += (uint)publishPacket.Payload.Length;
             }
 
-            var remainingLengthSize = MqttBufferWriter.GetLengthOfVariableInteger(remainingLength);
+            var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength);
 
-            var headerSize = 1 + remainingLengthSize;
+            var headerSize = FixedHeaderSize + remainingLengthSize;
             var headerOffset = 5 - headerSize;
 
             // Position cursor on correct offset on beginning of array (has leading 0x0)