more reliable websocket

Hecate2 · Feb 14, 2023 · e79c917 · Hecate2 · Feb 14, 2023 · e79c917
1 parent b8a46fb
commit e79c917
Showing 1 changed file with 34 additions and 9 deletions.
diff --git a/Fairy.WebSocket.Subscribe.cs b/Fairy.WebSocket.Subscribe.cs
@@ -1,17 +1,30 @@
 using Neo.Json;
 using Neo.Ledger;
 using Neo.Network.P2P.Payloads;
+using System.Collections.Concurrent;
 using System.Net.WebSockets;
 
 namespace Neo.Plugins
 {
     public partial class Fairy : RpcServer
     {
-        protected TaskCompletionSource<Block> committedBlock = new();
+        protected Block committedBlock;
+        protected ConcurrentDictionary<SemaphoreSlim, WebSocket> committedBlockSemaphores = new();
 
         protected void RegisterBlockchainEvents()
         {
-            Blockchain.Committed += delegate (NeoSystem @system, Block @block) { committedBlock.SetResult(block); };
+            Blockchain.Committed += delegate (NeoSystem @system, Block @block)
+            {
+                committedBlock = block;
+                List<SemaphoreSlim> keysToRemove = new();
+                foreach (var item in committedBlockSemaphores)
+                    if (item.Value.State == WebSocketState.Open)
+                        item.Key.Release();
+                    else
+                        keysToRemove.Add(item.Key);
+                foreach (SemaphoreSlim key in keysToRemove)
+                    committedBlockSemaphores.TryRemove(key, out _);
+            };
         }
 
         [WebsocketControlMethod]
@@ -47,18 +60,30 @@ protected virtual Action SubscribeCommittedBlock(WebSocket webSocket, JArray _pa
         {
             return async () =>
             {
+                SemaphoreSlim semaphore = new(1);
+                committedBlockSemaphores[semaphore] = webSocket;
                 while (true)
                 {
-                    Block block = await committedBlock.Task;
-                    if (webSocket.State == WebSocketState.Open)
+                    try
                     {
-                        await webSocket.SendAsync(block.ToJson(system.Settings).ToByteArray(false), WebSocketMessageType.Text, true, CancellationToken.None);
-                        if (cancellationToken.IsCancellationRequested)
-                            return;
+                        await semaphore.WaitAsync();
+                        switch (webSocket.State)
+                        {
+                            case WebSocketState.Open:
+                                await webSocket.SendAsync(committedBlock.ToJson(system.Settings).ToByteArray(false), WebSocketMessageType.Text, true, CancellationToken.None);
+                                if (cancellationToken.IsCancellationRequested)
+                                    return;
+                                break;
+                            case WebSocketState.Closed:
+                                await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, webSocket.State.ToString(), CancellationToken.None);
+                                webSocket.Dispose();
+                                return;
+                            default:
+                                break;
+                        }
                     }
-                    else
+                    catch
                     {
-                        await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, webSocket.State.ToString(), CancellationToken.None);
                         webSocket.Dispose();
                         return;
                     }