Extend MqttPacketReader constructor to allow more unit tests.

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -32,7 +32,8 @@ namespace MQTTnet.AspNetCore
             }
 
             var bodySlice = copy.Slice(0, bodyLength);
-            packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray(), 0)));
+            var buffer = bodySlice.GetArray();
+            packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length)));
             consumed = bodySlice.End;
             observed = bodySlice.End;
             return true;

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -201,7 +201,7 @@ namespace MQTTnet.Adapter
                     bodyOffset += readBytes;
                 } while (bodyOffset < body.Length);
 
-                return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0));
+                return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0, body.Length));
             }
             finally
             {

Source/MQTTnet/Internal/TestMqttChannel.cs

@@ -14,11 +14,7 @@ namespace MQTTnet.Internal
             _stream = stream;
         }
 
-        public void Dispose()
-        {
-        }
-
-        public string Endpoint { get; }
+        public string Endpoint { get; } = "<Test channel>";
 
         public Task ConnectAsync(CancellationToken cancellationToken)
         {
@@ -39,5 +35,9 @@ namespace MQTTnet.Internal
         {
             return _stream.WriteAsync(buffer, offset, count, cancellationToken);
         }
+
+        public void Dispose()
+        {
+        }
     }
 }

Source/MQTTnet/Serializer/MqttPacketBodyReader.cs

@@ -6,17 +6,20 @@ namespace MQTTnet.Serializer
     public class MqttPacketBodyReader
     {
         private readonly byte[] _buffer;
-        private int _offset;
+        private readonly int _length;
 
-        public MqttPacketBodyReader(byte[] buffer, int offset)
+        private int _offset;
+        
+        public MqttPacketBodyReader(byte[] buffer, int offset, int length)
         {
             _buffer = buffer;
             _offset = offset;
+            _length = length;
         }
 
-        public int Length => _buffer.Length - _offset;
+        public int Length => _length - _offset;
 
-        public bool EndOfStream => _offset == _buffer.Length;
+        public bool EndOfStream => _offset == _length;
 
         public byte ReadByte()
         {
@@ -26,7 +29,7 @@ namespace MQTTnet.Serializer
 
         public ArraySegment<byte> ReadRemainingData()
         {
-            return new ArraySegment<byte>(_buffer, _offset, _buffer.Length - _offset);
+            return new ArraySegment<byte>(_buffer, _offset, _length - _offset);
         }
 
         public ushort ReadUInt16()
@@ -53,9 +56,9 @@ namespace MQTTnet.Serializer
 
         private void ValidateReceiveBuffer(ushort length)
         {
-            if (_buffer.Length < _offset + length)
+            if (_length < _offset + length)
             {
-                throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_buffer.Length} bytes");
+                throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_length} bytes");
             }
         }
 

Source/MQTTnet/Serializer/MqttPacketReader.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Threading;
+using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Exceptions;

Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -58,7 +58,7 @@ namespace MQTTnet.Benchmarks
 
                 var receivedPacket = new ReceivedMqttPacket(
                     header.Flags,
-                    new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength));
+                    new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength, _serializedPacket.Array.Length));
 
                 _serializer.Deserialize(receivedPacket);
             }

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Linq;
 using System.Text;
 using System.Threading;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -149,6 +150,53 @@ namespace MQTTnet.Core.Tests
             DeserializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310);
         }
 
+        [TestMethod]
+        public void Serialize_LargePacket()
+        {
+            var serializer = new MqttPacketSerializer { ProtocolVersion = MqttProtocolVersion.V311 };
+
+            const int payloadLength = 80000;
+
+            var payload = new byte[payloadLength];
+
+            var value = 0;
+            for (var i = 0; i < payloadLength; i++)
+            {
+                if (value > 255)
+                {
+                    value = 0;
+                }
+
+                payload[i] = (byte)value;
+            }
+
+            var publishPacket = new MqttPublishPacket
+            {
+                Topic = "abcdefghijklmnopqrstuvwxyz0123456789",
+                Payload = payload
+            };
+
+            var buffer = serializer.Serialize(publishPacket);
+            var testChannel = new TestMqttChannel(new MemoryStream(buffer.Array, buffer.Offset, buffer.Count));
+
+            var header = MqttPacketReader.ReadFixedHeaderAsync(
+                testChannel, 
+                new byte[2],
+                new byte[1], 
+                CancellationToken.None).GetAwaiter().GetResult();
+
+            var eof = buffer.Offset + buffer.Count;
+
+            var receivedPacket = new ReceivedMqttPacket(
+                header.Flags,
+                new MqttPacketBodyReader(buffer.Array, eof - header.RemainingLength, buffer.Count + buffer.Offset));
+
+            var packet = (MqttPublishPacket)serializer.Deserialize(receivedPacket);
+
+            Assert.AreEqual(publishPacket.Topic, packet.Topic);
+            Assert.IsTrue(publishPacket.Payload.SequenceEqual(packet.Payload));
+        }
+
         [TestMethod]
         public void SerializeV311_MqttDisconnectPacket()
         {
@@ -463,7 +511,7 @@ namespace MQTTnet.Core.Tests
         {
             var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
             var data = serializer.Serialize(packet);
-            
+
             Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data)));
         }
 
@@ -479,10 +527,10 @@ namespace MQTTnet.Core.Tests
                 var fixedHeader = new byte[2];
                 var singleByteBuffer = new byte[1];
                 var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult();
-                
+
                 using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength))
                 {
-                    var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0)));
+                    var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length)));
                     var buffer2 = serializer.Serialize(deserializedPacket);
 
                     Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
@@ -507,7 +555,7 @@ namespace MQTTnet.Core.Tests
 
                 using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength))
                 {
-                    return (T)serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0)));
+                    return (T)serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length)));
                 }
             }
         }

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -375,6 +375,9 @@ namespace MQTTnet.Core.Tests
                 var c1 = await serverAdapter.ConnectTestClient("c1");
 
                 await c1.PublishAndWaitForAsync(s, new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+
+                await Task.Delay(250);
+
                 await c1.DisconnectAsync();
             }
             finally
@@ -396,7 +399,7 @@ namespace MQTTnet.Core.Tests
                 c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
                 await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained").Build());
 
-                await Task.Delay(500);
+                await Task.Delay(250);
             }
             finally
             {