Merge pull request #197 from AArnott/closeStreamWithMxChannel

Dispose of streams used as mxstream channels when channels close

Author: AArnott

src/Nerdbank.Streams.Tests/MultiplexingStreamTests.cs

@@ -205,6 +205,33 @@ public async Task Dispose_DisposesChannels()
 #pragma warning restore CS0618 // Type or member is obsolete
     }
 
+    [Fact]
+    public async Task ChannelDispose_ClosesExistingStream()
+    {
+        var ms = new MonitoringStream(new MemoryStream());
+        var disposal = new AsyncManualResetEvent();
+        ms.Disposed += (s, e) => disposal.Set();
+
+        var channel = this.mx1.CreateChannel(new MultiplexingStream.ChannelOptions { ExistingPipe = ms.UsePipe() });
+        channel.Dispose();
+        await disposal.WaitAsync(this.TimeoutToken);
+    }
+
+    [Fact]
+    public async Task RemoteChannelClose_ClosesExistingStream()
+    {
+        var ms = new MonitoringStream(new MemoryStream());
+        var disposal = new AsyncManualResetEvent();
+        ms.Disposed += (s, e) => disposal.Set();
+
+        var ch1 = this.mx1.CreateChannel(new MultiplexingStream.ChannelOptions { ExistingPipe = ms.UsePipe() });
+        await this.WaitForEphemeralChannelOfferToPropagateAsync();
+        var ch2 = this.mx2.AcceptChannel(ch1.Id);
+
+        ch2.Dispose();
+        await disposal.WaitAsync(this.TimeoutToken);
+    }
+
     [Fact]
     public async Task CreateChannelAsync_ThrowsAfterDisposal()
     {

src/Nerdbank.Streams/PipeExtensions.cs

@@ -414,10 +414,13 @@ internal static Task LinkToAsync(this PipeReader reader, PipeWriter writer, bool
                             break;
                         }
                     }
+
+                    await reader.CompleteAsync().ConfigureAwait(false);
                 }
                 catch (Exception ex)
                 {
                     await writer.CompleteAsync(ex).ConfigureAwait(false);
+                    await reader.CompleteAsync(ex).ConfigureAwait(false);
                 }
             });
         }