Cancel LinkToAsync when Channel shuts down

This will add assurance that streams/pipes that are connected to the Channel will shutdown because their reader/writer has completed.

Author: AArnott

src/Nerdbank.Streams.Tests/MultiplexingStreamTests.cs

@@ -208,7 +208,7 @@ public async Task Dispose_DisposesChannels()
     [Fact]
     public async Task ChannelDispose_ClosesExistingStream()
     {
-        var ms = new MonitoringStream(new MemoryStream());
+        var ms = new MonitoringStream(FullDuplexStream.CreatePair().Item1);
         var disposal = new AsyncManualResetEvent();
         ms.Disposed += (s, e) => disposal.Set();
 
@@ -220,7 +220,7 @@ public async Task ChannelDispose_ClosesExistingStream()
     [Fact]
     public async Task RemoteChannelClose_ClosesExistingStream()
     {
-        var ms = new MonitoringStream(new MemoryStream());
+        var ms = new MonitoringStream(FullDuplexStream.CreatePair().Item1);
         var disposal = new AsyncManualResetEvent();
         ms.Disposed += (s, e) => disposal.Set();
 

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -38,6 +38,11 @@ public class Channel : IDisposableObservable, IDuplexPipe
             /// </summary>
             private readonly TaskCompletionSource<object?> completionSource = new TaskCompletionSource<object?>();
 
+            /// <summary>
+            /// The source for a token that will be canceled when this channel has completed.
+            /// </summary>
+            private readonly CancellationTokenSource disposalTokenSource = new CancellationTokenSource();
+
             /// <summary>
             /// The source for the <see cref="OptionsApplied"/> property. May be null if options were provided in ctor.
             /// </summary>
@@ -229,6 +234,11 @@ public PipeWriter Output
             /// </summary>
             public MultiplexingStream MultiplexingStream { get; }
 
+            /// <summary>
+            /// Gets a token that is canceled just before <see cref="Completion" /> has transitioned to its final state.
+            /// </summary>
+            internal CancellationToken DisposalToken => this.disposalTokenSource.Token;
+
             internal OfferParameters OfferParams { get; }
 
             internal string Name => this.OfferParams.Name;
@@ -297,6 +307,7 @@ public void Dispose()
                     Action<object?, object> finalDisposalAction = (exOrAntecedent, state) =>
                     {
                         var self = (Channel)state;
+                        self.disposalTokenSource.Cancel();
                         self.completionSource.TrySetResult(null);
                         self.MultiplexingStream.OnChannelDisposed(self);
                     };
@@ -574,7 +585,7 @@ private void ApplyChannelOptions(ChannelOptions channelOptions)
                         if (channelOptions.ExistingPipe is object)
                         {
                             Assumes.NotNull(this.channelIO);
-                            this.DisposeSelfOnFailure(this.channelIO.LinkToAsync(channelOptions.ExistingPipe));
+                            this.DisposeSelfOnFailure(this.channelIO.LinkToAsync(channelOptions.ExistingPipe, this.DisposalToken));
                             this.existingPipeGiven = true;
                         }
                         else