Add test for design for communicating exceptions across channels

Test for #299

src/Nerdbank.Streams.Tests/MultiplexingStreamTests.cs

@@ -3,12 +3,11 @@
 
 using System;
 using System.Buffers;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
 using System.IO.Pipelines;
 using System.Linq;
-using System.Runtime.CompilerServices;
-using System.Runtime.ExceptionServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft;
@@ -583,6 +582,39 @@ public async Task DisposeChannel_WhileRemoteEndIsTransmitting(int length)
         await writerTask;
     }
 
+    /// <summary>
+    /// Verifies that when a channel's writer is completed with a fault, that fault is communicated to the remote reader.
+    /// </summary>
+    [Theory]
+    [InlineData(typeof(InvalidOperationException))]
+    [InlineData(typeof(Exception))]
+    [InlineData(typeof(KeyNotFoundException))]
+    public async Task UserWriteFaultCommunicatedToUserReader(Type faultType)
+    {
+        var (ch1, ch2) = await this.EstablishChannelsAsync(string.Empty);
+        byte[] buffer = this.GetRandomBuffer(3);
+        await ch1.Output.WriteAsync(buffer, this.TimeoutToken);
+        Exception fault = (Exception)Activator.CreateInstance(faultType, "some exception")!;
+        await ch1.Output.CompleteAsync(fault);
+
+        // Enter a read loop since it's undefined exactly which read will throw.
+        try
+        {
+            while (true)
+            {
+                ReadResult readResult = await ch2.Input.ReadAsync(this.TimeoutToken);
+                ch2.Input.AdvanceTo(readResult.Buffer.End);
+
+                // If reading completed without throwing, that's a failure.
+                Assert.False(readResult.IsCompleted);
+            }
+        }
+        catch (RemoteChannelException wrapperException)
+        {
+            Assert.Same(fault, wrapperException.InnerException);
+        }
+    }
+
     [Fact]
     public async Task WriteLargeBuffer()
     {

src/Nerdbank.Streams/RemoteChannelException.cs

@@ -0,0 +1,48 @@
+// Copyright (c) Andrew Arnott. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Nerdbank.Streams
+{
+    using System;
+    using System.IO.Pipelines;
+    using System.Threading;
+
+    /// <summary>
+    /// An exception thrown from <see cref="PipeReader.ReadAsync(CancellationToken)"/>
+    /// when called on the <see cref="MultiplexingStream.Channel.Input"/> property from a
+    /// <see cref="MultiplexingStream.Channel"/> whose remote counterpart completed
+    /// their <see cref="MultiplexingStream.Channel.Output"/> with a fault
+    /// using <see cref="PipeWriter.CompleteAsync(System.Exception?)"/>.
+    /// </summary>
+    [Serializable]
+    public class RemoteChannelException : Exception
+    {
+        /// <summary>Initializes a new instance of the <see cref="RemoteChannelException"/> class.</summary>
+        public RemoteChannelException()
+        {
+        }
+
+        /// <summary>Initializes a new instance of the <see cref="RemoteChannelException"/> class.</summary>
+        /// <inheritdoc cref="Exception(string)"/>
+        public RemoteChannelException(string message)
+            : base(message)
+        {
+        }
+
+        /// <summary>Initializes a new instance of the <see cref="RemoteChannelException"/> class.</summary>
+        /// <inheritdoc cref="Exception(string, Exception)"/>
+        public RemoteChannelException(string message, Exception inner)
+            : base(message, inner)
+        {
+        }
+
+        /// <summary>Initializes a new instance of the <see cref="RemoteChannelException"/> class.</summary>
+        /// <inheritdoc cref="Exception(System.Runtime.Serialization.SerializationInfo, System.Runtime.Serialization.StreamingContext)"/>
+        protected RemoteChannelException(
+          System.Runtime.Serialization.SerializationInfo info,
+          System.Runtime.Serialization.StreamingContext context)
+            : base(info, context)
+        {
+        }
+    }
+}

src/Nerdbank.Streams/netcoreapp2.1/PublicAPI.Unshipped.txt

@@ -1,3 +1,8 @@
 Nerdbank.Streams.MultiplexingStream.QualifiedChannelId.QualifiedChannelId() -> void
+Nerdbank.Streams.RemoteChannelException
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException() -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message, System.Exception! inner) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(System.Runtime.Serialization.SerializationInfo! info, System.Runtime.Serialization.StreamingContext context) -> void
 static Nerdbank.Streams.PipeExtensions.AsPrebufferedStreamAsync(this System.IO.Pipelines.PipeReader! pipeReader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<System.IO.Stream!>!
 static Nerdbank.Streams.StreamExtensions.AsStream(this System.Buffers.ReadOnlySequence<byte> readOnlySequence, System.Action<object?>? disposeAction, object? disposeActionArg) -> System.IO.Stream!

src/Nerdbank.Streams/netcoreapp3.1/PublicAPI.Unshipped.txt

@@ -1,3 +1,8 @@
 Nerdbank.Streams.MultiplexingStream.QualifiedChannelId.QualifiedChannelId() -> void
+Nerdbank.Streams.RemoteChannelException
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException() -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message, System.Exception! inner) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(System.Runtime.Serialization.SerializationInfo! info, System.Runtime.Serialization.StreamingContext context) -> void
 static Nerdbank.Streams.PipeExtensions.AsPrebufferedStreamAsync(this System.IO.Pipelines.PipeReader! pipeReader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<System.IO.Stream!>!
 static Nerdbank.Streams.StreamExtensions.AsStream(this System.Buffers.ReadOnlySequence<byte> readOnlySequence, System.Action<object?>? disposeAction, object? disposeActionArg) -> System.IO.Stream!

src/Nerdbank.Streams/netstandard2.0/PublicAPI.Unshipped.txt

@@ -1,3 +1,8 @@
 Nerdbank.Streams.MultiplexingStream.QualifiedChannelId.QualifiedChannelId() -> void
+Nerdbank.Streams.RemoteChannelException
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException() -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(string! message, System.Exception! inner) -> void
+Nerdbank.Streams.RemoteChannelException.RemoteChannelException(System.Runtime.Serialization.SerializationInfo! info, System.Runtime.Serialization.StreamingContext context) -> void
 static Nerdbank.Streams.PipeExtensions.AsPrebufferedStreamAsync(this System.IO.Pipelines.PipeReader! pipeReader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<System.IO.Stream!>!
 static Nerdbank.Streams.StreamExtensions.AsStream(this System.Buffers.ReadOnlySequence<byte> readOnlySequence, System.Action<object?>? disposeAction, object? disposeActionArg) -> System.IO.Stream!