Reduce memory allocationt.

Samples/Client/Client_Publish_Samples.cs

@@ -31,12 +31,10 @@ public static async Task Publish_Application_Message()
 
             await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-            var applicationMessage = new MqttApplicationMessageBuilder()
-                .WithTopic("samples/temperature/living_room")
-                .WithPayload("19.5")
-                .Build();
-
-            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
+            await mqttClient.PublishStringAsync(
+                topic: "samples/temperature/living_room",
+                payload: "19.5",
+                cancellationToken: CancellationToken.None);
 
             await mqttClient.DisconnectAsync();
 
@@ -61,27 +59,21 @@ public static async Task Publish_Multiple_Application_Messages()
                 .Build();
 
             await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
-
-            var applicationMessage = new MqttApplicationMessageBuilder()
-                .WithTopic("samples/temperature/living_room")
-                .WithPayload("19.5")
-                .Build();
-
-            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
-
-            applicationMessage = new MqttApplicationMessageBuilder()
-                .WithTopic("samples/temperature/living_room")
-                .WithPayload("20.0")
-                .Build();
-
-            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
-
-            applicationMessage = new MqttApplicationMessageBuilder()
-                .WithTopic("samples/temperature/living_room")
-                .WithPayload("21.0")
-                .Build();
-
-            await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);
+
+            await mqttClient.PublishStringAsync(
+                topic: "samples/temperature/living_room",
+                payload: "19.5",
+                cancellationToken: CancellationToken.None);
+
+            await mqttClient.PublishStringAsync(
+                topic: "samples/temperature/living_room",
+                payload: "20.0",
+                cancellationToken: CancellationToken.None);
+
+            await mqttClient.PublishStringAsync(
+                topic: "samples/temperature/living_room",
+                payload: "21.0",
+                cancellationToken: CancellationToken.None);
 
             await mqttClient.DisconnectAsync();
 

Samples/MQTTnet.Samples.csproj

@@ -14,7 +14,7 @@
         <NuGetAuditMode>all</NuGetAuditMode>
         <NuGetAudit>true</NuGetAudit>
         <NuGetAuditLevel>low</NuGetAuditLevel>
-        <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
     </PropertyGroup>
 
     <ItemGroup>

Samples/RpcClient/RpcClient_Samples.cs

@@ -38,7 +38,7 @@ public static async Task Send_Request()
             {
                 // Access to a fully featured application message is not supported for RPC calls!
                 // The method will throw an exception when the response was not received in time.
-                await mqttRpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+                await mqttRpcClient.ExecuteTimeOutAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
             }
 
             Console.WriteLine("The RPC call was successful.");

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -92,8 +92,8 @@ public static async Task Persist_Retained_Messages()
     sealed class MqttRetainedMessageModel
     {
         public string? ContentType { get; set; }
-        public byte[]? CorrelationData { get; set; }
-        public byte[]? Payload { get; set; }
+        public ReadOnlyMemory<byte> CorrelationData { get; set; }
+        public ReadOnlySequence<byte> Payload { get; set; }
         public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; }
         public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
         public string? ResponseTopic { get; set; }
@@ -110,7 +110,7 @@ public static MqttRetainedMessageModel Create(MqttApplicationMessage message)
 
                 // Create a copy of the buffer from the payload segment because
                 // it cannot be serialized and deserialized with the JSON serializer.
-                Payload = message.Payload.ToArray(),
+                Payload = new ReadOnlySequence<byte>(message.Payload.ToArray()),
                 UserProperties = message.UserProperties,
                 ResponseTopic = message.ResponseTopic,
                 CorrelationData = message.CorrelationData,
