From b477ddbe6064e58a61581d85b19650e355e7ef4f Mon Sep 17 00:00:00 2001 From: christian <6939810+chkr1011@users.noreply.github.com> Date: Wed, 22 Oct 2025 11:35:56 +0200 Subject: [PATCH] Start using readonly memory --- Source/MQTTnet.Tests/Server/Load_Tests.cs | 5 +- .../MqttClientConnectResultFactory.cs | 2 +- .../MqttApplicationMessageFactory.cs | 2 +- Source/MQTTnet/Formatter/MqttBufferReader.cs | 2 +- Source/MQTTnet/Formatter/MqttBufferWriter.cs | 126 ++++++++++++------ .../Formatter/MqttConnectPacketFactory.cs | 3 +- .../Formatter/V3/MqttV3PacketFormatter.cs | 16 +-- .../Formatter/V5/MqttV5PacketDecoder.cs | 5 +- .../Formatter/V5/MqttV5PacketEncoder.cs | 6 +- .../Formatter/V5/MqttV5PropertiesWriter.cs | 8 +- .../MqttEnhancedAuthenticationEventArgs.cs | 4 +- Source/MQTTnet/Packets/MqttAuthPacket.cs | 2 +- Source/MQTTnet/Packets/MqttConnAckPacket.cs | 2 +- Source/MQTTnet/Packets/MqttConnectPacket.cs | 11 +- Source/MQTTnet/Packets/MqttPublishPacket.cs | 4 +- 15 files changed, 124 insertions(+), 74 deletions(-) diff --git a/Source/MQTTnet.Tests/Server/Load_Tests.cs b/Source/MQTTnet.Tests/Server/Load_Tests.cs index 7573e50e..75490860 100644 --- a/Source/MQTTnet.Tests/Server/Load_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Load_Tests.cs @@ -1,5 +1,6 @@ #if DEBUG +using System.Globalization; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -52,7 +53,7 @@ public sealed class Load_Tests : BaseTestClass for (var j = 0; j < 1000; j++) { - publishPacket.Topic = j.ToString(); + publishPacket.Topic = j.ToString(CultureInfo.InvariantCulture); await client.SendAsync(publishPacket, CancellationToken.None); } @@ -139,7 +140,7 @@ public sealed class Load_Tests : BaseTestClass for (var j = 0; j < 1000; j++) { - var message = applicationMessageBuilder.WithTopic(j.ToString()).Build(); + var message = applicationMessageBuilder.WithTopic(j.ToString(CultureInfo.InvariantCulture)).Build(); await client.PublishAsync(message); } diff --git a/Source/MQTTnet/Connecting/MqttClientConnectResultFactory.cs b/Source/MQTTnet/Connecting/MqttClientConnectResultFactory.cs index 18a23c46..e4b64922 100644 --- a/Source/MQTTnet/Connecting/MqttClientConnectResultFactory.cs +++ b/Source/MQTTnet/Connecting/MqttClientConnectResultFactory.cs @@ -63,7 +63,7 @@ public static class MqttClientConnectResultFactory RetainAvailable = connAckPacket.RetainAvailable, AssignedClientIdentifier = connAckPacket.AssignedClientIdentifier, AuthenticationMethod = connAckPacket.AuthenticationMethod, - AuthenticationData = connAckPacket.AuthenticationData, + AuthenticationData = connAckPacket.AuthenticationData.ToArray(), MaximumPacketSize = connAckPacket.MaximumPacketSize, ReasonString = connAckPacket.ReasonString, ReceiveMaximum = connAckPacket.ReceiveMaximum, diff --git a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs index 4c18c2f2..d4dd6888 100644 --- a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs +++ b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs @@ -21,7 +21,7 @@ public static class MqttApplicationMessageFactory Dup = publishPacket.Dup, ResponseTopic = publishPacket.ResponseTopic, ContentType = publishPacket.ContentType, - CorrelationData = publishPacket.CorrelationData, + CorrelationData = publishPacket.CorrelationData.ToArray(), MessageExpiryInterval = publishPacket.MessageExpiryInterval, SubscriptionIdentifiers = publishPacket.SubscriptionIdentifiers, TopicAlias = publishPacket.TopicAlias, diff --git a/Source/MQTTnet/Formatter/MqttBufferReader.cs b/Source/MQTTnet/Formatter/MqttBufferReader.cs index 6c4bbea8..6211b86b 100644 --- a/Source/MQTTnet/Formatter/MqttBufferReader.cs +++ b/Source/MQTTnet/Formatter/MqttBufferReader.cs @@ -58,7 +58,7 @@ public sealed class MqttBufferReader return value; } - public byte[] ReadRemainingData() + public ReadOnlyMemory ReadRemainingData() { var bufferLength = BytesLeft; if (bufferLength == 0) diff --git a/Source/MQTTnet/Formatter/MqttBufferWriter.cs b/Source/MQTTnet/Formatter/MqttBufferWriter.cs index c31f56f9..e5006f2b 100644 --- a/Source/MQTTnet/Formatter/MqttBufferWriter.cs +++ b/Source/MQTTnet/Formatter/MqttBufferWriter.cs @@ -2,6 +2,8 @@ // 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.Buffers; +using System.Buffers.Binary; using System.Runtime.CompilerServices; using System.Text; using MQTTnet.Exceptions; @@ -108,31 +110,47 @@ public sealed class MqttBufferWriter WriteBinary(propertyWriter._buffer, 0, propertyWriter.Length); } - public void WriteBinary(byte[] value) + public void WriteBinary(ReadOnlySequence value) { - if (value == null || value.Length == 0) + if (value.IsEmpty) { - EnsureAdditionalCapacity(2); - - _buffer[_position] = 0; - _buffer[_position + 1] = 0; - - IncreasePosition(2); + WriteEmptyBinary(); + return; } - else + + EnsureAdditionalCapacity(value.Length + 2); + + _buffer[_position] = (byte)(value.Length >> 8); + _buffer[_position + 1] = (byte)value.Length; + + Advance(2); + + foreach (var segment in value) { - var valueLength = value.Length; - - EnsureAdditionalCapacity(valueLength + 2); - - _buffer[_position] = (byte)(valueLength >> 8); - _buffer[_position + 1] = (byte)valueLength; - - MqttMemoryHelper.Copy(value, 0, _buffer, _position + 2, valueLength); - IncreasePosition(valueLength + 2); + MqttMemoryHelper.Copy(segment, 0, _buffer, _position + 2, segment.Length); + Advance(segment.Length); } } + public void WriteBinary(ReadOnlyMemory value) + { + if (value.Length == 0) + { + WriteEmptyBinary(); + return; + } + + EnsureAdditionalCapacity(value.Length + 2); + + _buffer[_position] = (byte)(value.Length >> 8); + _buffer[_position + 1] = (byte)value.Length; + + Advance(2); + + MqttMemoryHelper.Copy(value, 0, _buffer, _position + 2, (int)value.Length); + Advance(value.Length); + } + public void WriteBinary(byte[] buffer, int offset, int count) { ArgumentNullException.ThrowIfNull(buffer); @@ -145,15 +163,7 @@ public sealed class MqttBufferWriter EnsureAdditionalCapacity(count); MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count); - IncreasePosition(count); - } - - public void WriteByte(byte @byte) - { - EnsureAdditionalCapacity(1); - - _buffer[_position] = @byte; - IncreasePosition(1); + Advance(count); } public void WriteString(string value) @@ -165,7 +175,7 @@ public sealed class MqttBufferWriter _buffer[_position] = 0; _buffer[_position + 1] = 0; - IncreasePosition(2); + Advance(2); } else { @@ -189,26 +199,52 @@ public sealed class MqttBufferWriter _buffer[_position] = (byte)(writtenBytes >> 8); _buffer[_position + 1] = (byte)writtenBytes; - IncreasePosition(writtenBytes + 2); + Advance(writtenBytes + 2); } } + public void WriteByte(byte @byte) + { + const int size = sizeof(byte); + var span = GetSpan(size); + + span[0] = @byte; + Advance(size); + } + public void WriteTwoByteInteger(ushort value) { - EnsureAdditionalCapacity(2); + const int size = sizeof(ushort); + var span = GetSpan(size); - _buffer[_position] = (byte)(value >> 8); - IncreasePosition(1); - _buffer[_position] = (byte)value; - IncreasePosition(1); + BinaryPrimitives.WriteUInt16BigEndian(span, value); + + Advance(size); } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + Span GetSpan(int size) + { + var freeSpace = _buffer.Length - _position; + if (freeSpace < size) + { + EnsureCapacity(_buffer.Length + size - freeSpace); + } + + return _buffer.AsSpan(_position, size); + } + + + + + + public void WriteVariableByteInteger(uint value) { if (value == 0) { _buffer[_position] = 0; - IncreasePosition(1); + Advance(1); return; } @@ -216,7 +252,7 @@ public sealed class MqttBufferWriter if (value <= 127) { _buffer[_position] = (byte)value; - IncreasePosition(1); + Advance(1); return; } @@ -238,11 +274,11 @@ public sealed class MqttBufferWriter size++; } while (x > 0); - IncreasePosition(size); + Advance(size); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - void EnsureAdditionalCapacity(int additionalCapacity) + void EnsureAdditionalCapacity(long additionalCapacity) { var bufferLength = _buffer.Length; @@ -256,7 +292,7 @@ public sealed class MqttBufferWriter } [MethodImpl(MethodImplOptions.AggressiveInlining)] - void EnsureCapacity(int capacity) + void EnsureCapacity(long capacity) { var newBufferLength = _buffer.Length; @@ -275,7 +311,7 @@ public sealed class MqttBufferWriter } [MethodImpl(MethodImplOptions.AggressiveInlining)] - void IncreasePosition(int length) + void Advance(int length) { _position += length; @@ -286,4 +322,14 @@ public sealed class MqttBufferWriter Length = _position; } } + + void WriteEmptyBinary() + { + EnsureAdditionalCapacity(2); + + _buffer[_position] = 0; + _buffer[_position + 1] = 0; + + Advance(2); + } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs index 2bebde5d..6513c7af 100644 --- a/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs @@ -2,6 +2,7 @@ // 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.Buffers; using MQTTnet.Packets; namespace MQTTnet.Formatter; @@ -40,7 +41,7 @@ public static class MqttConnectPacketFactory connectPacket.WillFlag = true; connectPacket.WillTopic = clientOptions.WillTopic; connectPacket.WillQoS = clientOptions.WillQualityOfServiceLevel; - connectPacket.WillMessage = clientOptions.WillPayload; + connectPacket.WillMessage = new ReadOnlySequence(clientOptions.WillPayload); connectPacket.WillRetain = clientOptions.WillRetain; connectPacket.WillDelayInterval = clientOptions.WillDelayInterval; connectPacket.WillContentType = clientOptions.WillContentType; diff --git a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs index 51c22f13..3d9c32cc 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs @@ -201,7 +201,7 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter packet.WillRetain = willRetain; packet.WillTopic = _bufferReader.ReadString(); - packet.WillMessage = _bufferReader.ReadBinaryData(); + packet.WillMessage = new ReadOnlySequence(_bufferReader.ReadBinaryData()); } if (usernameFlag) @@ -271,7 +271,7 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter if (!_bufferReader.EndOfStream) { - packet.PayloadSegment = new ArraySegment(_bufferReader.ReadRemainingData()); + packet.Payload = new ReadOnlySequence(_bufferReader.ReadRemainingData()); } return packet; @@ -430,12 +430,12 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter } } - if (packet.Password != null && packet.Username == null) + if (packet.Password.Length > 0 && packet.Username == null) { throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); } - if (packet.Password != null) + if (packet.Password.Length > 0) { connectFlags |= 0x40; } @@ -460,7 +460,7 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter bufferWriter.WriteString(packet.Username); } - if (packet.Password != null) + if (packet.Password.Length > 0) { bufferWriter.WriteBinary(packet.Password); } @@ -501,12 +501,12 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter } } - if (packet.Password != null && packet.Username == null) + if (packet.Password.Length > 0 && packet.Username == null) { throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); } - if (packet.Password != null) + if (packet.Password.Length > 0) { connectFlags |= 0x40; } @@ -531,7 +531,7 @@ public sealed class MqttV3PacketFormatter : IMqttPacketFormatter bufferWriter.WriteString(packet.Username); } - if (packet.Password != null) + if (packet.Password.Length > 0) { bufferWriter.WriteBinary(packet.Password); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs index 51d09901..a9e58ffb 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs @@ -2,6 +2,7 @@ // 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.Buffers; using MQTTnet.Adapter; using MQTTnet.Exceptions; using MQTTnet.Packets; @@ -314,7 +315,7 @@ public sealed class MqttV5PacketDecoder } packet.WillTopic = _bufferReader.ReadString(); - packet.WillMessage = _bufferReader.ReadBinaryData(); + packet.WillMessage = new ReadOnlySequence(_bufferReader.ReadBinaryData()); packet.WillUserProperties = willPropertiesReader.CollectedUserProperties; } @@ -521,7 +522,7 @@ public sealed class MqttV5PacketDecoder if (!_bufferReader.EndOfStream) { - packet.PayloadSegment = new ArraySegment(_bufferReader.ReadRemainingData()); + packet.Payload = new ReadOnlySequence(_bufferReader.ReadBinaryData()); } return packet; diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs index 4aea8b01..c9b422a1 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs @@ -138,12 +138,12 @@ public sealed class MqttV5PacketEncoder(MqttBufferWriter bufferWriter) } } - if (packet.Password != null && packet.Username == null) + if (packet.Password.Length > 0 && packet.Username == null) { throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); } - if (packet.Password != null) + if (packet.Password.Length > 0) { connectFlags |= 0x40; } @@ -193,7 +193,7 @@ public sealed class MqttV5PacketEncoder(MqttBufferWriter bufferWriter) _bufferWriter.WriteString(packet.Username); } - if (packet.Password != null) + if (packet.Password.Length > 0) { _bufferWriter.WriteBinary(packet.Password); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs index 60c02d5f..c8473069 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs @@ -29,7 +29,7 @@ public sealed class MqttV5PropertiesWriter Write(MqttPropertyId.AssignedClientIdentifier, value); } - public void WriteAuthenticationData(byte[] value) + public void WriteAuthenticationData(ReadOnlyMemory value) { Write(MqttPropertyId.AuthenticationData, value); } @@ -44,7 +44,7 @@ public sealed class MqttV5PropertiesWriter Write(MqttPropertyId.ContentType, value); } - public void WriteCorrelationData(byte[] value) + public void WriteCorrelationData(ReadOnlyMemory value) { Write(MqttPropertyId.CorrelationData, value); } @@ -315,9 +315,9 @@ public sealed class MqttV5PropertiesWriter _bufferWriter.WriteString(value); } - void Write(MqttPropertyId id, byte[] value) + void Write(MqttPropertyId id, ReadOnlyMemory value) { - if (value == null) + if (value.Length == 0) { return; } diff --git a/Source/MQTTnet/Options/MqttEnhancedAuthenticationEventArgs.cs b/Source/MQTTnet/Options/MqttEnhancedAuthenticationEventArgs.cs index 909c4e35..d6a3a25d 100644 --- a/Source/MQTTnet/Options/MqttEnhancedAuthenticationEventArgs.cs +++ b/Source/MQTTnet/Options/MqttEnhancedAuthenticationEventArgs.cs @@ -26,7 +26,7 @@ public class MqttEnhancedAuthenticationEventArgs : EventArgs /// Gets the authentication data. /// Hint: MQTT 5 feature only. /// - public byte[] AuthenticationData => _initialAuthPacket.AuthenticationData; + public byte[] AuthenticationData => _initialAuthPacket.AuthenticationData.ToArray(); /// /// Gets the authentication method. @@ -67,7 +67,7 @@ public class MqttEnhancedAuthenticationEventArgs : EventArgs { return new ReceiveMqttEnhancedAuthenticationDataResult { - AuthenticationData = authPacket.AuthenticationData, + AuthenticationData = authPacket.AuthenticationData.ToArray(), AuthenticationMethod = authPacket.AuthenticationMethod, ReasonString = authPacket.ReasonString, ReasonCode = authPacket.ReasonCode, diff --git a/Source/MQTTnet/Packets/MqttAuthPacket.cs b/Source/MQTTnet/Packets/MqttAuthPacket.cs index 02d2a211..79a541d4 100644 --- a/Source/MQTTnet/Packets/MqttAuthPacket.cs +++ b/Source/MQTTnet/Packets/MqttAuthPacket.cs @@ -9,7 +9,7 @@ namespace MQTTnet.Packets; /// Added in MQTTv5.0.0. public sealed class MqttAuthPacket : MqttPacket { - public byte[] AuthenticationData { get; set; } + public ReadOnlyMemory AuthenticationData { get; set; } public string AuthenticationMethod { get; set; } diff --git a/Source/MQTTnet/Packets/MqttConnAckPacket.cs b/Source/MQTTnet/Packets/MqttConnAckPacket.cs index dc206d05..c07791e9 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacket.cs @@ -13,7 +13,7 @@ public sealed class MqttConnAckPacket : MqttPacket /// public string AssignedClientIdentifier { get; set; } - public byte[] AuthenticationData { get; set; } + public ReadOnlyMemory AuthenticationData { get; set; } public string AuthenticationMethod { get; set; } diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index 1f3b0091..ce40b6ed 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -2,13 +2,14 @@ // 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.Buffers; using MQTTnet.Protocol; namespace MQTTnet.Packets; public sealed class MqttConnectPacket : MqttPacket { - public byte[] AuthenticationData { get; set; } + public ReadOnlyMemory AuthenticationData { get; set; } public string AuthenticationMethod { get; set; } @@ -19,13 +20,13 @@ public sealed class MqttConnectPacket : MqttPacket public string ClientId { get; set; } - public byte[] WillCorrelationData { get; set; } + public ReadOnlyMemory WillCorrelationData { get; set; } public ushort KeepAlivePeriod { get; set; } public uint MaximumPacketSize { get; set; } - public byte[] Password { get; set; } + public ReadOnlyMemory Password { get; set; } public ushort ReceiveMaximum { get; set; } @@ -49,7 +50,7 @@ public sealed class MqttConnectPacket : MqttPacket public bool WillFlag { get; set; } - public byte[] WillMessage { get; set; } + public ReadOnlySequence WillMessage { get; set; } public uint WillMessageExpiryInterval { get; set; } @@ -69,7 +70,7 @@ public sealed class MqttConnectPacket : MqttPacket { var passwordText = string.Empty; - if (Password != null) + if (Password.Length > 0) { passwordText = "****"; } diff --git a/Source/MQTTnet/Packets/MqttPublishPacket.cs b/Source/MQTTnet/Packets/MqttPublishPacket.cs index 7d47fad7..5875efab 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacket.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacket.cs @@ -11,7 +11,7 @@ public sealed class MqttPublishPacket : MqttPacketWithIdentifier { public string ContentType { get; set; } - public byte[] CorrelationData { get; set; } + public ReadOnlyMemory CorrelationData { get; set; } public bool Dup { get; set; } @@ -19,7 +19,7 @@ public sealed class MqttPublishPacket : MqttPacketWithIdentifier public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified; - public ArraySegment PayloadSegment { set => Payload = new ReadOnlySequence(value); } + //public ArraySegment PayloadSegment { set => Payload = new ReadOnlySequence(value); } public ReadOnlySequence Payload { get; set; }