SingleSegmentPayloadOwner: use payloadFactory

Source/MQTTnet.Server/MqttServerExtensions.cs

@@ -111,8 +111,7 @@ public static async Task InjectBinaryAsync(
        bool retain = false,
        CancellationToken cancellationToken = default)
     {
-        await using var payloadOwner = MqttPayloadOwnerFactory.CreateSingleSegment(payloadSize, out var payloadMemory);
-        payloadFactory?.Invoke(payloadMemory);
+        await using var payloadOwner = MqttPayloadOwnerFactory.CreateSingleSegment(payloadSize, payloadFactory);
         await server.InjectSequenceAsync(clientId, topic, payloadOwner.Payload, qualityOfServiceLevel, retain, cancellationToken);
     }
 

Source/MQTTnet.Tests/Internal/MqttPayloadOwnerFactory_Test.cs

@@ -19,11 +19,16 @@ public class MqttPayloadOwnerFactory_Test
         public async Task CreateSingleSegmentTest()
         {
             var size = 10;
-            await using var owner = MqttPayloadOwnerFactory.CreateSingleSegment(size, out var payloadMemory);
-            Random.Shared.NextBytes(payloadMemory.Span);
+            var buffer = new byte[size];
+            Random.Shared.NextBytes(buffer);
+
+            await using var owner = MqttPayloadOwnerFactory.CreateSingleSegment(size, payload =>
+            {
+                buffer.AsSpan().CopyTo(payload.Span);
+            });
 
             Assert.AreEqual(size, owner.Payload.Length);
-            Assert.IsTrue(payloadMemory.Span.SequenceEqual(owner.Payload.ToArray()));
+            Assert.IsTrue(buffer.AsSpan().SequenceEqual(owner.Payload.ToArray()));
         }
 
         [TestMethod]

Source/MQTTnet.Tests/Server/Publishing_Tests.cs

@@ -27,7 +27,7 @@ public async Task Disconnect_While_Publishing()
                 server.InterceptingPublishAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
 
                 var client = await testEnvironment.ConnectClient();
-                await client.PublishStringAsync("test", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+                await client.PublishStringAsync("test",null, qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
             }
         }
 

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -32,7 +32,7 @@ public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
             await client1.SubscribeAsync("A");
 
             var client2 = await testEnvironment.ConnectClient();
-            await client2.PublishStringAsync("A");
+            await client2.PublishStringAsync("A", null);
 
             await LongTestDelay();
 
@@ -59,7 +59,7 @@ public async Task Fire_Event_On_Client_Acknowledges_QoS_1()
             await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtLeastOnce);
 
             var client2 = await testEnvironment.ConnectClient();
-            await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+            await client2.PublishStringAsync("A",null, qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
 
             await LongTestDelay();
 
@@ -94,7 +94,7 @@ public async Task Fire_Event_On_Client_Acknowledges_QoS_2()
             await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.ExactlyOnce);
 
             var client2 = await testEnvironment.ConnectClient();