@@ -128,7 +128,7 @@ public MqttApplicationMessage ToApplicationMessage()
             return new MqttApplicationMessage
             {
                 Topic = Topic,
-                PayloadSegment = new ArraySegment<byte>(Payload ?? Array.Empty<byte>()),
+                Payload = Payload,
                 PayloadFormatIndicator = PayloadFormatIndicator,
                 ResponseTopic = ResponseTopic,
                 CorrelationData = CorrelationData,

Samples/Server/Server_Simple_Samples.cs

@@ -44,18 +44,13 @@ public static async Task Publish_Message_From_Broker()
          * See _Run_Minimal_Server_ for more information.
          */
 
-        using (var mqttServer = await StartMqttServer())
-        {
-            // Create a new message using the builder as usual.
-            var message = new MqttApplicationMessageBuilder().WithTopic("HelloWorld").WithPayload("Test").Build();
+        using var mqttServer = await StartMqttServer();
 
-            // Now inject the new message at the broker.
-            await mqttServer.InjectApplicationMessage(
-                new InjectedMqttApplicationMessage(message)
-                {
-                    SenderClientId = "SenderClientId"
-                });
-        }
+        // Now inject the new message at the broker.
+        await mqttServer.InjectStringAsync(
+            clientId: "SenderClientId",
+            topic: "HelloWorld",
+            payload: "Test");
     }
 
     public static async Task Run_Minimal_Server()

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -37,7 +37,7 @@
         <NuGetAudit>true</NuGetAudit>
         <NuGetAuditLevel>low</NuGetAuditLevel>
         <NuGetAuditLevel>low</NuGetAuditLevel>
-        <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
     </PropertyGroup>
 
     <ItemGroup>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -231,8 +231,8 @@ static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
 
         var span = output.GetSpan(buffer.Length);
 
-        buffer.Packet.AsSpan().CopyTo(span);
-        int offset = buffer.Packet.Count;
+        buffer.Packet.Span.CopyTo(span);
+        int offset = buffer.Packet.Length;
         buffer.Payload.CopyTo(destination: span.Slice(offset));
         output.Advance(buffer.Length);
     }

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -48,9 +48,7 @@ public static bool TryDecode(
         }
 
         var bodySlice = copy.Slice(0, bodyLength);
-        var bodySegment = GetArraySegment(ref bodySlice);
-
-        var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, bodySegment, headerLength + bodyLength);
+        var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, bodySlice, headerLength + bodyLength);
         if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
         {
             formatter.DetectProtocolVersion(receivedMqttPacket);
@@ -62,19 +60,7 @@ public static bool TryDecode(
         bytesRead = headerLength + bodyLength;
         return true;
     }
-
-    static ArraySegment<byte> GetArraySegment(ref ReadOnlySequence<byte> input)
-    {
-        if (input.IsSingleSegment && MemoryMarshal.TryGetArray(input.First, out var segment))
-        {
-            return segment;
-        }
-
-        // Should be rare
-        var array = input.ToArray();
-        return new ArraySegment<byte>(array);
-    }
-
+
 
     static void ThrowProtocolViolationException(ReadOnlySpan<byte> valueSpan, int index)
     {

Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -76,12 +76,12 @@ public void Setup()
             _channelAdapter = new MqttChannelAdapter(channel, serializer, new MqttNetEventLogger());
         }
 
-        static byte[] Join(params ArraySegment<byte>[] chunks)
+        static byte[] Join(params ReadOnlyMemory<byte>[] chunks)
         {
             var buffer = new MemoryStream();
             foreach (var chunk in chunks)
             {
-                buffer.Write(chunk.Array, chunk.Offset, chunk.Count);
+                buffer.Write(chunk.Span);
             }
 
             return buffer.ToArray();

Source/MQTTnet.Benchmarks/JsonPayloadBenchmark.cs

@@ -0,0 +1,82 @@
+// 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 BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using MQTTnet.Internal;
+using System.Text.Json;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net80)]
+    [MemoryDiagnoser]
+    public class JsonPayloadBenchmark : BaseBenchmark
+    {
+        [Params(1 * 1024, 4 * 1024, 8 * 1024)]
+        public int PayloadSize { get; set; }
+        private Model model;
+
+
+        [GlobalSetup]
+        public void Setup()
+        {
+            var stringValue = new char[PayloadSize];
+            model = new Model { StringValue = new string(stringValue) };
+        }
+
+        [Benchmark]
+        public ValueTask SerializeToString_Payload_SendAsync()
+        {
+            string payload = JsonSerializer.Serialize(model);
+            var message = new MqttApplicationMessageBuilder()
+                .WithTopic("t")
+                .WithPayload(payload)
+                .Build();
+
+            // send message async
+            return ValueTask.CompletedTask;
+        }
+
+        [Benchmark]
+        public ValueTask SerializeToUtf8Bytes_Payload_SendAsync()
+        {
+            byte[] payload = JsonSerializer.SerializeToUtf8Bytes(model);
+            var message = new MqttApplicationMessageBuilder()
+                .WithTopic("t")
+                .WithPayload(payload)
+                .Build();
+
+            // send message async
+            return ValueTask.CompletedTask;
+        }
+
+        [Benchmark(Baseline = true)]
+        public async ValueTask MqttPayloadOwnerFactory_Payload_SendAsync()
+        {
+            await using var payloadOwner = await MqttPayloadOwnerFactory.CreateMultipleSegmentAsync(async writer =>
+               await JsonSerializer.SerializeAsync(writer.AsStream(leaveOpen: true), model));
+
+            var message = new MqttApplicationMessageBuilder()
+                .WithTopic("t")
+                .WithPayload(payloadOwner.Payload)
+                .Build();
+
+            // send message async
+        }
+
+
+
+        public class Model
+        {
+            public string StringValue { get; set; }
+
+            public int IntValue { get; set; }
+
+            public bool BoolValue { get; set; }
+
+            public double DoubleValue { get; set; }
+        }
+    }
+}

Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -14,7 +14,7 @@
         <NuGetAuditMode>all</NuGetAuditMode>
         <NuGetAudit>true</NuGetAudit>
         <NuGetAuditLevel>low</NuGetAuditLevel>
