Merge pull request #330 from AArnott/fix252

Avoid 1st chance exceptions in one-way mxstream channels

Author: AArnott

src/Nerdbank.Streams.Tests/MultiplexingStreamTests.cs

@@ -388,6 +388,23 @@ public async Task CommunicateOverOneChannel()
         await this.TransmitAndVerifyAsync(b, a, Guid.NewGuid().ToByteArray());
     }
 
+    [Fact]
+    public async Task ChannelWithExistingSimplexChannel()
+    {
+        var transmittingPipe = new Pipe();
+        var receivingPipe = new Pipe();
+        var mx1ChannelTask = this.mx1.OfferChannelAsync(string.Empty, new MultiplexingStream.ChannelOptions { ExistingPipe = new DuplexPipe(transmittingPipe.Reader) }, this.TimeoutToken);
+        var mx2ChannelTask = this.mx2.AcceptChannelAsync(string.Empty, new MultiplexingStream.ChannelOptions { ExistingPipe = new DuplexPipe(receivingPipe.Writer) }, this.TimeoutToken);
+        var channels = await WhenAllSucceedOrAnyFail(mx1ChannelTask, mx2ChannelTask).WithCancellation(this.TimeoutToken);
+
+        var buffer = this.GetBuffer(3);
+        await transmittingPipe.Writer.WriteAsync(buffer, this.TimeoutToken);
+
+        ReadOnlySequence<byte> readBytes = await this.ReadAtLeastAsync(receivingPipe.Reader, buffer.Length);
+        Assert.Equal(buffer.Length, readBytes.Length);
+        Assert.Equal(buffer, readBytes.ToArray());
+    }
+
     [Fact]
     [Trait("SkipInCodeCoverage", "true")] // far too slow and times out
     [Trait("Stress", "true")]

src/Nerdbank.Streams/DuplexPipe.cs

@@ -50,6 +50,16 @@ public DuplexPipe(PipeWriter output)
         /// <inheritdoc />
         public PipeWriter Output { get; }
 
+        /// <summary>
+        /// Checks whether the given <see cref="PipeReader"/> is known to be already completed.
+        /// </summary>
+        /// <param name="reader">The reader to check.</param>
+        /// <returns><see langword="true" /> if the <paramref name="reader"/> is known to be completed; <see langword="false"/> if the <paramref name="reader"/> is not known to be completed.</returns>
+        /// <remarks>
+        /// This method <em>may</em> return <see langword="false"/> for a completed <see cref="PipeReader"/>.
+        /// </remarks>
+        internal static bool IsDefinitelyCompleted(PipeReader reader) => reader is CompletedPipeReader;
+
         private class CompletedPipeWriter : PipeWriter
         {
             internal static readonly PipeWriter Singleton = new CompletedPipeWriter();

src/Nerdbank.Streams/PipeExtensions.cs

@@ -435,6 +435,12 @@ internal static Task LinkToAsync(this PipeReader reader, PipeWriter writer, Canc
             {
                 try
                 {
+                    if (DuplexPipe.IsDefinitelyCompleted(reader))
+                    {
+                        await writer.CompleteAsync().ConfigureAwait(false);
+                        return;
+                    }
+
                     while (true)
                     {
                         cancellationToken.ThrowIfCancellationRequested();