Fix StreamPipeReader to better emulate other PipeReader implementations

Add many more StreamPipeReader tests and fix up some behaviors

Author: AArnott

ext/MessagePack

@@ -0,0 +1 @@
+Subproject commit a3e2a28e785fab6a413164e5bec278a7b83758d6

src/Nerdbank.Streams.Tests/IOPipelinesStreamPipeReaderTests.cs

@@ -0,0 +1,16 @@
+// Copyright (c) Andrew Arnott. All rights reserved.
+// Licensed under the MIT license. See LICENSE.txt file in the project root for full license information.
+
+using System.IO;
+using System.IO.Pipelines;
+using Xunit.Abstractions;
+
+public class IOPipelinesStreamPipeReaderTests : StreamPipeReaderTestBase
+{
+    public IOPipelinesStreamPipeReaderTests(ITestOutputHelper logger)
+        : base(logger)
+    {
+    }
+
+    protected override PipeReader CreatePipeReader(Stream stream, int hintSize = 0) => PipeReader.Create(stream, new StreamPipeReaderOptions(bufferSize: hintSize == 0 ? -1 : hintSize));
+}

src/Nerdbank.Streams.Tests/StreamPipeReaderTestBase.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT license. See LICENSE.txt file in the project root for full license information.
 
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.IO;
 using System.IO.Pipelines;
@@ -22,24 +23,19 @@ public StreamPipeReaderTestBase(ITestOutputHelper logger)
     {
     }
 
+    protected virtual bool EmulatePipelinesStreamPipeReader => true;
+
     [Fact]
     public void ThrowsOnNull()
     {
         Assert.Throws<ArgumentNullException>(() => this.CreatePipeReader((Stream)null!));
     }
 
-    [Fact]
-    public void NonReadableStream()
-    {
-        var unreadableStream = new Mock<Stream>(MockBehavior.Strict);
-        unreadableStream.SetupGet(s => s.CanRead).Returns(false);
-        Assert.Throws<ArgumentException>(() => this.CreatePipeReader(unreadableStream.Object));
-        unreadableStream.VerifyAll();
-    }
-
-    [Fact]
+    [SkippableFact]
     public async Task Stream()
     {
+        Skip.If(this is IOPipelinesStreamPipeReaderTests, "OnWriterCompleted isn't supported.");
+
         byte[] expectedBuffer = this.GetRandomBuffer(2048);
         var stream = new MemoryStream(expectedBuffer);
         var reader = this.CreatePipeReader(stream, sizeHint: 50);
@@ -141,8 +137,110 @@ public async Task TryRead()
     }
 
     [Fact]
