Merge pull request #101 from AArnott/dev/andarno/FixDisposal

Fix MultiplexingStream disposal

Author: AArnott

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -303,8 +303,23 @@ internal void OnContentWritingCompleted()
             {
                 this.DisposeSelfOnFailure(Task.Run(async delegate
                 {
-                    var writer = await this.GetReceivedMessagePipeWriterAsync().ConfigureAwait(false);
-                    writer.Complete();
+                    if (!this.IsDisposed)
+                    {
+                        try
+                        {
+                            var writer = await this.GetReceivedMessagePipeWriterAsync().ConfigureAwait(false);
+                            writer.Complete();
+                        }
+                        catch (ObjectDisposedException)
+                        {
+                            this.mxStreamIOWriter?.Complete();
+                        }
+                    }
+                    else
+                    {
+                        this.mxStreamIOWriter?.Complete();
+                    }
+
                     this.mxStreamIOWriterCompleted.Set();
                 }));
             }

src/Nerdbank.Streams/MultiplexingStream.cs

@@ -707,58 +707,23 @@ private async Task ReadStreamAsync()
                             break;
                     }
                 }
-
-                await this.CompleteWritingOnAllChannelsAsync().ConfigureAwait(false);
             }
             catch (EndOfStreamException)
             {
                 // When we unexpectedly hit an end of stream, just close up shop.
-                await this.CompleteWritingOnAllChannelsAsync().ConfigureAwait(false);
-            }
-            catch (OperationCanceledException)
-            {
-                // We don't need to propagate a cancellation exception either.
-                await this.CompleteWritingOnAllChannelsAsync().ConfigureAwait(false);
             }
-            catch (Exception ex)
-            {
-                await this.CompleteWritingOnAllChannelsAsync(ex).ConfigureAwait(false);
-                throw;
-            }
-
-            this.Dispose();
-        }
-
-        private async Task CompleteWritingOnAllChannelsAsync(Exception ex = null)
-        {
-            ValueTask<PipeWriter>[] pipeWriters = null;
-            try
+            finally
             {
-                int writersCount;
                 lock (this.syncObject)
                 {
-                    writersCount = this.openChannels.Count;
-                    pipeWriters = ArrayPool<ValueTask<PipeWriter>>.Shared.Rent(writersCount);
-                    int i = 0;
                     foreach (var entry in this.openChannels)
                     {
-                        pipeWriters[i++] = entry.Value.GetReceivedMessagePipeWriterAsync();
+                        entry.Value.OnContentWritingCompleted();
                     }
                 }
-
-                for (int i = 0; i < writersCount; i++)
-                {
-                    var writer = await pipeWriters[i].ConfigureAwait(false);
-                    writer.Complete(ex);
-                }
-            }
-            finally
-            {
-                if (pipeWriters != null)
-                {
-                    ArrayPool<ValueTask<PipeWriter>>.Shared.Return(pipeWriters);
-                }
             }
+
+            this.Dispose();
         }
 
         /// <summary>