-        <AnalysisLevel>latest-Recommended</AnalysisLevel>
+        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
     </PropertyGroup>
 
     <ItemGroup>

Source/MQTTnet.Benchmarks/MqttBufferReaderBenchmark.cs

@@ -2,42 +2,43 @@
 // 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;
+using System;
+using System.Runtime.InteropServices;
+using System.Text;
 
 namespace MQTTnet.Benchmarks
 {
     [SimpleJob(RuntimeMoniker.Net60)]
     [MemoryDiagnoser]
     public class MqttBufferReaderBenchmark
     {
-        byte[] _buffer;
-        int _bufferLength;
+        ArraySegment<byte> _buffer;
 
         [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;
+            if (MemoryMarshal.TryGetArray(writer.GetWrittenMemory(), out var segment))
+            {
+                _buffer = segment;
+            }
         }
 
         [Benchmark]
         public void Use_Span()
         {
-            var span = _buffer.AsSpan(0, _bufferLength);
-            Encoding.UTF8.GetString(span);
+            Encoding.UTF8.GetString(_buffer.AsSpan());
         }
-        
+
         [Benchmark]
         public void Use_Encoding()
         {
-            Encoding.UTF8.GetString(_buffer, 0, _bufferLength);
+            Encoding.UTF8.GetString(_buffer.Array, _buffer.Offset, _buffer.Count);
         }
     }
 }

Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs

@@ -15,8 +15,8 @@ namespace MQTTnet.Benchmarks
     public class MqttPacketReaderWriterBenchmark : BaseBenchmark
     {
         readonly byte[] _demoPayload = new byte[1024];
-        
-        byte[] _readPayload;
+
+        ReadOnlyMemory<byte> _readPayload;
 
         [GlobalCleanup]
         public void GlobalCleanup()
@@ -27,7 +27,7 @@ public void GlobalCleanup()
         public void GlobalSetup()
         {
             TestEnvironment.EnableLogger = false;
-            
+
             var writer = new MqttBufferWriter(4096, 65535);
             writer.WriteString("A relative short string.");
             writer.WriteBinary(_demoPayload);
@@ -42,18 +42,16 @@ public void GlobalSetup()
             writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug");
             writer.WriteBinary(_demoPayload);
 
-            _readPayload = new ArraySegment<byte>(writer.GetBuffer(), 0, writer.Length).ToArray();
+            _readPayload = writer.GetWrittenMemory();
         }
 
         [Benchmark]
         public void Read_100_000_Messages()
         {
-            var reader = new MqttBufferReader();
-            reader.SetBuffer(_readPayload, 0, _readPayload.Length);
-
             for (var i = 0; i < 100000; i++)
             {
-                reader.Seek(0);
+                var reader = new MqttBufferReader();
+                reader.SetBuffer(_readPayload);
 
                 reader.ReadString();
                 reader.ReadBinaryData();
@@ -69,7 +67,7 @@ public void Read_100_000_Messages()
                 reader.ReadBinaryData();
             }
         }
-        
+
         [Benchmark]
         public void Write_100_000_Messages()
         {