+    public void TryRead_FalseStillTurnsOnReadingMode()
+    {
+        var reader = this.CreatePipeReader(new MemoryStream(new byte[] { 1, 2, 3 }));
+
+        bool tryReadResult = reader.TryRead(out var readResult);
+
+        // The non-strict PipeReader is timing sensitive and may or may not be able to synchronously produce a read.
+        // So only assert the False result for the strict readers.
+        if (this.EmulatePipelinesStreamPipeReader)
+        {
+            Assert.False(tryReadResult);
+        }
+
+        reader.AdvanceTo(readResult.Buffer.End);
+    }
+
+    [Fact]
+    public void TryRead_FalseCanBeCalledRepeatedly()
+    {
+        var reader = this.CreatePipeReader(new MemoryStream(new byte[] { 1, 2, 3 }));
+
+        // Verify that it's safe to call TryRead repeatedly when it returns False.
+        Assert.False(reader.TryRead(out var readResult));
+        Assert.False(reader.TryRead(out readResult));
+    }
+
+    [Fact]
+    public async Task TryRead_AdvanceTo_AfterEndReached()
+    {
+        var reader = this.CreatePipeReader(new MemoryStream(new byte[] { 1, 2, 3 }));
+
+        var readResult = await reader.ReadAsync(this.TimeoutToken);
+        reader.AdvanceTo(readResult.Buffer.End);
+
+        reader.TryRead(out readResult);
+        reader.AdvanceTo(readResult.Buffer.End);
+    }
+
+    [Theory, PairwiseData]
+    public async Task ReadAsync_TwiceInARow(bool emptyStream)
+    {
+        var stream = new MemoryStream(emptyStream ? Array.Empty<byte>() : new byte[3]);
+        var reader = this.CreatePipeReader(stream, sizeHint: 50);
+        var result1 = await reader.ReadAsync(this.TimeoutToken);
+        if (emptyStream)
+        {
+            Assert.True(result1.IsCompleted);
+        }
+
+        if (this.EmulatePipelinesStreamPipeReader)
+        {
+            var result2 = await reader.ReadAsync(this.TimeoutToken);
+            Assert.Equal(result1.Buffer.Length, result2.Buffer.Length);
+            Assert.Equal(emptyStream, result2.IsCompleted);
+        }
+        else
+        {
+            await Assert.ThrowsAsync<InvalidOperationException>(async () => await reader.ReadAsync(this.TimeoutToken));
+        }
+    }
+
+    [Theory, PairwiseData]
+    public async Task TryRead_TwiceInARow(bool emptyStream)
+    {
+        var stream = new MemoryStream(emptyStream ? Array.Empty<byte>() : new byte[3]);
+        var reader = this.CreatePipeReader(stream, sizeHint: 50);
+
+        // Read asynchronously once to guarantee that there's a non-empty buffer in the reader.
+        var readResult = await reader.ReadAsync(this.TimeoutToken);
+        reader.AdvanceTo(readResult.Buffer.Start);
+
+        Assert.Equal(!emptyStream || !this.EmulatePipelinesStreamPipeReader, reader.TryRead(out _));
+        if (this.EmulatePipelinesStreamPipeReader)
+        {
+            Assert.Equal(!emptyStream, reader.TryRead(out _));
+        }
+        else
+        {
+            Assert.Throws<InvalidOperationException>(() => reader.TryRead(out _));
+        }
+    }
+
+    [Fact]
+    public void AdvanceTo_BeforeRead()
+    {
+        var stream = new MemoryStream();
+        var reader = this.CreatePipeReader(stream, sizeHint: 50);
+        if (this.EmulatePipelinesStreamPipeReader)
+        {
+            reader.AdvanceTo(default);
+        }
+        else
+        {
+            Assert.Throws<InvalidOperationException>(() => reader.AdvanceTo(default));
+        }
+
+        var ex = Assert.ThrowsAny<Exception>(() => reader.AdvanceTo(ReadOnlySequence<byte>.Empty.Start));
+        Assert.True(ex is InvalidCastException || ex is InvalidOperationException);
+    }
+
+    [SkippableFact]
     public async Task OnWriterCompleted()
     {
+        Skip.If(this is IOPipelinesStreamPipeReaderTests, "OnWriterCompleted isn't supported.");
         byte[] expectedBuffer = this.GetRandomBuffer(50);
         var stream = new MemoryStream(expectedBuffer);
         var reader = this.CreatePipeReader(stream, sizeHint: 50);
@@ -184,19 +282,50 @@ public async Task CancelPendingRead_WithCancellationToken()
     }
 
     [Fact]
