WIP: HttpsConnectionMiddleware

src/Kestrel.Core/Adapter/Internal/AdaptedPipeline.cs

@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.IO.Pipelines;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
@@ -12,16 +13,16 @@
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
 {
-    public class AdaptedPipeline : IDuplexPipe
+    public class AdaptedPipeline : IDuplexPipe, IDisposable
     {
         private static readonly int MinAllocBufferSize = KestrelMemoryPool.MinimumSegmentSize / 2;
 
         private readonly IDuplexPipe _transport;
 
-        public AdaptedPipeline(IDuplexPipe transport,
-                               Pipe inputPipe,
+        public AdaptedPipeline(Pipe inputPipe,
                                Pipe outputPipe,
-                               IKestrelTrace log)
+                               ILogger log,
+                               IDuplexPipe transport)
         {
             _transport = transport;
             Input = inputPipe;
@@ -33,7 +34,7 @@ public AdaptedPipeline(IDuplexPipe transport,
 
         public Pipe Output { get; }
 
-        public IKestrelTrace Log { get; }
+        public ILogger Log { get; }
 
         PipeReader IDuplexPipe.Input => Input.Reader;
 
@@ -64,6 +65,11 @@ private async Task WriteOutputAsync(Stream stream)
 
                     try
                     {
+                        if (result.IsCanceled)
+                        {
+                            break;
+                        }
+
                         if (buffer.IsEmpty)
                         {
                             if (result.IsCompleted)
@@ -111,7 +117,11 @@ private async Task WriteOutputAsync(Stream stream)
             finally
             {
                 Output.Reader.Complete();
+
                 _transport.Output.Complete();
+
+                // Cancel any pending flushes due to back-pressure
+                Input.Writer.CancelPendingFlush();
             }
         }
 
@@ -149,7 +159,7 @@ private async Task ReadInputAsync(Stream stream)
 
                     var result = await Input.Writer.FlushAsync();
 
-                    if (result.IsCompleted)
+                    if (result.IsCompleted || result.IsCanceled)
                     {
                         break;
                     }
@@ -163,10 +173,20 @@ private async Task ReadInputAsync(Stream stream)
             finally
             {
                 Input.Writer.Complete(error);
+
                 // The application could have ended the input pipe so complete
                 // the transport pipe as well
                 _transport.Input.Complete();
+
+                // Cancel any pending reads from the application
+                Output.Reader.CancelPendingRead();
             }
         }
+
+        public void Dispose()
+        {
+            Input.Reader.Complete();
+            Output.Writer.Complete();
+        }
     }
 }

src/Kestrel.Core/Adapter/Internal/RawStream.cs

@@ -61,18 +61,18 @@ public override int Read(byte[] buffer, int offset, int count)
         {
             // ValueTask uses .GetAwaiter().GetResult() if necessary
             // https://github.com/dotnet/corefx/blob/f9da3b4af08214764a51b2331f3595ffaf162abe/src/System.Threading.Tasks.Extensions/src/System/Threading/Tasks/ValueTask.cs#L156
-            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count)).Result;
+            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count), default).Result;
         }
 
         public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count)).AsTask();
+            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count), cancellationToken).AsTask();
         }
 
 #if NETCOREAPP2_1
         public override ValueTask<int> ReadAsync(Memory<byte> destination, CancellationToken cancellationToken = default)
         {
-            return ReadAsyncInternal(destination);
+            return ReadAsyncInternal(destination, cancellationToken);
         }
 #elif NETSTANDARD2_0
 #else
@@ -115,11 +115,11 @@ public override Task FlushAsync(CancellationToken cancellationToken)
             return WriteAsync(null, 0, 0, cancellationToken);
         }
 
