Merge remote-tracking branch 'origin/develop' into develop

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -162,11 +162,12 @@ namespace MQTTnet.AspNetCore
                 var buffer = formatter.Encode(packet);
                 var msg = buffer.AsMemory();
                 var output = _output;
-                msg.CopyTo(output.GetMemory(msg.Length));
+                var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
-                BytesSent += msg.Length;
+                if (result.IsCompleted)
+                {
+                    BytesSent += msg.Length;
+                }
                 PacketFormatterAdapter.FreeBuffer();
-                output.Advance(msg.Length);
-                await output.FlushAsync().ConfigureAwait(false);
             }
             finally
             {

Tests/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs

@@ -4,13 +4,21 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
 {
     public class DuplexPipeMockup : IDuplexPipe
     {
+        public DuplexPipeMockup()
+        {
+            var pool = new LimitedMemoryPool();
+            var pipeOptions = new PipeOptions(pool);
+            Receive = new Pipe(pipeOptions);
+            Send = new Pipe(pipeOptions);
+        }
+
         PipeReader IDuplexPipe.Input => Receive.Reader;
 
         PipeWriter IDuplexPipe.Output => Send.Writer;
 
 
-        public Pipe Receive { get; set; } = new Pipe();
+        public Pipe Receive { get; set; } 
 
-        public Pipe Send { get; set; } = new Pipe();
+        public Pipe Send { get; set; }
     }
 }

Tests/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs

@@ -0,0 +1,18 @@
+using System.Buffers;
+
+namespace MQTTnet.AspNetCore.Tests.Mockups
+{
+    public class LimitedMemoryPool : MemoryPool<byte>
+    {
+        protected override void Dispose(bool disposing)
+        {
+        }
+
+        public override IMemoryOwner<byte> Rent(int minBufferSize = -1)
+        {
+            return new MemoryOwner(minBufferSize);
+        }
+
+        public override int MaxBufferSize { get; }
+    }
+}

Tests/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Buffers;
+
+namespace MQTTnet.AspNetCore.Tests.Mockups
+{
+    public class MemoryOwner : IMemoryOwner<byte>
+    {
+        private readonly byte[] _raw;
+
+        public MemoryOwner(int size)
+        {
+            if (size <= 0)
+            {
+                size = 1024;
+            }
+
+            if (size > 4096)
+            {
+                size = 4096;
+            }
+
+            _raw = ArrayPool<byte>.Shared.Rent(size);
+            Memory = _raw;
+        }
+
+        public void Dispose()
+        {
+            ArrayPool<byte>.Shared.Return(_raw);
+        }
+
+        public Memory<byte> Memory { get; }
+    }
+}

Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -47,5 +47,21 @@ namespace MQTTnet.AspNetCore.Tests
 
             await Task.WhenAll(tasks).ConfigureAwait(false);
         }
+
+
+        [TestMethod]
+        public async Task TestLargePacket()
+        {
+            var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
+            var pipe = new DuplexPipeMockup();
+            var connection = new DefaultConnectionContext();
+            connection.Transport = pipe;
+            var ctx = new MqttConnectionContext(serializer, connection);
+
+            await ctx.SendPacketAsync(new MqttPublishPacket() { Payload = new byte[20_000] }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
+
+            var readResult = await pipe.Send.Reader.ReadAsync();
+            Assert.IsTrue(readResult.Buffer.Length > 20000);
+        }
     }
 }