-    public async Task Complete_DoesNotCauseStreamDisposal()
+    public async Task Complete_MayCauseStreamDisposal()
     {
         var stream = new HalfDuplexStream();
-        var reader = this.CreatePipeReader(stream);
+        var ms = new MonitoringStream(stream);
+        var disposal = new AsyncManualResetEvent();
+        ms.Disposed += (s, e) => disposal.Set();
+        var reader = this.CreatePipeReader(ms);
         reader.Complete();
 
-        var timeout = ExpectedTimeoutToken;
-        while (!stream.IsDisposed && !timeout.IsCancellationRequested)
+        if (this is IOPipelinesStreamPipeReaderTests)
+        {
+            await disposal.WaitAsync(this.TimeoutToken);
+        }
+        else
         {
-            await Task.Yield();
+            await Assert.ThrowsAsync<OperationCanceledException>(() => disposal.WaitAsync(ExpectedTimeoutToken));
         }
+    }
+
+    [Fact]
+    public async Task CancelPendingRead()
+    {
+        var stream = new HalfDuplexStream();
+        var reader = this.CreatePipeReader(stream, sizeHint: 50);
 
-        Assert.False(stream.IsDisposed);
+        ValueTask<ReadResult> readTask = reader.ReadAsync(this.TimeoutToken);
+        reader.CancelPendingRead();
+        var readResult = await readTask.AsTask().WithCancellation(this.TimeoutToken);
+        Assert.True(readResult.IsCanceled);
+
+        // Verify we can read after that without cancellation.
+        readTask = reader.ReadAsync(this.TimeoutToken);
+        stream.Write(new byte[] { 1, 2, 3 }, 0, 3);
+        await stream.FlushAsync(this.TimeoutToken);
+        readResult = await readTask;
+        Assert.False(readResult.IsCanceled);
+        Assert.Equal(3, readResult.Buffer.Length);
+        reader.AdvanceTo(readResult.Buffer.End);
+
+        // Now cancel again
+        readTask = reader.ReadAsync(this.TimeoutToken);
+        reader.CancelPendingRead();
+        readResult = await readTask;
+        Assert.True(readResult.IsCanceled);
     }
 
     protected abstract PipeReader CreatePipeReader(Stream stream, int sizeHint = 0);

src/Nerdbank.Streams.Tests/StreamUsePipeReaderTests.cs

@@ -22,6 +22,8 @@ public StreamUsePipeReaderTests(ITestOutputHelper logger)
     {
     }
 
+    protected override bool EmulatePipelinesStreamPipeReader => false;
+
     [Fact]
     public async Task StreamFails()
     {
@@ -51,5 +53,14 @@ public async Task Complete_CausesWriterCompletion()
         await writerCompletion.WithCancellation(this.TimeoutToken);
     }
 
+    [Fact]
+    public void NonReadableStream()
+    {
+        var unreadableStream = new Mock<Stream>(MockBehavior.Strict);
+        unreadableStream.SetupGet(s => s.CanRead).Returns(false);
+        Assert.Throws<ArgumentException>(() => this.CreatePipeReader(unreadableStream.Object));
+        unreadableStream.VerifyAll();
+    }
+
     protected override PipeReader CreatePipeReader(Stream stream, int hintSize = 0) => stream.UsePipeReader(hintSize);
 }

src/Nerdbank.Streams.Tests/StreamUseStrictPipeReaderTests.cs

@@ -41,33 +41,5 @@ public async Task StreamFails()
         Assert.Same(expectedException, actualException);
     }
 
-    [Fact] // Bizarre behavior when using the built-in Pipe class: https://github.com/dotnet/corefx/issues/31696
-    public async Task CancelPendingRead()
-    {
-        var stream = new HalfDuplexStream();
-        var reader = this.CreatePipeReader(stream, sizeHint: 50);
-
-        ValueTask<ReadResult> readTask = reader.ReadAsync(this.TimeoutToken);
-        reader.CancelPendingRead();
-        var readResult = await readTask.AsTask().WithCancellation(this.TimeoutToken);
-        Assert.True(readResult.IsCanceled);
-        ////reader.AdvanceTo(readResult.Buffer.End);
-
-        // Verify we can read after that without cancellation.
-        readTask = reader.ReadAsync(this.TimeoutToken);
-        stream.Write(new byte[] { 1, 2, 3 }, 0, 3);
-        await stream.FlushAsync(this.TimeoutToken);
-        readResult = await readTask;
-        Assert.False(readResult.IsCanceled);
-        Assert.Equal(3, readResult.Buffer.Length);
-        reader.AdvanceTo(readResult.Buffer.End);
-
-        // Now cancel again
-        readTask = reader.ReadAsync(this.TimeoutToken);
-        reader.CancelPendingRead();
-        readResult = await readTask;
-        Assert.True(readResult.IsCanceled);
-    }
-
     protected override PipeReader CreatePipeReader(Stream stream, int sizeHint = 0) => stream.UseStrictPipeReader(sizeHint);
 }

src/Nerdbank.Streams/PipeExtensions.cs