-            await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
+            await client2.PublishStringAsync("A",null, qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
 
             await LongTestDelay();
 

Source/MQTTnet.Tests/Server/Security_Tests.cs

@@ -58,7 +58,7 @@ await validClient.ConnectAsync(
                             .WithClientId("CLIENT")
                             .Build());
 
-                    await validClient.PublishStringAsync("HELLO 1");
+                    await validClient.PublishStringAsync("HELLO 1", null);
 
                     // The following code tries to connect a new client with the same client ID but invalid
                     // credentials. This should block the second client but keep the first one up and running.
@@ -81,11 +81,11 @@ await invalidClient.ConnectAsync(
 
                     await LongTestDelay();
 
-                    await validClient.PublishStringAsync("HELLO 2");
+                    await validClient.PublishStringAsync("HELLO 2", null);
 
                     await LongTestDelay();
 
-                    await validClient.PublishStringAsync("HELLO 3");
+                    await validClient.PublishStringAsync("HELLO 3", null);
 
                     await LongTestDelay();
 

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -300,7 +300,7 @@ public async Task Set_Session_Item()
                 Assert.AreEqual(MqttClientSubscribeResultCode.GrantedQoS0, subscribeResult.Items.First().ResultCode);
 
                 var client2 = await testEnvironment.ConnectClient();
-                await client2.PublishStringAsync("x");
+                await client2.PublishStringAsync("x",null);
 
                 await Task.Delay(1000);
 

Source/MQTTnet.Tests/Server/Status_Tests.cs

@@ -158,7 +158,7 @@ public async Task Track_Sent_Application_Messages()
 
             for (var i = 1; i < 25; i++)
             {
-                await c1.PublishStringAsync("a");
+                await c1.PublishStringAsync("a",null);
                 await Task.Delay(50);
 
                 var clientStatus = await server.GetClientsAsync();

Source/MQTTnet.Tests/Server/Subscribe_Tests.cs

@@ -187,7 +187,7 @@ public async Task Intercept_Subscription()
 
                 await client.SubscribeAsync("b");
 
-                await client.PublishStringAsync("a");
+                await client.PublishStringAsync("a", null);
 
                 await Task.Delay(500);
 
@@ -323,15 +323,15 @@ public async Task Subscribe_Multiple_In_Multiple_Request()
 
                 var c2 = await testEnvironment.ConnectClient();
 
-                await c2.PublishStringAsync("a");
+                await c2.PublishStringAsync("a", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 1);
 
-                await c2.PublishStringAsync("b");
+                await c2.PublishStringAsync("b", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 2);
 
-                await c2.PublishStringAsync("c");
+                await c2.PublishStringAsync("c", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 3);
             }
@@ -357,15 +357,15 @@ public async Task Subscribe_Multiple_In_Single_Request()
 
                 var c2 = await testEnvironment.ConnectClient();
 
-                await c2.PublishStringAsync("a");
+                await c2.PublishStringAsync("a", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 1);
 
-                await c2.PublishStringAsync("b");
+                await c2.PublishStringAsync("b", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 2);
 
-                await c2.PublishStringAsync("c");
+                await c2.PublishStringAsync("c", null);
                 await Task.Delay(100);
                 Assert.AreEqual(receivedMessagesCount, 3);
             }

Source/MQTTnet/Internal/MqttPayloadOwnerFactory.cs

@@ -17,12 +17,12 @@ public static class MqttPayloadOwnerFactory
         /// Create owner for a single segment payload
         /// </summary>
         /// <param name="payloadSize"></param>
-        /// <param name="payloadMemory"></param>
+        /// <param name="payloadFactory"></param>
         /// <returns></returns>
-        public static MqttPayloadOwner CreateSingleSegment(int payloadSize, out Memory<byte> payloadMemory)
+        public static MqttPayloadOwner CreateSingleSegment(int payloadSize, Action<Memory<byte>> payloadFactory)
         {
-            ArgumentOutOfRangeException.ThrowIfNegativeOrZero(payloadSize);
-            return SingleSegmentPayloadOwner.Create(payloadSize, out payloadMemory);
+            ArgumentNullException.ThrowIfNull(payloadFactory);
+            return SingleSegmentPayloadOwner.Create(payloadSize, payloadFactory);
         }
 
         /// <summary>
@@ -42,22 +42,38 @@ private sealed class SingleSegmentPayloadOwner : MqttPayloadOwner
             private readonly byte[] _buffer;
             public override ReadOnlySequence<byte> Payload { get; }
 
-            private SingleSegmentPayloadOwner(byte[] buffer, ReadOnlyMemory<byte> payloadMemory)
+            private SingleSegmentPayloadOwner(byte[] buffer, ReadOnlyMemory<byte> payload)
             {
                 _buffer = buffer;
-                Payload = new ReadOnlySequence<byte>(payloadMemory);
+                Payload = new ReadOnlySequence<byte>(payload);
             }
 
-            public static MqttPayloadOwner Create(int payloadSize, out Memory<byte> payloadMemory)
+            public static MqttPayloadOwner Create(int payloadSize, Action<Memory<byte>> payloadFactory)
             {
-                var buffer = ArrayPool<byte>.Shared.Rent(payloadSize);
-                payloadMemory = buffer.AsMemory(0, payloadSize);
-                return new SingleSegmentPayloadOwner(buffer, payloadMemory);
+                byte[] buffer;
+                Memory<byte> payload;
+
+                if (payloadSize <= 0)
+                {
+                    buffer = Array.Empty<byte>();
+                    payload = Memory<byte>.Empty;
+                }
+                else
+                {
+                    buffer = ArrayPool<byte>.Shared.Rent(payloadSize);
+                    payload = buffer.AsMemory(0, payloadSize);
+                }
+
+                payloadFactory.Invoke(payload);
+                return new SingleSegmentPayloadOwner(buffer, payload);
             }
 
             protected override ValueTask DisposeAsync(bool disposing)
             {
-                ArrayPool<byte>.Shared.Return(_buffer);
+                if (_buffer.Length > 0)
+                {
+                    ArrayPool<byte>.Shared.Return(_buffer);
+                }
                 return ValueTask.CompletedTask;
             }
         }

Source/MQTTnet/MqttClientExtensions.cs

@@ -97,8 +97,7 @@ public static async Task<MqttClientPublishResult> PublishBinaryAsync(
        bool retain = false,
        CancellationToken cancellationToken = default)
     {
-        await using var payloadOwner = MqttPayloadOwnerFactory.CreateSingleSegment(payloadSize, out var payloadMemory);
-        payloadFactory?.Invoke(payloadMemory);
+        await using var payloadOwner = MqttPayloadOwnerFactory.CreateSingleSegment(payloadSize, payloadFactory);
         return await mqttClient.PublishSequenceAsync(topic, payloadOwner.Payload, qualityOfServiceLevel, retain, cancellationToken);
     }