Only cancel LinkToAsync in one direction

This avoids data loss when pipes complete while others are still propagating received data.

Author: AArnott

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -112,11 +112,17 @@ public class Channel : IDisposableObservable, IDuplexPipe
             private PipeWriter? mxStreamIOWriter;
 
             /// <summary>
-            /// The I/O to expose on this channel. Will be <c>null</c> if <see cref="ChannelOptions.ExistingPipe"/>
-            /// was set to a non-null value when this channel was created.
+            /// The I/O to expose on this channel if <see cref="ChannelOptions.ExistingPipe"/> was not specified;
+            /// otherwise it is the buffering pipe we use as an intermediary with the specified <see cref="ChannelOptions.ExistingPipe"/>.
             /// </summary>
             private IDuplexPipe? channelIO;
 
+            /// <summary>
+            /// The value of <see cref="ChannelOptions.ExistingPipe"/> as it was when we received it.
+            /// We don't use this field, but we set it for diagnostic purposes later.
+            /// </summary>
+            private IDuplexPipe? existingPipe;
+
             /// <summary>
             /// A value indicating whether this <see cref="Channel"/> was created or accepted with a non-null value for <see cref="ChannelOptions.ExistingPipe"/>.
             /// </summary>
@@ -585,8 +591,14 @@ private void ApplyChannelOptions(ChannelOptions channelOptions)
                         if (channelOptions.ExistingPipe is object)
                         {
                             Assumes.NotNull(this.channelIO);
-                            this.DisposeSelfOnFailure(this.channelIO.LinkToAsync(channelOptions.ExistingPipe, this.DisposalToken));
+                            this.existingPipe = channelOptions.ExistingPipe;
                             this.existingPipeGiven = true;
+
+                            // We always want to write ALL received data to the user's ExistingPipe, rather than truncating it on disposal, so don't use a cancellation token in that direction.
+                            this.DisposeSelfOnFailure(this.channelIO.Input.LinkToAsync(channelOptions.ExistingPipe.Output));
+
+                            // Upon disposal, we no longer want to continue reading from the user's ExistingPipe into our buffer since we won't be propagating it any further, so use our DisposalToken.
+                            this.DisposeSelfOnFailure(channelOptions.ExistingPipe.Input.LinkToAsync(this.channelIO.Output, this.DisposalToken));
                         }
                         else
                         {