wip: ws async improvements

CypherPotato · CypherPotato · commit d8e1f7cd5460 · 2025-03-20T00:34:40.000-03:00
diff --git a/extensions/Sisk.SslProxy/Sisk.SslProxy.csproj b/extensions/Sisk.SslProxy/Sisk.SslProxy.csproj
@@ -22,7 +22,7 @@
 		<PackageTags>http-server,http,web framework</PackageTags>
 		<RepositoryType>git</RepositoryType>
 
-		<Version>1.4.1-alpha9</Version>
+		<Version>1.4.1.1-alpha9</Version>
 		<AssemblyVersion>1.4.1</AssemblyVersion>
 		<FileVersion>1.4.1</FileVersion>
 
diff --git a/src/GlobalSuppressions.cs b/src/GlobalSuppressions.cs
@@ -10,5 +10,4 @@
 using System.Diagnostics.CodeAnalysis;
 
 [assembly: SuppressMessage ( "Usage", "CA2225:Operator overloads have named alternates", Scope = "member", Target = "~M:Sisk.Core.Http.HttpStatusInformation.op_Implicit(System.Net.HttpStatusCode)~Sisk.Core.Http.HttpStatusInformation" )]
-[assembly: SuppressMessage ( "Usage", "CA2225:Operator overloads have named alternates", Scope = "member", Target = "~M:Sisk.Core.Http.HttpStatusInformation.op_Implicit(System.Int32)~Sisk.Core.Http.HttpStatusInformation" )]
-[assembly: SuppressMessage ( "Json", "SYSLIB0020:JsonSerializerOptions.IgnoreNullValues is obsolete" )]
+[assembly: SuppressMessage ( "Usage", "CA2225:Operator overloads have named alternates", Scope = "member", Target = "~M:Sisk.Core.Http.HttpStatusInformation.op_Implicit(System.Int32)~Sisk.Core.Http.HttpStatusInformation" )]
diff --git a/src/Http/Streams/HttpRequestEventSource.cs b/src/Http/Streams/HttpRequestEventSource.cs
@@ -233,10 +233,18 @@ internal async ValueTask FlushAsync () {
         public void Dispose () {
             if (isDisposed)
                 return;
+
+            GC.SuppressFinalize ( this );
+
             Close ();
             sendQueue.Clear ();
             terminatingMutex.Dispose ();
             isDisposed = true;
         }
+
+        /// <exclude/>
+        ~HttpRequestEventSource () {
+            Dispose ();
+        }
     }
 }
diff --git a/src/Http/Streams/HttpWebSocket.cs b/src/Http/Streams/HttpWebSocket.cs
@@ -8,7 +8,8 @@
 // Repository:  https://github.com/sisk-http/core
 
 using System.Net.WebSockets;
