From d0584413ca731563523edec9e791b467bffbc5ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=80=81=E4=B9=9D?= <366193849@qq.com> Date: Thu, 23 Feb 2023 03:53:55 +0800 Subject: [PATCH] Add PayloadSegmentable feature (#1585) --- .github/workflows/ReleaseNotes.md | 1 + .../MqttConnectionContextTest.cs | 2 +- .../ReaderExtensionsTest.cs | 2 +- .../MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 5 +- Source/MQTTnet.TestApp/ClientTest.cs | 11 +- Source/MQTTnet.TestApp/PerformanceTest.cs | 8 +- Source/MQTTnet.TestApp/ServerTest.cs | 15 ++- .../MqttClient/MqttClient_Connection_Tests.cs | 2 + ...MqttPacketSerialization_V3_Binary_Tests.cs | 13 ++- .../MqttPacketSerialization_V3_Tests.cs | 6 +- .../MqttPacketSerialization_V5_Tests.cs | 6 +- .../MqttApplicationMessageTest.cs | 48 +++++++++ Source/MQTTnet.Tests/Server/Security_Tests.cs | 2 +- .../MqttApplicationMessageFactory.cs | 2 +- .../Formatter/MqttPublishPacketFactory.cs | 12 ++- .../Formatter/V3/MqttV3PacketFormatter.cs | 30 +++--- .../Formatter/V5/MqttV5PacketDecoder.cs | 2 +- .../Formatter/V5/MqttV5PacketEncoder.cs | 19 ++-- Source/MQTTnet/MqttApplicationMessage.cs | 56 +++++++++- .../MQTTnet/MqttApplicationMessageBuilder.cs | 101 ++++++++++-------- .../MqttApplicationMessageExtensions.cs | 14 +-- Source/MQTTnet/Packets/MqttPublishPacket.cs | 7 +- .../Internal/MqttRetainedMessagesManager.cs | 16 ++- Source/MQTTnet/Server/MqttServerExtensions.cs | 2 +- 24 files changed, 260 insertions(+), 122 deletions(-) create mode 100644 Source/MQTTnet.Tests/MqttApplicationMessageTest.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index baf6356e..d534f1b7 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,4 +1,5 @@ * [Core] Added net7.0 builds (#1635, thanks to @YAJeff). +* [Core] Added support for payload segments (#1585, thanks to @xljiulang) * [Client] Added support for passing MQTT v5 options (User properties etc.) for disconnects. * [Client] An internal exception is no longer caught silently when calling _DisconnectAsync_ to indicate that the disconnect is not clean (BREAKING CHANGE). * [Client] MQTTv5 features are now checked and an exception is thrown if they are used when using protocol version 3.1.1 and lower. These checks can be disabled in client options. (BREAKING CHANGE!). diff --git a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index 865a49ce..e24074d4 100644 --- a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -103,7 +103,7 @@ namespace MQTTnet.AspNetCore.Tests connection.Transport = pipe; var ctx = new MqttConnectionContext(serializer, connection); - await ctx.SendPacketAsync(new MqttPublishPacket { Payload = new byte[20_000] }, CancellationToken.None).ConfigureAwait(false); + await ctx.SendPacketAsync(new MqttPublishPacket { PayloadSegment = new ArraySegment(new byte[20_000]) }, CancellationToken.None).ConfigureAwait(false); var readResult = await pipe.Send.Reader.ReadAsync(); Assert.IsTrue(readResult.Buffer.Length > 20000); diff --git a/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index d711067b..5c341084 100644 --- a/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -19,7 +19,7 @@ namespace MQTTnet.AspNetCore.Tests { var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535)); - var buffer = serializer.Encode(new MqttPublishPacket {Topic = "a", Payload = new byte[5]}).Join(); + var buffer = serializer.Encode(new MqttPublishPacket {Topic = "a", PayloadSegment = new byte[5]}).Join(); var sequence = new ReadOnlySequence(buffer.Array, buffer.Offset, buffer.Count); diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 864db187..aab296c9 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -4,6 +4,7 @@ using System; using System.Collections.Concurrent; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Client; @@ -119,7 +120,6 @@ namespace MQTTnet.Extensions.Rpc finally { _waitingCalls.TryRemove(responseTopic, out _); - await _mqttClient.UnsubscribeAsync(responseTopic, CancellationToken.None).ConfigureAwait(false); } } @@ -131,7 +131,8 @@ namespace MQTTnet.Extensions.Rpc return CompletedTask.Instance; } - awaitable.TrySetResult(eventArgs.ApplicationMessage.Payload); + var payloadBuffer = eventArgs.ApplicationMessage.PayloadSegment.ToArray(); + awaitable.TrySetResult(payloadBuffer); // Set this message to handled to that other code can avoid execution etc. eventArgs.IsHandled = true; diff --git a/Source/MQTTnet.TestApp/ClientTest.cs b/Source/MQTTnet.TestApp/ClientTest.cs index 9f233256..274515fa 100644 --- a/Source/MQTTnet.TestApp/ClientTest.cs +++ b/Source/MQTTnet.TestApp/ClientTest.cs @@ -33,9 +33,18 @@ namespace MQTTnet.TestApp client.ApplicationMessageReceivedAsync += e => { + var payloadText = string.Empty; + if (e.ApplicationMessage.PayloadSegment.Count > 0) + { + payloadText = Encoding.UTF8.GetString( + e.ApplicationMessage.PayloadSegment.Array, + e.ApplicationMessage.PayloadSegment.Offset, + e.ApplicationMessage.PayloadSegment.Count); + } + Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); - Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}"); + Console.WriteLine($"+ Payload = {payloadText}"); Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}"); Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); Console.WriteLine(); diff --git a/Source/MQTTnet.TestApp/PerformanceTest.cs b/Source/MQTTnet.TestApp/PerformanceTest.cs index 03b01d05..c1bd991e 100644 --- a/Source/MQTTnet.TestApp/PerformanceTest.cs +++ b/Source/MQTTnet.TestApp/PerformanceTest.cs @@ -196,7 +196,7 @@ namespace MQTTnet.TestApp return new MqttApplicationMessage { Topic = "A/B/C", - Payload = Encoding.UTF8.GetBytes(Payload), + PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes(Payload)), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }; } @@ -229,7 +229,7 @@ namespace MQTTnet.TestApp var message = new MqttApplicationMessage { Topic = "A/B/C", - Payload = Encoding.UTF8.GetBytes("Hello World"), + PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes("Hello World")), QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }; @@ -280,7 +280,7 @@ namespace MQTTnet.TestApp var message = new MqttApplicationMessage { Topic = "A/B/C", - Payload = Encoding.UTF8.GetBytes("Hello World"), + PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes("Hello World")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }; @@ -331,7 +331,7 @@ namespace MQTTnet.TestApp var message = new MqttApplicationMessage { Topic = "A/B/C", - Payload = Encoding.UTF8.GetBytes("Hello World"), + PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes("Hello World")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }; diff --git a/Source/MQTTnet.TestApp/ServerTest.cs b/Source/MQTTnet.TestApp/ServerTest.cs index a0925def..bd5961bc 100644 --- a/Source/MQTTnet.TestApp/ServerTest.cs +++ b/Source/MQTTnet.TestApp/ServerTest.cs @@ -5,6 +5,7 @@ using System; using System.Collections.Generic; using System.IO; +using System.Linq; using System.Text; using System.Threading.Tasks; using MQTTnet.Diagnostics; @@ -100,7 +101,7 @@ namespace MQTTnet.TestApp { // Replace the payload with the timestamp. But also extending a JSON // based payload with the timestamp is a suitable use case. - e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O")); + e.ApplicationMessage.PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes(DateTime.Now.ToString("O"))); } if (e.ApplicationMessage.Topic == "not_allowed_topic") @@ -143,10 +144,16 @@ namespace MQTTnet.TestApp mqttServer.InterceptingPublishAsync += e => { - MqttNetConsoleLogger.PrintToConsole( - $"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0])}'", - ConsoleColor.Magenta); + var payloadText = string.Empty; + if (e.ApplicationMessage.PayloadSegment.Count > 0) + { + payloadText = Encoding.UTF8.GetString( + e.ApplicationMessage.PayloadSegment.Array, + e.ApplicationMessage.PayloadSegment.Offset, + e.ApplicationMessage.PayloadSegment.Count); + } + MqttNetConsoleLogger.PrintToConsole($"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{payloadText}'", ConsoleColor.Magenta); return CompletedTask.Instance; }; diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs index 84502e2e..87ecf0b4 100644 --- a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs @@ -150,6 +150,8 @@ namespace MQTTnet.Tests.Clients.MqttClient // Perform a clean disconnect. await client.DisconnectAsync(disconnectOptions); + + await LongTestDelay(); Assert.IsNotNull(eventArgs); Assert.IsNotNull(eventArgs.UserProperties); diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs index 6c32b24a..4cda6c0a 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs @@ -111,7 +111,7 @@ namespace MQTTnet.Tests.Formatter PacketIdentifier = 123, Dup = true, Retain = true, - Payload = Encoding.ASCII.GetBytes("HELLO"), + PayloadSegment = new ArraySegment(Encoding.ASCII.GetBytes("HELLO")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Topic = "A/B/C" }; @@ -318,10 +318,9 @@ namespace MQTTnet.Tests.Formatter var publishPacket = new MqttPublishPacket { Topic = "abcdefghijklmnopqrstuvwxyz0123456789", - Payload = payload + PayloadSegment = new ArraySegment(payload) }; - var serializationHelper = new MqttPacketSerializationHelper(); var buffer = serializationHelper.Encode(publishPacket); @@ -329,17 +328,17 @@ namespace MQTTnet.Tests.Formatter Assert.IsNotNull(publishPacketCopy); Assert.AreEqual(publishPacket.Topic, publishPacketCopy.Topic); - CollectionAssert.AreEqual(publishPacket.Payload, publishPacketCopy.Payload); + CollectionAssert.AreEqual(publishPacket.PayloadSegment.ToArray(), publishPacketCopy.PayloadSegment.ToArray()); // Now modify the payload and test again. - publishPacket.Payload = Encoding.UTF8.GetBytes("MQTT"); + publishPacket.PayloadSegment = new ArraySegment(Encoding.UTF8.GetBytes("MQTT")); buffer = serializationHelper.Encode(publishPacket); var publishPacketCopy2 = serializationHelper.Decode(buffer) as MqttPublishPacket; Assert.IsNotNull(publishPacketCopy2); Assert.AreEqual(publishPacket.Topic, publishPacketCopy2.Topic); - CollectionAssert.AreEqual(publishPacket.Payload, publishPacketCopy2.Payload); + CollectionAssert.AreEqual(publishPacket.PayloadSegment.ToArray(), publishPacketCopy2.PayloadSegment.ToArray()); } [TestMethod] @@ -463,7 +462,7 @@ namespace MQTTnet.Tests.Formatter PacketIdentifier = 123, Dup = true, Retain = true, - Payload = Encoding.ASCII.GetBytes("HELLO"), + PayloadSegment = new ArraySegment(Encoding.ASCII.GetBytes("HELLO")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Topic = "A/B/C" }; diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs index e1135529..53ea016d 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs @@ -2,7 +2,9 @@ // 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.Collections.Generic; +using System.Linq; using System.Text; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Exceptions; @@ -296,7 +298,7 @@ namespace MQTTnet.Tests.Formatter PacketIdentifier = 123, Dup = true, Retain = true, - Payload = Encoding.ASCII.GetBytes("Payload"), + PayloadSegment = new ArraySegment(Encoding.ASCII.GetBytes("Payload")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Topic = "Topic", ResponseTopic = "/Response", @@ -320,7 +322,7 @@ namespace MQTTnet.Tests.Formatter Assert.AreEqual(publishPacket.PacketIdentifier, deserialized.PacketIdentifier); Assert.AreEqual(publishPacket.Dup, deserialized.Dup); Assert.AreEqual(publishPacket.Retain, deserialized.Retain); - CollectionAssert.AreEqual(publishPacket.Payload, deserialized.Payload); + CollectionAssert.AreEqual(publishPacket.PayloadSegment.ToArray(), deserialized.PayloadSegment.ToArray()); Assert.AreEqual(publishPacket.QualityOfServiceLevel, deserialized.QualityOfServiceLevel); Assert.AreEqual(publishPacket.Topic, deserialized.Topic); Assert.AreEqual(null, deserialized.ResponseTopic); // Not supported in v3.1.1. diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs index fd7a70e2..1f963437 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs @@ -2,7 +2,9 @@ // 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.Collections.Generic; +using System.Linq; using System.Text; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Formatter; @@ -257,7 +259,7 @@ namespace MQTTnet.Tests.Formatter PacketIdentifier = 123, Dup = true, Retain = true, - Payload = Encoding.ASCII.GetBytes("Payload"), + PayloadSegment = new ArraySegment(Encoding.ASCII.GetBytes("Payload")), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Topic = "Topic", ResponseTopic = "/Response", @@ -281,7 +283,7 @@ namespace MQTTnet.Tests.Formatter Assert.AreEqual(publishPacket.PacketIdentifier, deserialized.PacketIdentifier); Assert.AreEqual(publishPacket.Dup, deserialized.Dup); Assert.AreEqual(publishPacket.Retain, deserialized.Retain); - CollectionAssert.AreEqual(publishPacket.Payload, deserialized.Payload); + CollectionAssert.AreEqual(publishPacket.PayloadSegment.ToArray(), deserialized.PayloadSegment.ToArray()); Assert.AreEqual(publishPacket.QualityOfServiceLevel, deserialized.QualityOfServiceLevel); Assert.AreEqual(publishPacket.Topic, deserialized.Topic); Assert.AreEqual(publishPacket.ResponseTopic, deserialized.ResponseTopic); diff --git a/Source/MQTTnet.Tests/MqttApplicationMessageTest.cs b/Source/MQTTnet.Tests/MqttApplicationMessageTest.cs new file mode 100644 index 00000000..26da8211 --- /dev/null +++ b/Source/MQTTnet.Tests/MqttApplicationMessageTest.cs @@ -0,0 +1,48 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace MQTTnet.Tests +{ + [TestClass] + public class MqttApplicationMessageTest + { + [TestMethod] + public void PayloadSegment() + { + var message = new MqttApplicationMessage(); + Assert.AreEqual(0, message.Payload.Length); + Assert.AreEqual(0, message.PayloadSegment.Count); + + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload)); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array)); + + message.Payload = new byte[] { 1, 2 }; + Assert.AreEqual(2, message.Payload.Length); + Assert.AreEqual(2, message.PayloadSegment.Count); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload)); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array)); + + message.Payload = new byte[] { 1, 2, 3 }; + Assert.AreEqual(3, message.Payload.Length); + Assert.AreEqual(3, message.PayloadSegment.Count); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload)); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array)); + + message.PayloadSegment = new ArraySegment(new byte[] { 1, 2, 3 }); + Assert.AreEqual(3, message.Payload.Length); + Assert.AreEqual(3, message.PayloadSegment.Count); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload)); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array)); + + message.PayloadSegment = new ArraySegment(new byte[] { 1, 2, 3 }, 1, 1); + Assert.AreEqual(1, message.Payload.Length); + Assert.AreEqual(1, message.PayloadSegment.Count); + Assert.IsTrue(object.ReferenceEquals(message.Payload, message.Payload)); + Assert.IsFalse(object.ReferenceEquals(message.Payload, message.PayloadSegment.Array)); + } + } +} diff --git a/Source/MQTTnet.Tests/Server/Security_Tests.cs b/Source/MQTTnet.Tests/Server/Security_Tests.cs index bc03e9c3..299c3873 100644 --- a/Source/MQTTnet.Tests/Server/Security_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Security_Tests.cs @@ -164,7 +164,7 @@ namespace MQTTnet.Tests.Server var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithCredentials(userName, password).Build(); - var ex = await Assert.ThrowsExceptionAsync(async () => await client.ConnectAsync(clientOptions)); + var ex = await Assert.ThrowsExceptionAsync(() => client.ConnectAsync(clientOptions)); Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, ex.Result.ResultCode); } } diff --git a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs index 1fd23bf7..4564ddf8 100644 --- a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs +++ b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Formatter return new MqttApplicationMessage { Topic = publishPacket.Topic, - Payload = publishPacket.Payload, + PayloadSegment = publishPacket.PayloadSegment, QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, Retain = publishPacket.Retain, Dup = publishPacket.Dup, diff --git a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs index ab33f7a9..2f3f3d3d 100644 --- a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs @@ -16,7 +16,7 @@ namespace MQTTnet.Formatter return new MqttPublishPacket { Topic = publishPacket.Topic, - Payload = publishPacket.Payload, + PayloadSegment = publishPacket.PayloadSegment, Retain = publishPacket.Retain, QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, Dup = publishPacket.Dup, @@ -36,7 +36,7 @@ namespace MQTTnet.Formatter var packet = new MqttPublishPacket { Topic = applicationMessage.Topic, - Payload = applicationMessage.Payload, + PayloadSegment = applicationMessage.PayloadSegment, QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, Retain = applicationMessage.Retain, Dup = applicationMessage.Dup, @@ -65,10 +65,16 @@ namespace MQTTnet.Formatter throw new MqttProtocolViolationException("The CONNECT packet contains no will message (WillFlag)."); } + ArraySegment willMessageBuffer = default; + if (connectPacket.WillMessage?.Length > 0) + { + willMessageBuffer = new ArraySegment(connectPacket.WillMessage); + } + var packet = new MqttPublishPacket { Topic = connectPacket.WillTopic, - Payload = connectPacket.WillMessage, + PayloadSegment = willMessageBuffer, QualityOfServiceLevel = connectPacket.WillQoS, Retain = connectPacket.WillRetain, ContentType = connectPacket.WillContentType, diff --git a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs index 2924d86e..92fbb836 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs @@ -54,12 +54,12 @@ namespace MQTTnet.Formatter.V3 return DecodePubRelPacket(receivedMqttPacket.Body); case MqttControlPacketType.PubComp: return DecodePubCompPacket(receivedMqttPacket.Body); - + case MqttControlPacketType.PingReq: return MqttPingReqPacket.Instance; case MqttControlPacketType.PingResp: return MqttPingRespPacket.Instance; - + case MqttControlPacketType.Connect: return DecodeConnectPacket(receivedMqttPacket.Body); case MqttControlPacketType.ConnAck: @@ -73,7 +73,7 @@ namespace MQTTnet.Formatter.V3 } case MqttControlPacketType.Disconnect: return DisconnectPacket; - + case MqttControlPacketType.Subscribe: return DecodeSubscribePacket(receivedMqttPacket.Body); case MqttControlPacketType.SubAck: @@ -103,9 +103,10 @@ namespace MQTTnet.Formatter.V3 var remainingLength = (uint)(_bufferWriter.Length - 5); var publishPacket = packet as MqttPublishPacket; - if (publishPacket?.Payload != null) + var payloadSegment = publishPacket?.PayloadSegment; + if (payloadSegment != null) { - remainingLength += (uint)publishPacket.Payload.Length; + remainingLength += (uint)payloadSegment.Value.Count; } var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength); @@ -119,18 +120,13 @@ namespace MQTTnet.Formatter.V3 _bufferWriter.WriteVariableByteInteger(remainingLength); var buffer = _bufferWriter.GetBuffer(); - var firstSegment = new ArraySegment(buffer, headerOffset, _bufferWriter.Length - headerOffset); - if (publishPacket?.Payload != null) - { - var payloadSegment = new ArraySegment(publishPacket.Payload, 0, publishPacket.Payload.Length); - return new MqttPacketBuffer(firstSegment, payloadSegment); - } - - return new MqttPacketBuffer(firstSegment); + return payloadSegment == null + ? new MqttPacketBuffer(firstSegment) + : new MqttPacketBuffer(firstSegment, payloadSegment.Value); } - + MqttPacket DecodeConnAckPacket(ArraySegment body) { ThrowIfBodyIsEmpty(body); @@ -177,7 +173,7 @@ namespace MQTTnet.Formatter.V3 var tryPrivate = (protocolVersion & 0x80) > 0; protocolVersion &= 0x7F; - + if (protocolVersion != 3 && protocolVersion != 4) { throw new MqttProtocolViolationException("MQTT protocol version do not match MQTT v3."); @@ -282,7 +278,7 @@ namespace MQTTnet.Formatter.V3 if (!_bufferReader.EndOfStream) { - packet.Payload = _bufferReader.ReadRemainingData(); + packet.PayloadSegment = new ArraySegment(_bufferReader.ReadRemainingData()); } return packet; @@ -421,7 +417,7 @@ namespace MQTTnet.Formatter.V3 { protocolVersion |= 0x80; } - + bufferWriter.WriteByte((byte)protocolVersion); byte connectFlags = 0x0; diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs index 9d460411..0b6627ce 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs @@ -528,7 +528,7 @@ namespace MQTTnet.Formatter.V5 if (!_bufferReader.EndOfStream) { - packet.Payload = _bufferReader.ReadRemainingData(); + packet.PayloadSegment = new ArraySegment(_bufferReader.ReadRemainingData()); } return packet; diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs index ed94c0e4..eaddaaad 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs @@ -37,9 +37,11 @@ namespace MQTTnet.Formatter.V5 var remainingLength = (uint)_bufferWriter.Length - 5; var publishPacket = packet as MqttPublishPacket; - if (publishPacket?.Payload != null) + var payloadSegment = publishPacket?.PayloadSegment; + + if (payloadSegment != null) { - remainingLength += (uint)publishPacket.Payload.Length; + remainingLength += (uint)payloadSegment.Value.Count; } var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength); @@ -53,18 +55,13 @@ namespace MQTTnet.Formatter.V5 _bufferWriter.WriteVariableByteInteger(remainingLength); var buffer = _bufferWriter.GetBuffer(); - var firstSegment = new ArraySegment(buffer, headerOffset, _bufferWriter.Length - headerOffset); - if (publishPacket?.Payload != null) - { - var payloadSegment = new ArraySegment(publishPacket.Payload, 0, publishPacket.Payload.Length); - return new MqttPacketBuffer(firstSegment, payloadSegment); - } - - return new MqttPacketBuffer(firstSegment); + return payloadSegment == null + ? new MqttPacketBuffer(firstSegment) + : new MqttPacketBuffer(firstSegment, payloadSegment.Value); } - + byte EncodeAuthPacket(MqttAuthPacket packet) { _bufferWriter.WriteByte((byte)packet.ReasonCode); diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index ff947378..bc161b6c 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -2,14 +2,19 @@ // 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.Collections.Generic; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; +using System; +using System.Collections.Generic; namespace MQTTnet { public sealed class MqttApplicationMessage { + byte[] _payloadCache; + ArraySegment _payloadSegment = EmptyBuffer.ArraySegment; + /// /// Gets or sets the content type. /// The content type must be a UTF-8 encoded string. The content type value identifies the kind of UTF-8 encoded @@ -48,10 +53,53 @@ namespace MQTTnet public uint MessageExpiryInterval { get; set; } /// - /// Gets or sets the payload. - /// The payload is the data bytes sent via the MQTT protocol. + /// Gets or sets the payload. + /// The payload is the data bytes sent via the MQTT protocol. /// - public byte[] Payload { get; set; } + [Obsolete("Use PayloadSegment instead. This property will be removed in a future release.")] + public byte[] Payload + { + get + { + if (_payloadSegment.Array == null) + { + return null; + } + + // just reference from _payloadSegment.Array + if (_payloadSegment.Count == _payloadSegment.Array.Length) + { + return _payloadSegment.Array; + } + + // copy from _payloadSegment + if (_payloadCache == null) + { + _payloadCache = new byte[_payloadSegment.Count]; + MqttMemoryHelper.Copy(_payloadSegment.Array, _payloadSegment.Offset, _payloadCache, 0, _payloadCache.Length); + } + + return _payloadCache; + } + set + { + _payloadCache = null; + _payloadSegment = value == null || value.Length == 0 ? EmptyBuffer.ArraySegment : new ArraySegment(value); + } + } + + /// + /// Get or set ArraySegment style of Payload. + /// + public ArraySegment PayloadSegment + { + get => _payloadSegment; + set + { + _payloadCache = null; + _payloadSegment = value; + } + } /// /// Gets or sets the payload format indicator. diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index 16fbbaab..a8a740a0 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -2,13 +2,16 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using MQTTnet.Exceptions; +using MQTTnet.Internal; +using MQTTnet.Packets; +using MQTTnet.Protocol; +using System; using System.Collections.Generic; using System.IO; using System.Linq; +using System.Runtime.InteropServices; using System.Text; -using MQTTnet.Exceptions; -using MQTTnet.Packets; -using MQTTnet.Protocol; namespace MQTTnet { @@ -17,7 +20,8 @@ namespace MQTTnet string _contentType; byte[] _correlationData; uint _messageExpiryInterval; - byte[] _payload; + ArraySegment _payloadSegment; + MqttPayloadFormatIndicator _payloadFormatIndicator; MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce; string _responseTopic; @@ -37,7 +41,7 @@ namespace MQTTnet var applicationMessage = new MqttApplicationMessage { Topic = _topic, - Payload = _payload, + PayloadSegment = _payloadSegment, QualityOfServiceLevel = _qualityOfServiceLevel, Retain = _retain, ContentType = _contentType, @@ -85,7 +89,15 @@ namespace MQTTnet public MqttApplicationMessageBuilder WithPayload(byte[] payload) { - _payload = payload; + _payloadSegment = payload == null || payload.Length == 0 + ? EmptyBuffer.ArraySegment + : new ArraySegment(payload); + return this; + } + + public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment payloadSegment) + { + _payloadSegment = payloadSegment; return this; } @@ -93,70 +105,73 @@ namespace MQTTnet { if (payload == null) { - _payload = null; - return this; + return WithPayload(default(byte[])); } - _payload = payload as byte[] ?? payload.ToArray(); + if (payload is byte[] byteArray) + { + return WithPayload(byteArray); + } - return this; + if (payload is ArraySegment arraySegment) + { + return WithPayloadSegment(arraySegment); + } + + return WithPayload(payload.ToArray()); } public MqttApplicationMessageBuilder WithPayload(Stream payload) { - if (payload == null) - { - _payload = null; - return this; - } - - return WithPayload(payload, payload.Length - payload.Position); + return payload == null + ? WithPayload(default(byte[])) + : WithPayload(payload, payload.Length - payload.Position); } public MqttApplicationMessageBuilder WithPayload(Stream payload, long length) { - if (payload == null) + if (payload == null || length == 0) { - _payload = null; - return this; + return WithPayload(default(byte[])); } - if (payload.Length == 0) + var payloadBuffer = new byte[length]; + var totalRead = 0; + do { - _payload = null; - } - else - { - _payload = new byte[length]; - - var totalRead = 0; - do + var bytesRead = payload.Read(payloadBuffer, totalRead, payloadBuffer.Length - totalRead); + if (bytesRead == 0) { - var bytesRead = payload.Read(_payload, totalRead, _payload.Length - totalRead); - if (bytesRead == 0) - { - break; - } + break; + } - totalRead += bytesRead; - } while (totalRead < length); - } + totalRead += bytesRead; + } while (totalRead < length); - return this; + return WithPayload(payloadBuffer); } public MqttApplicationMessageBuilder WithPayload(string payload) { - if (payload == null) + if (string.IsNullOrEmpty(payload)) { - _payload = null; - return this; + return WithPayload(default(byte[])); } - _payload = string.IsNullOrEmpty(payload) ? null : Encoding.UTF8.GetBytes(payload); - return this; + var payloadBuffer = Encoding.UTF8.GetBytes(payload); + return WithPayload(payloadBuffer); } + +#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1 + public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory payloadSegment) + { + return MemoryMarshal.TryGetArray(payloadSegment, out var segment) + ? WithPayloadSegment(segment) + : WithPayload(payloadSegment.ToArray()); + } +#endif + /// /// Adds the payload format indicator to the message. /// MQTT 5.0.0+ feature. diff --git a/Source/MQTTnet/MqttApplicationMessageExtensions.cs b/Source/MQTTnet/MqttApplicationMessageExtensions.cs index 1bfd69a5..1f3b9059 100644 --- a/Source/MQTTnet/MqttApplicationMessageExtensions.cs +++ b/Source/MQTTnet/MqttApplicationMessageExtensions.cs @@ -11,19 +11,13 @@ namespace MQTTnet { public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage) { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - - if (applicationMessage.Payload == null) + if (applicationMessage == null) { - return null; + throw new ArgumentNullException(nameof(applicationMessage)); } - if (applicationMessage.Payload.Length == 0) - { - return string.Empty; - } - - return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length); + var payloadSegment = applicationMessage.PayloadSegment; + return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count); } } } diff --git a/Source/MQTTnet/Packets/MqttPublishPacket.cs b/Source/MQTTnet/Packets/MqttPublishPacket.cs index 10aad7b0..3869ea39 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacket.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacket.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; using System.Collections.Generic; using MQTTnet.Protocol; @@ -17,10 +18,10 @@ namespace MQTTnet.Packets public uint MessageExpiryInterval { get; set; } - public byte[] Payload { get; set; } - public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified; + public ArraySegment PayloadSegment { get; set; } + public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } = MqttQualityOfServiceLevel.AtMostOnce; public string ResponseTopic { get; set; } @@ -38,7 +39,7 @@ namespace MQTTnet.Packets public override string ToString() { return - $"Publish: [Topic={Topic}] [Payload.Length={Payload?.Length}] [QoSLevel={QualityOfServiceLevel}] [Dup={Dup}] [Retain={Retain}] [PacketIdentifier={PacketIdentifier}]"; + $"Publish: [Topic={Topic}] [PayloadLength={PayloadSegment.Count}] [QoSLevel={QualityOfServiceLevel}] [Dup={Dup}] [Retain={Retain}] [PacketIdentifier={PacketIdentifier}]"; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index 93ebf7e4..4dc0b2c0 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -25,6 +25,7 @@ namespace MQTTnet.Server if (logger == null) throw new ArgumentNullException(nameof(logger)); + _logger = logger.WithSource(nameof(MqttRetainedMessagesManager)); } @@ -68,7 +69,8 @@ namespace MQTTnet.Server lock (_messages) { - var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0; + var payloadSegment = applicationMessage.PayloadSegment; + var hasPayload = payloadSegment.Count > 0; if (!hasPayload) { @@ -84,8 +86,7 @@ namespace MQTTnet.Server } else { - if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || - !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? EmptyBuffer.Array)) + if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !SequenceEqual(existingMessage.PayloadSegment, payloadSegment)) { _messages[applicationMessage.Topic] = applicationMessage; saveIsRequired = true; @@ -150,5 +151,14 @@ namespace MQTTnet.Server await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } } + + private static bool SequenceEqual(ArraySegment source, ArraySegment target) + { +#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1 + return source.AsSpan().SequenceEqual(target); +#else + return source.Count == target.Count && Enumerable.SequenceEqual(source, target); +#endif + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServerExtensions.cs b/Source/MQTTnet/Server/MqttServerExtensions.cs index 13746e50..72cdac24 100644 --- a/Source/MQTTnet/Server/MqttServerExtensions.cs +++ b/Source/MQTTnet/Server/MqttServerExtensions.cs @@ -41,7 +41,7 @@ namespace MQTTnet.Server new MqttApplicationMessage { Topic = topic, - Payload = payloadBuffer, + PayloadSegment = new ArraySegment(payloadBuffer), QualityOfServiceLevel = qualityOfServiceLevel, Retain = retain }));