Merge pull request #508 from AArnott/main_from_v2.8

Merge v2.8 into main
dotnet · Jul 22, 2022 · b5d7fe1 · b5d7fe1
2 parents bae88b8 + b857951
commit b5d7fe1
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 10 deletions.
diff --git a/src/Nerdbank.Streams/MultiplexingStream.Channel.cs b/src/Nerdbank.Streams/MultiplexingStream.Channel.cs
@@ -1067,10 +1067,6 @@ public override bool TryRead(out ReadResult readResult)
 
                 public override ValueTask CompleteAsync(Exception? exception = null) => this.inner.CompleteAsync(exception);
 
-                public override Task CopyToAsync(PipeWriter destination, CancellationToken cancellationToken = default) => this.inner.CopyToAsync(destination, cancellationToken);
-
-                public override Task CopyToAsync(Stream destination, CancellationToken cancellationToken = default) => this.inner.CopyToAsync(destination, cancellationToken);
-
                 [Obsolete]
                 public override void OnWriterCompleted(Action<Exception?, object?> callback, object? state) => this.inner.OnWriterCompleted(callback, state);
 

diff --git a/test/Nerdbank.Streams.Tests/MultiplexingStreamV2Tests.cs b/test/Nerdbank.Streams.Tests/MultiplexingStreamV2Tests.cs
@@ -136,6 +136,28 @@ public async Task Backpressure_ExistingPipe()
         await writeTask;
     }
 
+    [Fact]
+    public async Task Backpressure_CopyToAsync()
+    {
+        long backpressureThreshold = this.mx1.DefaultChannelReceivingWindowSize;
+        (MultiplexingStream.Channel a, MultiplexingStream.Channel b) = await this.EstablishChannelsAsync("a");
+
+        byte[]? hugeChunk = new byte[backpressureThreshold * 2]; // enough to fill the remote and local windows
+        a.Output.Write(hugeChunk);
+        Task flushTask = Task.Run(async delegate
+        {
+            await a.Output.FlushAsync(this.TimeoutToken);
+            await a.Output.CompleteAsync();
+        });
+
+        // Now read from the channel and verify it unblocks the writer, using CopyToAsync specifically.
+        long drainedBytesCount = await this.DrainReaderTillCompletedAsync(b.Input, useCopyToAsync: true);
+        Assert.Equal(hugeChunk.Length, drainedBytesCount);
+
+        await flushTask.WithCancellation(this.TimeoutToken);
+        await CompleteChannelsAsync(a, b);
+    }
+
     /// <summary>
     /// Regression test for <see href="https://github.com/AArnott/Nerdbank.Streams/issues/253">#253</see>.
     /// </summary>

diff --git a/test/Nerdbank.Streams.Tests/TestBase.cs b/test/Nerdbank.Streams.Tests/TestBase.cs
@@ -146,17 +146,30 @@ public async Task DrainAsync(PipeReader reader, long requiredLength)
         }
     }
 
-    public async Task DrainReaderTillCompletedAsync(PipeReader reader)
+    public async Task<long> DrainReaderTillCompletedAsync(PipeReader reader, bool useCopyToAsync = false)
     {
-        while (true)
+        long bytesDrained = 0;
+        if (useCopyToAsync)
         {
-            ReadResult readResult = await reader.ReadAsync(this.TimeoutToken);
-            reader.AdvanceTo(readResult.Buffer.End);
-            if (readResult.IsCompleted)
+            MemoryStream ms = new();
+            await reader.CopyToAsync(ms, this.TimeoutToken);
+            bytesDrained = ms.Length;
+        }
+        else
+        {
+            while (true)
             {
-                break;
+                ReadResult readResult = await reader.ReadAsync(this.TimeoutToken);
+                bytesDrained += readResult.Buffer.Length;
+                reader.AdvanceTo(readResult.Buffer.End);
+                if (readResult.IsCompleted)
+                {
+                    break;
+                }
             }
         }
+
+        return bytesDrained;
     }
 
     internal byte[] GetBuffer(int length)