@@ -78,6 +78,7 @@ public static PipeReader UsePipeReader(this Stream stream, int sizeHint = 0, Pip
 
         /// <summary>
         /// Creates a <see cref="PipeReader"/> that reads from the specified <see cref="Stream"/> exactly as told to do so.
+        /// It does *not* close the <paramref name="stream"/> when completed.
         /// </summary>
         /// <param name="stream">The stream to read from using a pipe.</param>
         /// <param name="sizeHint">A hint at the size of messages that are commonly transferred. Use 0 for a commonly reasonable default.</param>
@@ -91,7 +92,7 @@ public static PipeReader UseStrictPipeReader(this Stream stream, int sizeHint =
             Requires.NotNull(stream, nameof(stream));
             Requires.Argument(stream.CanRead, nameof(stream), "Stream must be readable.");
 
-            return new StreamPipeReader(stream, sizeHint);
+            return new StreamPipeReader(stream, sizeHint, leaveOpen: true);
         }
 
         /// <summary>

src/Nerdbank.Streams/StreamPipeReader.cs

@@ -4,6 +4,7 @@
 namespace Nerdbank.Streams
 {
     using System;
+    using System.Buffers;
     using System.Collections.Generic;
     using System.IO;
     using System.IO.Pipelines;
@@ -19,10 +20,17 @@ internal class StreamPipeReader : PipeReader
     {
         private readonly object syncObject = new object();
 
+        /// <summary><inheritdoc cref="StreamPipeReader(Stream, int, bool)" path="/param[@name='stream']"/></summary>
         private readonly Stream stream;
 
+        /// <summary>
+        /// May be 0 for a reasonable default as determined by the <see cref="IBufferWriter{T}.GetMemory"/> method.
+        /// </summary>
         private readonly int bufferSize;
 
+        /// <summary><inheritdoc cref="StreamPipeReader(Stream, int, bool)" path="/param[@name='leaveOpen']"/></summary>
+        private readonly bool leaveOpen;
+
         private readonly Sequence<byte> buffer = new Sequence<byte>();
 
         private SequencePosition examined;
@@ -39,12 +47,19 @@ internal class StreamPipeReader : PipeReader
 
         private List<(Action<Exception?, object?>, object?)>? writerCompletedCallbacks;
 
-        internal StreamPipeReader(Stream stream, int bufferSize)
+        /// <summary>
+        /// Initializes a new instance of the <see cref="StreamPipeReader"/> class.
+        /// </summary>
+        /// <param name="stream">The stream to read from.</param>
+        /// <param name="bufferSize">A hint at the size of messages that are commonly transferred. Use 0 for a commonly reasonable default.</param>
+        /// <param name="leaveOpen"><c>true</c> to leave the underlying <paramref name="stream"/> open after calling <see cref="PipeReader.Complete(Exception)"/>; <c>false</c> to close the stream.</param>
+        internal StreamPipeReader(Stream stream, int bufferSize, bool leaveOpen)
         {
             Requires.NotNull(stream, nameof(stream));
             Requires.Argument(stream.CanRead, nameof(stream), "Stream must be readable.");
             this.stream = stream;
             this.bufferSize = bufferSize;
+            this.leaveOpen = leaveOpen;
         }
 
         /// <inheritdoc />
@@ -71,6 +86,10 @@ public override void Complete(Exception? exception = null)
                 this.isReaderCompleted = true;
                 this.readerException = exception;
                 this.buffer.Reset();
+                if (!this.leaveOpen)
+                {
+                    this.stream.Dispose();
+                }
             }
         }
 

src/shipping.ruleset

@@ -70,6 +70,7 @@
   </Rules>
   <Rules AnalyzerId="StyleCop.Analyzers" RuleNamespace="StyleCop.Analyzers">
     <Rule Id="SA1130" Action="None" />
+    <Rule Id="SA1133" Action="Hidden" />
     <Rule Id="SA1600" Action="Hidden" />
   </Rules>
   <Rules AnalyzerId="Microsoft.VisualStudio.Threading.Analyzers" RuleNamespace="Microsoft.VisualStudio.Threading.Analyzers">