-using System.Runtime.CompilerServices;
+using System.Text;
+using Sisk.Core.Internal;
 
 namespace Sisk.Core.Http.Streams {
 
@@ -17,8 +18,10 @@ namespace Sisk.Core.Http.Streams {
     /// </summary>
     public sealed class HttpWebSocket : IDisposable {
         bool isListening = true;
+        bool isDisposed;
         readonly HttpStreamPingPolicy pingPolicy;
 
+        internal SemaphoreSlim sendSemaphore = new SemaphoreSlim ( 1 );
         internal WebSocketMessage? lastMessage;
         internal CancellationTokenSource asyncListenerToken = null!;
         internal ManualResetEvent closeEvent = new ManualResetEvent ( false );
@@ -178,40 +181,22 @@ public HttpWebSocket WithPing ( string probeMessage, TimeSpan interval ) {
             return this;
         }
 
-        /// <summary>
-        /// Asynchronously sends an message to the remote point.
-        /// </summary>
-        /// <param name="message">The target message which will be as an encoded UTF-8 string.</param>
-        public Task<bool> SendAsync ( object message ) {
-            return Task.FromResult ( Send ( message ) );
-        }
-
         /// <summary>
         /// Asynchronously sends an text message to the remote point.
         /// </summary>
         /// <param name="message">The target message which will be as an encoded UTF-8 string.</param>
-        public Task<bool> SendAsync ( string message ) {
-            return Task.FromResult ( Send ( message ) );
+        public ValueTask<bool> SendAsync ( string message ) {
+            ArgumentNullException.ThrowIfNull ( message );
+
+            return SendInternalAsync ( Encoding.UTF8.GetBytes ( message ), WebSocketMessageType.Text );
         }
 
         /// <summary>
         /// Asynchronously sends an binary message to the remote point.
         /// </summary>
         /// <param name="buffer">The target message which will be as an encoded UTF-8 string.</param>
-        public Task<bool> SendAsync ( byte [] buffer ) {
-            return Task.FromResult ( Send ( buffer ) );
-        }
-
-        /// <summary>
-        /// Sends an text message to the remote point.
-        /// </summary>
-        /// <param name="message">The target message which will be as an encoded UTF-8 string.</param>
-        public bool Send ( object message ) {
-            string? t = message.ToString ();
-            if (t is null)
-                throw new ArgumentNullException ( nameof ( message ) );
-
-            return Send ( t );
+        public ValueTask<bool> SendAsync ( ReadOnlyMemory<byte> buffer ) {
+            return SendInternalAsync ( buffer, WebSocketMessageType.Binary );
         }
 
         /// <summary>
@@ -220,9 +205,7 @@ public bool Send ( object message ) {
         /// <param name="message">The target message which will be as an encoded using the request preferred encoding.</param>
         public bool Send ( string message ) {
             ArgumentNullException.ThrowIfNull ( message );
-
-            byte [] messageBytes = request.RequestEncoding.GetBytes ( message );
-            return SendInternal ( messageBytes, WebSocketMessageType.Text );
+            return SendAsync ( message ).GetSyncronizedResult ();
         }
 
         /// <summary>
@@ -239,15 +222,15 @@ public bool Send ( string message ) {
         /// <param name="length">The number of items in the memory.</param>
         public bool Send ( byte [] buffer, int start, int length ) {
             ReadOnlyMemory<byte> span = new ReadOnlyMemory<byte> ( buffer, start, length );
-            return SendInternal ( span, WebSocketMessageType.Binary );
+            return SendAsync ( span ).GetSyncronizedResult ();
         }
 
         /// <summary>
         /// Sends an binary message to the remote point.
         /// </summary>
         /// <param name="buffer">The target byte memory.</param>
         public bool Send ( ReadOnlyMemory<byte> buffer ) {
-            return SendInternal ( buffer, WebSocketMessageType.Binary );
+            return SendAsync ( buffer ).GetSyncronizedResult ();
         }
 
         /// <summary>
@@ -263,7 +246,7 @@ public HttpResponse Close () {
                     // the resources of this websocket
                     try {
                         ctx.WebSocket.CloseOutputAsync ( WebSocketCloseStatus.NormalClosure, null, CancellationToken.None )
-                            .Wait ();
+                            .GetAwaiter ().GetResult ();
                     }
                     catch (Exception) {
                         ;
@@ -282,41 +265,43 @@ public HttpResponse Close () {
             };
         }
 
-        [MethodImpl ( MethodImplOptions.Synchronized )]
-        private bool SendInternal ( ReadOnlyMemory<byte> buffer, WebSocketMessageType msgType ) {
-            if (_isClosed) { return false; }
+        private async ValueTask<bool> SendInternalAsync ( ReadOnlyMemory<byte> buffer, WebSocketMessageType msgType ) {
+            if (_isClosed)
+                return false;
 
             if (closeTimeout.TotalMilliseconds > 0)
                 asyncListenerToken?.CancelAfter ( closeTimeout );
 
+            await sendSemaphore.WaitAsync ( asyncListenerToken?.Token ?? default );
             try {
-                int totalLength = buffer.Length;
-                int chunks = (int) Math.Ceiling ( (double) totalLength / BUFFER_LENGTH );
+                try {
+                    int totalLength = buffer.Length;
+                    int chunks = (int) Math.Ceiling ( (double) totalLength / BUFFER_LENGTH );
 
-                for (int i = 0; i < chunks; i++) {
-                    int ca = i * BUFFER_LENGTH;
-                    int cb = Math.Min ( ca + BUFFER_LENGTH, buffer.Length );
+                    for (int i = 0; i < chunks; i++) {
+                        int ca = i * BUFFER_LENGTH;
+                        int cb = Math.Min ( ca + BUFFER_LENGTH, buffer.Length );
 
-                    ReadOnlyMemory<byte> chunk = buffer [ ca..cb ];
+                        ReadOnlyMemory<byte> chunk = buffer [ ca..cb ];
 
-                    var sendVt = ctx.WebSocket.SendAsync ( chunk, msgType, i + 1 == chunks, asyncListenerToken?.Token ?? default );
-                    if (!sendVt.IsCompleted) {
-                        sendVt.AsTask ().GetAwaiter ().GetResult ();
+                        await ctx.WebSocket.SendAsync ( chunk, msgType, i + 1 == chunks, asyncListenerToken?.Token ?? default );
+                        length += chunk.Length;
                     }
 
-                    length += chunk.Length;
+                    attempt = 0;
                 }
-
-                attempt = 0;
-            }
-            catch (Exception) {
-                attempt++;
-                if (MaxAttempts >= 0 && attempt >= MaxAttempts) {
-                    Close ();
-                    return false;
+                catch (Exception) {
+                    attempt++;
+                    if (MaxAttempts >= 0 && attempt >= MaxAttempts) {
+                        Close ();
+                        return false;
+                    }
                 }
+                return true;
+            }
+            finally {
+                sendSemaphore.Release ();
             }
-            return true;
         }
 
         /// <summary>
@@ -365,11 +350,22 @@ public void WaitForClose () {
 
         /// <inheritdoc/>
         public void Dispose () {
+            if (isDisposed)
+                return;
+
+            GC.SuppressFinalize ( this );
+
             Close ();
             pingPolicy.Dispose ();
             closeEvent.Dispose ();
             waitNextEvent.Dispose ();
             receiveThread.Join ();
+            isDisposed = true;
+        }
+
+        /// <exclude/>
+        ~HttpWebSocket () {
+            Dispose ();
         }
     }
 
diff --git a/src/Internal/AsyncHelpers.cs b/src/Internal/AsyncHelpers.cs
@@ -14,6 +14,6 @@ public static T GetSyncronizedResult<T> ( this ValueTask<T> task ) {
         if (task.IsCompleted) {
             return task.Result;
         }
-        return task.GetAwaiter ().GetResult ();
+        return task.AsTask ().GetAwaiter ().GetResult ();
     }
 }
diff --git a/tests/Server.cs b/tests/Server.cs
@@ -22,6 +22,13 @@ public static void AssemblyInit ( TestContext testContext ) {
                         Status = HttpStatusInformation.Ok
                     };
                 } );
+                router.MapGet ( "/tests/plaintext/chunked", delegate ( HttpRequest request ) {
+                    return new HttpResponse () {
+                        Content = new StringContent ( "Hello, world!", Encoding.UTF8, "text/plain" ),
+                        Status = HttpStatusInformation.Ok,
+                        SendChunked = true
+                    };
+                } );
             } )
             .Build ();
 
diff --git a/tests/Tests/HttpResponseTests.cs b/tests/Tests/HttpResponseTests.cs
@@ -17,4 +17,44 @@ public async Task OkPlainText () {
             Assert.AreEqual ( response.Content.Headers.ContentType?.CharSet, "utf-8" );
         }
     }
+
+    [TestMethod]
+    public async Task OkPlainTextChunked () {
+        using (var client = Server.GetHttpClient ()) {
+
+            var request = new HttpRequestMessage ( HttpMethod.Get, "tests/plaintext/chunked" );
+            var response = await client.SendAsync ( request );
+            var content = await response.Content.ReadAsStringAsync ();
+
+            Assert.IsTrue ( response.IsSuccessStatusCode );
+            Assert.IsTrue ( response.Headers.TransferEncodingChunked == true );
+            Assert.AreEqual ( content, "Hello, world!", ignoreCase: true );
+            Assert.AreEqual ( response.Content.Headers.ContentType?.MediaType, "text/plain" );
+            Assert.AreEqual ( response.Content.Headers.ContentType?.CharSet, "utf-8" );
+        }
+    }
+
+    [TestMethod]
+    public async Task NotFound () {
+        using (var client = Server.GetHttpClient ()) {
+
+            var request = new HttpRequestMessage ( HttpMethod.Get, "tests/not-found" );
+            var response = await client.SendAsync ( request );
+            var content = await response.Content.ReadAsStringAsync ();
+
+            Assert.IsTrue ( response.StatusCode == System.Net.HttpStatusCode.NotFound );
+        }
+    }
+
+    [TestMethod]
+    public async Task MethodNotAllowed () {
+        using (var client = Server.GetHttpClient ()) {
+
+            var request = new HttpRequestMessage ( HttpMethod.Post, "tests/plaintext" );
+            var response = await client.SendAsync ( request );
+            var content = await response.Content.ReadAsStringAsync ();
+
+            Assert.IsTrue ( response.StatusCode == System.Net.HttpStatusCode.MethodNotAllowed );
+        }
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -233,10 +233,18 @@ internal async ValueTask FlushAsync () {`
`233`	`233`	`public void Dispose () {`
`234`	`234`	`if (isDisposed)`
`235`	`235`	`return;`
	`236`	`+`
	`237`	`+ GC.SuppressFinalize ( this );`
	`238`	`+`
`236`	`239`	`Close ();`
`237`	`240`	`sendQueue.Clear ();`
`238`	`241`	`terminatingMutex.Dispose ();`
`239`	`242`	`isDisposed = true;`
`240`	`243`	`}`
	`244`	`+`
	`245`	`+ /// <exclude/>`
	`246`	`+ ~HttpRequestEventSource () {`
	`247`	`+ Dispose ();`
	`248`	`+ }`
`241`	`249`	`}`
`242`	`250`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,6 @@ public static T GetSyncronizedResult<T> ( this ValueTask<T> task ) {`
`14`	`14`	`if (task.IsCompleted) {`
`15`	`15`	`return task.Result;`
`16`	`16`	`}`
`17`		`- return task.GetAwaiter ().GetResult ();`
	`17`	`+ return task.AsTask ().GetAwaiter ().GetResult ();`
`18`	`18`	`}`
`19`	`19`	`}`