Glue asynchronous handlers with synchronous callbacks provided by Fleck

Author: Kaliumhexacyanoferrat

Modules/Websockets/Handler/WebsocketConnection.cs

@@ -47,26 +47,26 @@ public sealed class WebsocketConnection : IWebSocketConnection, IWebsocketConnec
     #region Initialization
 
     public WebsocketConnection(ISocket socket, IRequest request, List<string> supportedProtocols,
-        Action<IWebsocketConnection>? onOpen,
-        Action<IWebsocketConnection>? onClose,
-        Action<IWebsocketConnection, string>? onMessage,
-        Action<IWebsocketConnection, byte[]>? onBinary,
-        Action<IWebsocketConnection, byte[]>? onPing,
-        Action<IWebsocketConnection, byte[]>? onPong,
-        Action<IWebsocketConnection, Exception>? onError)
+        Func<IWebsocketConnection, Task>? onOpen,
+        Func<IWebsocketConnection, Task>? onClose,
+        Func<IWebsocketConnection, string, Task>? onMessage,
+        Func<IWebsocketConnection, byte[], Task>? onBinary,
+        Func<IWebsocketConnection, byte[], Task>? onPing,
+        Func<IWebsocketConnection, byte[], Task>? onPong,
+        Func<IWebsocketConnection, Exception, Task>? onError)
     {
         Socket = socket;
         Request = request;
 
         SupportedProtocols = supportedProtocols;
 
-        OnOpen = (onOpen != null) ? () => onOpen(this) : () => { };
-        OnClose = (onClose != null) ? () => onClose(this) : () => { };
-        OnMessage = (onMessage != null) ? x => onMessage(this, x) : x => { };
-        OnBinary = (onBinary != null) ? x => onBinary(this, x) : x => { };
-        OnPing = (onPing != null) ? x => onPing(this, x) : x => SendPongAsync(x);
-        OnPong = (onPong != null) ? x => onPong(this, x) : x => { };
-        OnError = (onError != null) ? x => onError(this, x) : x => { };
+        OnOpen = (onOpen != null) ? () => WebsocketDispatcher.Schedule(() => onOpen(this)) : () => { };
+        OnClose = (onClose != null) ? () => WebsocketDispatcher.Schedule(() => onClose(this)) : () => { };
+        OnMessage = (onMessage != null) ? x => WebsocketDispatcher.Schedule(() => onMessage(this, x)) : x => { };
+        OnBinary = (onBinary != null) ? x => WebsocketDispatcher.Schedule(() => onBinary(this, x)) : x => { };
+        OnPing = (onPing != null) ? x => WebsocketDispatcher.Schedule(() => onPing(this, x)) : x => WebsocketDispatcher.Schedule(() => SendPongAsync(x));
+        OnPong = (onPong != null) ? x => WebsocketDispatcher.Schedule(() => onPong(this, x)) : x => { };
+        OnError = (onError != null) ? x => WebsocketDispatcher.Schedule(() => onError(this, x)) : x => { };
     }
 
     #endregion

Modules/Websockets/Handler/WebsocketDispatcher.cs

@@ -0,0 +1,37 @@
+using Fleck;
+
+namespace GenHTTP.Modules.Websockets.Handler;
+
+/// <summary>
+/// Flecks uses synchronous callbacks for integration of user
+/// logic, but provides asynchronous methods for interacting
+/// with the websocket connection, which is counter-intuitive
+/// and leads to errors. GenHTTP´s integration forces asynchronous
+/// callbacks to be supplied by the user which requires us to dispatch
+/// them on the synchronous callbacks invoked by Fleck.
+/// </summary>
+public static class WebsocketDispatcher
+{
+    
+    /// <summary>
+    /// Schedules the given piece of work on a background
+    /// thread and logs any error using the regular Fleck
+    /// logging mechanism.
+    /// </summary>
+    /// <param name="work">The actual piece of work to be executed</param>
+    public static void Schedule(Func<Task> work)
+    {
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await work().ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                FleckLog.Error("Failed to run asynchronous event handler", e);
+            }
+        });
+    }
+    
+}

Testing/Acceptance/Modules/Websockets/InitializerTest.cs

@@ -1,4 +1,3 @@
-using Microsoft.VisualStudio.TestTools.UnitTesting;
 using WS = GenHTTP.Modules.Websockets;
 
 namespace GenHTTP.Testing.Acceptance.Modules.Websockets;
@@ -11,13 +10,13 @@ public class InitializerTest
     public void InitializeAll()
     {
         WS.Websocket.Create()
-          .OnOpen(s => { })
-          .OnClose(s => { })
-          .OnPing((s, b) => { s.SendPongAsync(b); })
-          .OnPong((s, b) => { })
-          .OnMessage((s, x) => { })
-          .OnBinary((s, x) => { })
-          .OnError((s, x) => { })
+          .OnOpen(s => Task.CompletedTask)
+          .OnClose(s => Task.CompletedTask)
+          .OnPing(async (s, b) => { await s.SendPongAsync(b); })
+          .OnPong((s, b) => Task.CompletedTask)
+          .OnMessage((s, x) => Task.CompletedTask)
+          .OnBinary((s, x) => Task.CompletedTask)
+          .OnError((s, x) => Task.CompletedTask)
           .Protocol("chat");
     }
 

Testing/Acceptance/Modules/Websockets/IntegrationTest.cs

@@ -1,9 +1,6 @@
 using GenHTTP.Testing.Acceptance.Utilities;
-
 using WS = GenHTTP.Modules.Websockets;
-
 using Microsoft.VisualStudio.TestTools.UnitTesting;
-
 using Websocket.Client;
 
 namespace GenHTTP.Testing.Acceptance.Modules.Websockets;
@@ -20,10 +17,12 @@ public async Task TestServer()
         var length = 0;
 
         var server = WS.Websocket.Create()
-                       .OnMessage((socket, msg) =>
+                       .OnMessage(async (socket, msg) =>
                        {
                            length += msg.Length;
-                           socket.SendAsync(msg);
+
+                           await socket.SendAsync(msg);
+
                            socket.Close();
                        });
 
@@ -54,23 +53,20 @@ public async Task TestDataTypes()
         var waitEvent = new ManualResetEvent(false);
 
         var server = WS.Websocket.Create()
-                       .OnOpen((socket) =>
+                       .OnOpen(async (socket) =>
                        {
-                           Task.Run(async () =>
-                           {
-                               await socket.SendPingAsync([42]);
-                               await socket.SendPongAsync([42]);
+                           await socket.SendPingAsync([42]);
+                           await socket.SendPongAsync([42]);
 
-                               await socket.SendAsync([42]);
+                           await socket.SendAsync([42]);
 
-                               Assert.IsTrue(socket.IsAvailable);
+                           Assert.IsTrue(socket.IsAvailable);
 
-                               Assert.IsTrue(socket.Request.Headers.Count > 0);
+                           Assert.IsTrue(socket.Request.Headers.Count > 0);
 
-                               socket.Close(42);
+                           socket.Close(42);
 
-                               waitEvent.Set();
-                           });
+                           waitEvent.Set();
                        });
 
         await using var host = await TestHost.RunAsync(server);