wip: cadente/chunked, stability improvements, covering 63/95

Author: CypherPotato

cadente/Sisk.Cadente.CoreEngine/CadenteHttpServerEngineResponse.cs

@@ -8,13 +8,7 @@
 // File name:   CadenteHttpServerEngineResponse.cs
 // Repository:  https://github.com/sisk-http/core
 
-using System;
-using System.Collections.Generic;
 using System.Globalization;
-using System.Linq;
-using System.Net;
-using System.Text;
-using System.Threading.Tasks;
 using Sisk.Core.Http;
 using Sisk.Core.Http.Engine;
 
@@ -27,7 +21,8 @@ public sealed class CadenteHttpServerEngineResponse : HttpServerEngineContextRes
         private readonly HttpHostContext.HttpResponse _response;
         private readonly HttpHostContext _httpHostContext;
         private CadenteHttpServerEngineContext? _context;
-        private WebHeaderCollection _headers = new ();
+        private Lazy<Stream> _outputStream;
+        private CadenteEngineHeaderList _headers;
 
         internal void SetContext ( CadenteHttpServerEngineContext context ) {
             _context = context;
@@ -41,6 +36,8 @@ internal void SetContext ( CadenteHttpServerEngineContext context ) {
         public CadenteHttpServerEngineResponse ( HttpHostContext.HttpResponse response, HttpHostContext httpHostContext ) {
             _response = response;
             _httpHostContext = httpHostContext;
+            _outputStream = new Lazy<Stream> ( () => _response.GetResponseStream ( SendChunked ) );
+            _headers = new CadenteEngineHeaderList ( _response );
         }
 
         /// <inheritdoc/>
@@ -50,14 +47,11 @@ public CadenteHttpServerEngineResponse ( HttpHostContext.HttpResponse response,
         /// <inheritdoc/>
         public override bool KeepAlive { get => _httpHostContext.KeepAlive; set => _httpHostContext.KeepAlive = value; }
         /// <inheritdoc/>
-        public override bool SendChunked { get => _response.SendChunked; set => _response.SendChunked = value; }
+        public override bool SendChunked { get; set; }
         /// <inheritdoc/>
         public override long ContentLength64 {
             get {
-                if (_response.ResponseStream is { CanSeek: true }) {
-                    return _response.ResponseStream.Length;
-                }
-                else if (_response.Headers.FirstOrDefault ( h => h.Name.Equals ( HttpKnownHeaderNames.ContentLength, StringComparison.OrdinalIgnoreCase ) ) is { IsEmpty: false } contentLenHeader) {
+                if (_response.Headers.FirstOrDefault ( h => h.Name.Equals ( HttpKnownHeaderNames.ContentLength, StringComparison.OrdinalIgnoreCase ) ) is { IsEmpty: false } contentLenHeader) {
                     return long.Parse ( contentLenHeader.Value );
                 }
                 else {
@@ -94,10 +88,10 @@ public override string? ContentType {
         }
 
         /// <inheritdoc/>
-        public override Stream OutputStream => _response.GetResponseStream ();
+        public override Stream OutputStream => _outputStream.Value;
 
         /// <inheritdoc/>
-        public override IHttpEngineHeaderList Headers => new CadenteEngineHeaderList ( _response );
+        public override IHttpEngineHeaderList Headers => _headers;
 
         /// <inheritdoc/>
         public override void Abort () {
@@ -111,14 +105,16 @@ public override void AppendHeader ( string name, string value ) {
 
         /// <inheritdoc/>
         public override void Close () {
-            _context?.CompleteProcessing ();
-            _response.ResponseStream?.Close ();
+            Dispose ();
         }
 
         /// <inheritdoc/>
         public override void Dispose () {
+            if (_outputStream.IsValueCreated) {
+                _outputStream.Value.Dispose ();
+            }
+
             _context?.CompleteProcessing ();
-            _response.ResponseStream?.Dispose ();
         }
     }
 }

cadente/Sisk.Cadente/HttpConnection.cs

@@ -9,7 +9,6 @@
 
 using System.Net;
 using Sisk.Cadente.HttpSerializer;
-using Sisk.Cadente.Streams;
 using Sisk.Core.Http;
 
 namespace Sisk.Cadente;
@@ -44,57 +43,27 @@ public async Task<HttpConnectionState> HandleConnectionEvents () {
         while (_connectionStream.CanRead && !disposedValue) {
 
             HttpRequestReader requestReader = new HttpRequestReader ( _connectionStream );
-            Stream? responseStream = null;
-
-            try {
-                if (requestReader.TryReadHttpRequest ( out HttpRequestBase? nextRequest ) == false) {
-                    return HttpConnectionState.ConnectionClosed;
-                }
-
-                HttpHostContext managedSession = new HttpHostContext ( _host, nextRequest, _client, _connectionStream );
-                await _host.InvokeContextCreated ( managedSession );
-
-                if (!managedSession.KeepAlive || !nextRequest.CanKeepAlive) {
-                    connectionCloseRequested = true;
-                    managedSession.Response.Headers.Set ( new HttpHeader ( HttpHeaderName.Connection, "close" ) );
-                }
-
-                if (managedSession.Response.ResponseStream is Stream { } s) {
-                    responseStream = s;
-                }
-                else {
-                    managedSession.Response.Headers.Set ( new HttpHeader ( HttpHeaderName.ContentLength, "0" ) );
-                }
-
-                Stream outputStream = _connectionStream;
-                if (responseStream is not null) {
-
-                    if (managedSession.Response.SendChunked || !responseStream.CanSeek) {
-
-                        managedSession.Response.Headers.Set ( new HttpHeader ( HttpHeaderName.TransferEncoding, "chunked" ) );
-                        responseStream = new HttpChunkedStream ( responseStream );
-                    }
-                    else {
-                        managedSession.Response.Headers.Set ( new HttpHeader ( HttpHeaderName.ContentLength, responseStream.Length.ToString () ) );
-                    }
-                }
-
-                if (managedSession.ResponseHeadersAlreadySent == false && !managedSession.WriteHttpResponseHeaders ()) {
-                    return HttpConnectionState.ResponseWriteException;
-                }
-
-                if (responseStream is not null) {
-                    await responseStream.CopyToAsync ( outputStream );
-                }
-
-                await _connectionStream.FlushAsync ();
-
-                if (connectionCloseRequested) {
-                    break;
-                }
+
+            if (requestReader.TryReadHttpRequest ( out HttpRequestBase? nextRequest ) == false) {
+                return HttpConnectionState.ConnectionClosed;
+            }
+
+            HttpHostContext managedSession = new HttpHostContext ( _host, nextRequest, _client, _connectionStream );
+            await _host.InvokeContextCreated ( managedSession );
+
+            if (!managedSession.KeepAlive || !nextRequest.CanKeepAlive) {
+                connectionCloseRequested = true;
+                managedSession.Response.Headers.Set ( new HttpHeader ( HttpHeaderName.Connection, "close" ) );
             }
-            finally {
-                responseStream?.Dispose ();
+
+            if (managedSession.ResponseHeadersAlreadySent == false && !managedSession.WriteHttpResponseHeaders ()) {
+                return HttpConnectionState.ResponseWriteException;
+            }
+
+            await _connectionStream.FlushAsync ();
+
+            if (connectionCloseRequested) {
+                break;
             }
         }
 

cadente/Sisk.Cadente/HttpHeaderExtensions.cs

@@ -36,4 +36,20 @@ public static void Set ( this List<HttpHeader> headers, in HttpHeader header ) {
             headers.Add ( header );
         }
     }
+
+    /// <summary>
+    /// Removes all <see cref="HttpHeader"/> with the given name from the list. Thread-safe.
+    /// </summary>
+    /// <param name="headers">The list of <see cref="HttpHeader"/> to modify.</param>
+    /// <param name="headerName">The name of the header to remove.</param>
+    public static void Remove ( this List<HttpHeader> headers, string headerName ) {
+        lock (((ICollection) headers).SyncRoot) {
+            var span = CollectionsMarshal.AsSpan ( headers );
+            for (int i = span.Length - 1; i >= 0; i--) {
+                if (Ascii.EqualsIgnoreCase ( span [ i ].NameBytes.Span, headerName )) {
+                    headers.RemoveAt ( i );
+                }
+            }
+        }
+    }
 }

cadente/Sisk.Cadente/HttpHost.cs

@@ -87,7 +87,7 @@ public void Start () {
         ObjectDisposedException.ThrowIf ( disposedValue, this );
 
         _listener.Server.NoDelay = true;
-        _listener.Server.LingerState = new LingerOption ( true, 0 );
+        _listener.Server.LingerState = new LingerOption ( true, 3 );
         _listener.Server.ReceiveBufferSize = HttpConnection.REQUEST_BUFFER_SIZE;
         _listener.Server.SendBufferSize = HttpConnection.RESPONSE_BUFFER_SIZE;
 

cadente/Sisk.Cadente/HttpHostContext.cs

@@ -11,6 +11,7 @@
 using System.Text;
 using Sisk.Cadente.HttpSerializer;
 using Sisk.Cadente.Streams;
+using Sisk.Core.Http;
 
 namespace Sisk.Cadente;
 
@@ -125,6 +126,7 @@ internal HttpRequest ( HttpRequestBase request, HttpRequestStream requestStream
     public sealed class HttpResponse {
         private Stream _baseOutputStream;
         private HttpHostContext _session;
+        private bool headersSent = false;
 
         /// <summary>
         /// Gets or sets the HTTP status code of the response.
@@ -141,17 +143,6 @@ public sealed class HttpResponse {
         /// </summary>
         public List<HttpHeader> Headers { get; set; }
 
-        /// <summary>
-        /// Gets or sets an boolean indicating if this <see cref="HttpResponse"/> should be send in chunks or not.
-        /// </summary>
-        public bool SendChunked { get; set; }
-
-        // MUST SPECIFY ResponseStream OR ResponseBytes, NOT BOTH
-        /// <summary>
-        /// Gets or sets the stream for the response content.
-        /// </summary>
-        public Stream? ResponseStream { get; set; }
-
         /// <summary>
         /// Asynchronously gets an event stream writer with UTF-8 encoding.
         /// </summary>
@@ -165,28 +156,45 @@ public sealed class HttpResponse {
         /// <returns>A task that represents the asynchronous operation, with a <see cref="HttpEventStreamWriter"/> as the result.</returns>
         /// <exception cref="InvalidOperationException">Thrown when unable to obtain an output stream for the response.</exception>
         public HttpEventStreamWriter GetEventStream ( Encoding encoding ) {
+            if (headersSent) {
+                throw new InvalidOperationException ( "Headers already sent for this HTTP response." );
+            }
+
             Headers.Set ( new HttpHeader ( "Content-Type", "text/event-stream" ) );
             Headers.Set ( new HttpHeader ( "Cache-Control", "no-cache" ) );
 
-            if (_session.WriteHttpResponseHeaders () == false) {
+            if (!_session.WriteHttpResponseHeaders ()) {
                 throw new InvalidOperationException ( "Unable to obtain the output stream for the response." );
             }
 
+            headersSent = true;
             return new HttpEventStreamWriter ( _baseOutputStream, encoding );
         }
 
         /// <summary>
-        /// Asynchronously gets the content stream for the response.
+        /// Gets the content stream for the response.
         /// </summary>
         /// <returns>A task that represents the asynchronous operation, with the response content stream as the result.</returns>
         /// <exception cref="InvalidOperationException">Thrown when unable to obtain an output stream for the response.</exception>
-        public Stream GetResponseStream () {
-            if (_session.WriteHttpResponseHeaders () == false) {
+        public Stream GetResponseStream ( bool chunked = false ) {
+            if (headersSent) {
+                throw new InvalidOperationException ( "Headers already sent for this HTTP response." );
+            }
+
+            if (chunked) {
+                Headers.Set ( new HttpHeader ( HttpHeaderName.TransferEncoding, "chunked" ) );
+                Headers.Remove ( HttpHeaderName.ContentLength );
+            }
+
+            if (!_session.WriteHttpResponseHeaders ()) {
                 throw new InvalidOperationException ( "Unable to obtain an output stream for the response." );
             }
 
-            ResponseStream = null;
-            return _baseOutputStream;
+            headersSent = true;
+            return chunked switch {
+                true => new HttpChunkedWriteStream ( _baseOutputStream ),
+                false => _baseOutputStream
+            };
         }
 
         internal HttpResponse ( HttpHostContext session, Stream httpSessionStream ) {

cadente/Sisk.Cadente/Streams/HttpChunkedStream.cs renamed to cadente/Sisk.Cadente/Streams/HttpChunkedReadStream.cs

@@ -7,20 +7,23 @@
 // File name:   HttpChunkedStream.cs
 // Repository:  https://github.com/sisk-http/core
 
+using System.Buffers;
 using System.Text;
 
 namespace Sisk.Cadente.Streams;
 
-internal class HttpChunkedStream : Stream {
+internal class HttpChunkedReadStream : Stream {
     private Stream _stream;
+    private byte [] _buffer;
     int written = 0;
     bool innerStreamReturnedZero = false;
 
-    const int CHUNKED_MAX_SIZE = 4096;
-    static readonly byte [] CrLf = [ 0x0D, 0x0A ];
+    private static readonly byte [] s_crlfBytes = "\r\n"u8.ToArray ();
+    private static readonly byte [] s_finalChunkBytes = "0\r\n\r\n"u8.ToArray ();
 
-    public HttpChunkedStream ( Stream stream ) {
+    public HttpChunkedReadStream ( Stream stream ) {
         _stream = stream;
+        _buffer = ArrayPool<byte>.Shared.Rent ( 128 * 1024 );
     }
 
     public override bool CanRead => _stream.CanRead;
@@ -42,34 +45,30 @@ public override int Read ( byte [] buffer, int offset, int count ) {
         if (innerStreamReturnedZero)
             return 0;
 
-        Span<byte> destination = buffer.AsSpan ( offset, count );
-        if (destination.Length < 2)
-            throw new ArgumentException ( "The provided buffer slice must be at least 2 bytes long.", nameof ( count ) );
-
-        Span<byte> readBuffer = stackalloc byte [ Math.Min ( destination.Length - 2, CHUNKED_MAX_SIZE ) ];
-        int read = _stream.Read ( readBuffer );
-        byte [] readBytesEncoded = Encoding.ASCII.GetBytes ( $"{read:x}\r\n" );
-
-        if (destination.Length < readBytesEncoded.Length + read + 2)
-            throw new ArgumentException ( "The provided buffer slice is not large enough to hold the chunked response.", nameof ( count ) );
-
-        if (read == 0) {
+        var position = 0;
+        var bytesRead = _stream.Read ( _buffer, 0, Math.Min ( _buffer.Length, count ) );
+        if (bytesRead == 0) {
             innerStreamReturnedZero = true;
+
+            // write last chunk
+            WriteToBuffer ( s_finalChunkBytes );
+            return s_finalChunkBytes.Length;
         }
 
-        ReadOnlySpan<byte> headerSpan = readBytesEncoded.AsSpan ();
-        headerSpan.CopyTo ( destination );
+        var bytesReadHex = Encoding.ASCII.GetBytes ( bytesRead.ToString ( "X" ) );
 
-        int copyStart = headerSpan.Length;
-        readBuffer [ 0..read ].CopyTo ( destination [ copyStart.. ] );
+        WriteToBuffer ( bytesReadHex );
+        WriteToBuffer ( s_crlfBytes );
+        WriteToBuffer ( _buffer.AsSpan ( 0, bytesRead ) );
+        WriteToBuffer ( s_crlfBytes );
 
-        int bufferEnd = headerSpan.Length + read;
-        destination [ bufferEnd + 0 ] = 0x0D;
-        destination [ bufferEnd + 1 ] = 0x0A;
+        return position;
 
-        written += read;
+        void WriteToBuffer ( Span<byte> data ) {
 
-        return bufferEnd + 2;
+            data.CopyTo ( buffer.AsSpan ( offset + position ) );
+            position += data.Length;
+        }
     }
 
     public override long Seek ( long offset, SeekOrigin origin ) {
@@ -83,4 +82,19 @@ public override void SetLength ( long value ) {
     public override void Write ( byte [] buffer, int offset, int count ) {
         throw new NotSupportedException ();
     }
+
+    protected override void Dispose ( bool disposing ) {
+
+        if (_stream == null)
+            return;
+
+        if (disposing) {
+
+            _stream.Dispose ();
+            ArrayPool<byte>.Shared.Return ( _buffer );
+
+            _stream = null!;
+            _buffer = null!;
+        }
+    }
 }