diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index b893e43e..6c770b9e 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/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; diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 0059f201..73965155 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/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 { diff --git a/Source/MQTTnet/Internal/TestMqttChannel.cs b/Source/MQTTnet/Internal/TestMqttChannel.cs index e48b8ede..08920e3c 100644 --- a/Source/MQTTnet/Internal/TestMqttChannel.cs +++ b/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; } = ""; public Task ConnectAsync(CancellationToken cancellationToken) { @@ -39,5 +35,9 @@ namespace MQTTnet.Internal { return _stream.WriteAsync(buffer, offset, count, cancellationToken); } + + public void Dispose() + { + } } } diff --git a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs b/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs index 2cf37dd4..f0e1b562 100644 --- a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs +++ b/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 ReadRemainingData() { - return new ArraySegment(_buffer, _offset, _buffer.Length - _offset); + return new ArraySegment(_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"); } } diff --git a/Source/MQTTnet/Serializer/MqttPacketReader.cs b/Source/MQTTnet/Serializer/MqttPacketReader.cs index 32a92855..ff8b54b3 100644 --- a/Source/MQTTnet/Serializer/MqttPacketReader.cs +++ b/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; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 8beb7b49..9b855805 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/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); } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index d8d23e29..c3a75f8e 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/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))); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index ea258f56..95bb67da 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/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 {