fixed: websocket server wrong execution thread

2022-12-25 03:13:01 -08:00
parent f34b05e6ff
commit e9418f4b22
10 changed files with 110 additions and 109 deletions
@@ -13,109 +13,112 @@ namespace Ragon.Core;

 public class NativeWebSocketServer : INetworkServer
 {
-    private ILogger _logger = LogManager.GetCurrentClassLogger();
-    private INetworkListener _networkListener;
-    private Stack<ushort> _sequencer;
-    private Executor _executor;
-    private HttpListener _httpListener;
-    private WebSocketConnection[] _connections;
-    private ushort _lastPeerId;
-    private CancellationTokenSource _cancellationTokenSource;
+  private ILogger _logger = LogManager.GetCurrentClassLogger();
+  private INetworkListener _networkListener;
+  private Stack<ushort> _sequencer;
+  private Executor _executor;
+  private HttpListener _httpListener;
+  private WebSocketConnection[] _connections;
+  private List<WebSocketConnection> _activeConnections;
+  private CancellationTokenSource _cancellationTokenSource;

-    public NativeWebSocketServer(Executor executor)
+  public NativeWebSocketServer(Executor executor)
+  {
+    _sequencer = new Stack<ushort>();
+    _connections = Array.Empty<WebSocketConnection>();
+    _activeConnections = new List<WebSocketConnection>();
+    _executor = executor;
+  }
+
+  public async void StartAccept(CancellationToken cancellationToken)
+  {
+    while (!cancellationToken.IsCancellationRequested)
    {
-        _sequencer = new Stack<ushort>();
-        _connections = Array.Empty<WebSocketConnection>();
-        _executor = executor;
+      var context = await _httpListener.GetContextAsync();
+      if (!context.Request.IsWebSocketRequest) continue;
+
+      var webSocketContext = await context.AcceptWebSocketAsync(null);
+      var webSocket = webSocketContext.WebSocket;
+
+      var peerId = _sequencer.Pop();
+      var connection = new WebSocketConnection(webSocket, peerId);
+      
+      _connections[peerId] = connection;
+      StartListen(connection, cancellationToken);
    }
+  }

-    public async Task StartAccept(CancellationToken cancellationToken)
+  async void StartListen(WebSocketConnection connection, CancellationToken cancellationToken)
+  {
+    _activeConnections.Add(connection);
+    _networkListener.OnConnected(connection);
+
+    var webSocket = connection.Socket;
+    var bytes = new byte[2048];
+    var buffer = new Memory<byte>(bytes);
+    while (
+      webSocket.State == WebSocketState.Open ||
+      !cancellationToken.IsCancellationRequested)
    {
-        while (!cancellationToken.IsCancellationRequested)
-        {
-            var context = await _httpListener.GetContextAsync();
-            if (!context.Request.IsWebSocketRequest) continue;
-
-            var webSocketContext = await context.AcceptWebSocketAsync(null);
-            var webSocket = webSocketContext.WebSocket;
-
-            var peerId = _sequencer.Pop();
-            var connection = new WebSocketConnection(webSocket, peerId);
-
-            _lastPeerId = peerId;
-            _connections[peerId] = connection;
-            _networkListener.OnConnected(connection);
-            _executor.Run(StartListen(connection, cancellationToken));
-        }
-    }
-
-    async Task StartListen(WebSocketConnection connection, CancellationToken cancellationToken)
-    {
-        var webSocket = connection.Socket;
-        var bytes = new byte[2048];
-        var buffer = new Memory<byte>(bytes);
-        while (
-            webSocket.State == WebSocketState.Open ||
-            !cancellationToken.IsCancellationRequested)
-        {
-            try
-            {
-                var result = await webSocket.ReceiveAsync(buffer, cancellationToken);
-                var dataRaw = buffer.Slice(0, result.Count);
-
-                _networkListener.OnData(connection, dataRaw.ToArray());
-            }
-            catch (Exception ex)
-            {
-                break;
-            }
-        }
+      try
+      {
+        var result = await webSocket.ReceiveAsync(buffer, cancellationToken);
+        var dataRaw = buffer.Slice(0, result.Count);
        
-        _sequencer.Push(connection.Id);
-        _networkListener.OnDisconnected(connection);
+        _networkListener.OnData(connection, dataRaw.ToArray());
+      }
+      catch (Exception ex)
+      {
+        break;
+      }
    }

-    public async void Poll()
-    {
-        foreach (var conn in _connections)
-        {
-            if (conn != null)
-            {
-                await conn.Flush();
-            }
-        }
-    }
+    _sequencer.Push(connection.Id);
+    _activeConnections.Remove(connection);
+    _networkListener.OnDisconnected(connection);
+  }

-    public void Start(
-        INetworkListener listener,
-        NetworkConfiguration configuration
-    )
-    {
-        _networkListener = listener;
-        _cancellationTokenSource = new CancellationTokenSource();
+  public void Poll()
+  {
+    Flush();
+  }

-        var limit = (ushort) configuration.LimitConnections;
-        for (ushort i = limit; i != 0; i--)
-            _sequencer.Push(i);
-        
-        _sequencer.Push(0);
+  public async void Flush()
+  {
+    foreach (var conn in _activeConnections)
+      await conn.Flush();
+  }

-        _connections = new WebSocketConnection[configuration.LimitConnections];
-        
-        _httpListener = new HttpListener();
-        _httpListener.Prefixes.Add($"http://127.0.0.1:{configuration.Port}/");
-        _httpListener.Start();
+  public void Start(
+    INetworkListener listener,
+    NetworkConfiguration configuration
+  )
+  {
+    _networkListener = listener;
+    _cancellationTokenSource = new CancellationTokenSource();

-        _executor.Run(StartAccept(_cancellationTokenSource.Token));
+    var limit = (ushort)configuration.LimitConnections;
+    for (ushort i = limit; i != 0; i--)
+      _sequencer.Push(i);

-        var protocolDecoded = RagonVersion.Parse(configuration.Protocol);
-        _logger.Info($"Network listening on http://*:{configuration.Port}/");
-        _logger.Info($"Protocol: {protocolDecoded}");
-    }
+    _sequencer.Push(0);

-    public void Stop()
-    {
-        _cancellationTokenSource.Cancel();
-        _httpListener.Stop();
-    }
+    _connections = new WebSocketConnection[configuration.LimitConnections];
+
+    _httpListener = new HttpListener();
+    _httpListener.Prefixes.Add($"http://127.0.0.1:{configuration.Port}/");
+    _httpListener.Start();
+
+    _executor.Run(() => StartAccept(_cancellationTokenSource.Token));
+
+    var protocolDecoded = RagonVersion.Parse(configuration.Protocol);
+    _logger.Info($"Listen at http://*:{configuration.Port}/");
+    _logger.Info($"Protocol: {protocolDecoded}");
+  }
+
+  public void Stop()
+  {
+    _cancellationTokenSource.Cancel();
+    _httpListener.Stop();
+  }
 }