Clean up pipe pair on transport start failure (#1836)

Author: mikaelm12

src/Microsoft.AspNetCore.Http.Connections.Client/HttpConnection.cs

@@ -345,6 +345,12 @@ private async Task StartTransport(Uri connectUrl, HttpTransportType transportTyp
             catch (Exception ex)
             {
                 Log.ErrorStartingTransport(_logger, transport, ex);
+
+                // Clean up pipes and null out transport when we fail to start.
+                pair.Transport.Input.Complete();
+                pair.Transport.Output.Complete();
+                pair.Application.Input.Complete();
+                pair.Application.Output.Complete();
                 _transport = null;
                 throw;
             }

test/Microsoft.AspNetCore.SignalR.Client.Tests/HttpConnectionTests.ConnectionLifecycle.cs

@@ -159,6 +159,28 @@ await WithConnectionAsync(
                 }
             }
 
+            [Fact]
+            public async Task PipesAreDisposedAfterTransportFailsToStart()
+            {
+                using (StartLog(out var loggerFactory))
+                {
+                    var writerTcs = new TaskCompletionSource<object>();
+                    var readerTcs = new TaskCompletionSource<object>();
+                    await WithConnectionAsync(
+                        CreateConnection(
+                            loggerFactory: loggerFactory,
+                            transport: new FakeTransport(writerTcs, readerTcs)),
+                        async (connection) =>
+                        {
+                            var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => connection.StartAsync(TransferFormat.Text));
+                            Assert.Equal("Unable to connect to the server with any of the available transports.", ex.Message);
+
+                            Assert.True(writerTcs.Task.IsCompleted);
+                            Assert.True(readerTcs.Task.IsCompleted);
+                        });
+                }
+            }
+
             [Fact]
             public async Task CanDisposeUnstartedConnection()
             {
@@ -356,6 +378,35 @@ await WithConnectionAsync(
                 }
             }
 
+            private class FakeTransport : ITransport
+            {
+                private IDuplexPipe _application;
+                private TaskCompletionSource<object> _writerTcs;
+                private TaskCompletionSource<object> _readerTcs;
+
+                public FakeTransport(TaskCompletionSource<object> writerTcs, TaskCompletionSource<object> readerTcs)
+                {
+                    _writerTcs = writerTcs;
+                    _readerTcs = readerTcs;
+                }
+
+                public Task StartAsync(Uri url, IDuplexPipe application, TransferFormat transferFormat, IConnection connection)
+                {
+                    _application = application;
+                    Action<Exception, object> onCompletedCallback = (ex, tcs) => { ((TaskCompletionSource<object>)tcs).TrySetResult(null); };
+                    _application.Input.OnWriterCompleted(onCompletedCallback, _writerTcs);
+                    _application.Output.OnReaderCompleted(onCompletedCallback, _readerTcs);
+                    throw new Exception();
+                }
+
+                public Task StopAsync()
+                {
+                    _application.Output.Complete();
+                    _application.Input.Complete();
+                    return Task.CompletedTask;
+                }
+            }
+
             private static async Task AssertDisposedAsync(HttpConnection connection)
             {
                 var exception =