-        private async ValueTask<int> ReadAsyncInternal(Memory<byte> destination)
+        private async ValueTask<int> ReadAsyncInternal(Memory<byte> destination, CancellationToken cancellationToken)
         {
             while (true)
             {
-                var result = await _input.ReadAsync();
+                var result = await _input.ReadAsync(cancellationToken);
                 var readableBuffer = result.Buffer;
                 try
                 {

src/Kestrel.Core/HttpsConnectionAdapterOptions.cs

@@ -2,12 +2,14 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.IO.Pipelines;
 using System.Net.Security;
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Server.Kestrel.Core;
+using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Https
 {
@@ -75,6 +77,12 @@ public HttpsConnectionAdapterOptions()
         /// </summary>
         public bool CheckCertificateRevocation { get; set; }
 
+        internal PipeScheduler Scheduler { get; set; } = PipeScheduler.ThreadPool;
+
+        internal long? MaxInputBufferSize { get; set; }
+
+        internal long? MaxOutputBufferSize { get; set; }
+
         /// <summary>
         /// Specifies the maximum amount of time allowed for the TLS/SSL handshake. This must be positive and finite.
         /// </summary>

src/Kestrel.Core/Internal/AddressBinder.cs

@@ -170,8 +170,7 @@ public async Task BindAsync(AddressBindContext context)
                 var httpsDefault = ParseAddress(Constants.DefaultServerHttpsAddress, out https);
                 context.ServerOptions.ApplyEndpointDefaults(httpsDefault);
 
-                if (httpsDefault.ConnectionAdapters.Any(f => f.IsHttps)
-                    || httpsDefault.TryUseHttps())
+                if (httpsDefault.IsTls || httpsDefault.TryUseHttps())
                 {
                     await httpsDefault.BindAsync(context).ConfigureAwait(false);
                     context.Logger.LogDebug(CoreStrings.BindingToDefaultAddresses,
@@ -254,7 +253,7 @@ public virtual async Task BindAsync(AddressBindContext context)
                     var options = ParseAddress(address, out var https);
                     context.ServerOptions.ApplyEndpointDefaults(options);
 
-                    if (https && !options.ConnectionAdapters.Any(f => f.IsHttps))
+                    if (https && !options.IsTls)
                     {
                         options.UseHttps();
                     }

src/Kestrel.Core/Internal/Http/Http1Connection.cs

@@ -71,7 +71,7 @@ public void Abort(ConnectionAbortedException abortReason)
             }
 
             // Abort output prior to calling OnIOCompleted() to give the transport the chance to complete the input
-            // with the correct error and message. 
+            // with the correct error and message.
             Output.Abort(abortReason);
 
             OnInputOrOutputCompleted();

src/Kestrel.Core/Internal/HttpConnection.cs

@@ -110,10 +110,10 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> http
 
                 if (_context.ConnectionAdapters.Count > 0)
                 {
-                    adaptedPipeline = new AdaptedPipeline(_adaptedTransport,
-                                                          new Pipe(AdaptedInputPipeOptions),
+                    adaptedPipeline = new AdaptedPipeline(new Pipe(AdaptedInputPipeOptions),
                                                           new Pipe(AdaptedOutputPipeOptions),
-                                                          Log);
+                                                          Log,
+                                                          _adaptedTransport);
 
                     _adaptedTransport = adaptedPipeline;
                 }
@@ -639,8 +639,12 @@ private void CloseUninitializedConnection(ConnectionAbortedException abortReason
 
             _context.ConnectionContext.Abort(abortReason);
 
-            _adaptedTransport.Input.Complete();
-            _adaptedTransport.Output.Complete();
+            // Back compat
+            if (_context.ConnectionAdapters.Count > 0)
+            {
+                _adaptedTransport.Input.Complete();
+                _adaptedTransport.Output.Complete();
+            }
         }
 
         private enum ProtocolSelectionState

src/Kestrel.Core/Internal/HttpConnectionMiddleware.cs

@@ -33,8 +33,6 @@ public HttpConnectionMiddleware(IList<IConnectionAdapter> adapters, ServiceConte
 
         public Task OnConnectionAsync(ConnectionContext connectionContext)
         {
-            // We need the transport feature so that we can cancel the output reader that the transport is using
-            // This is a bit of a hack but it preserves the existing semantics
             var memoryPoolFeature = connectionContext.Features.Get<IMemoryPoolFeature>();
 
             var httpConnectionContext = new HttpConnectionContext