✨ added transfer ownership, limit buffered events

Ragon.Server.WebSocketServer/Sources/WebSocketConnection.cs

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2023 Eduard Kargin <kargin.eduard@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using NLog;
+using System.Net.WebSockets;
+using Ragon.Server.IO;
+
+namespace Ragon.Server.WebSocketServer;
+
+public sealed class WebSocketConnection : INetworkConnection
+{
+    private Logger _logger = LogManager.GetCurrentClassLogger();
+    public ushort Id { get; }
+    public INetworkChannel Reliable { get; private set; }
+    public INetworkChannel Unreliable { get; private set; }
+
+    public WebSocket Socket { get; private set; }
+    private WebSocketReliableChannel[] _channels;
+
+    public WebSocketConnection(WebSocket webSocket, ushort peerId)
+    {
+        Id = peerId;
+        Socket = webSocket;
+
+        var reliableChannel = new WebSocketReliableChannel(webSocket);
+        var unreliableChannel = new WebSocketReliableChannel(webSocket);
+
+        _channels = new[] { reliableChannel, unreliableChannel };
+
+        Reliable = reliableChannel;
+        Unreliable = unreliableChannel;
+    }
+
+    public void Close()
+    {
+        Socket.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
+    }
+    
+    public async Task Flush()
+    {
+        foreach (var channel in _channels)
+        {
+            try
+            {
+                await channel.Flush();
+            }
+            catch (Exception ex)
+            {
+                _logger.Error(ex);
+            }
+        }
+    }
+}

Ragon.Server.WebSocketServer/Sources/WebSocketReliableChannel.cs

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2023 Eduard Kargin <kargin.eduard@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System.Net.WebSockets;
+using Ragon.Server.IO;
+
+namespace Ragon.Server.WebSocketServer;
+
+public class WebSocketReliableChannel : INetworkChannel
+{
+    private Queue<byte[]> _queue;
+    private WebSocket _socket;
+
+    public WebSocketReliableChannel(WebSocket webSocket)
+    {
+        _socket = webSocket;
+        _queue = new Queue<byte[]>(512);
+    }
+
+    public void Send(byte[] data)
+    {
+        _queue.Enqueue(data);
+    }
+
+    public async Task Flush()
+    {
+        while (_queue.TryDequeue(out var sendData) && _socket.State == WebSocketState.Open)
+            await _socket.SendAsync(sendData, WebSocketMessageType.Binary, WebSocketMessageFlags.EndOfMessage, CancellationToken.None);
+    }
+}

Ragon.Server.WebSocketServer/Sources/WebSocketServer.cs

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2023 Eduard Kargin <kargin.eduard@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System.Net;
+using System.Net.WebSockets;
+using NLog;
+using Ragon.Protocol;
+using Ragon.Server.IO;
+
+namespace Ragon.Server.WebSocketServer;
+
+public class WebSocketServer : INetworkServer
+{
+  public Executor Executor => _executor;
+  
+  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 WebSocketServer()
+  {
+    _sequencer = new Stack<ushort>();
+    _connections = Array.Empty<WebSocketConnection>();
+    _activeConnections = new List<WebSocketConnection>();
+    _executor = new Executor();
+  }
+  
+  public async void StartAccept(CancellationToken cancellationToken)
+  {
+    while (!cancellationToken.IsCancellationRequested)
+    {
+      var context = await _httpListener.GetContextAsync();
+      if (!context.Request.IsWebSocketRequest)
+      {
+        context.Response.StatusCode = 200;
+        context.Response.ContentLength64 = 0;
+        context.Response.Close();
+        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);
+    }
+  }
+
+  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)
+    {
+      try
+      {
+        var result = await webSocket.ReceiveAsync(buffer, cancellationToken);
+        var dataRaw = buffer.Slice(0, result.Count);
+        if (dataRaw.Length > 0)
+          _networkListener.OnData(connection, dataRaw.ToArray());
+      }
+      catch (Exception ex)
+      {
+        break;
+      }
+    }
+
+    _sequencer.Push(connection.Id);
+    _activeConnections.Remove(connection);
+    _networkListener.OnDisconnected(connection);
+  }
+
+  public void Update()
+  {
+    Flush();
+  }
+
+  public async void Flush()
+  {
+    foreach (var conn in _activeConnections)
+      await conn.Flush();
+  }
+
+  public void Start(
+    INetworkListener listener,
+    NetworkConfiguration configuration
+  )
+  {
+    _networkListener = listener;
+    _cancellationTokenSource = new CancellationTokenSource();
+
+    var limit = (ushort)configuration.LimitConnections;
+    for (ushort i = limit; i != 0; i--)
+      _sequencer.Push(i);
+
+    _sequencer.Push(0);
+
+    _connections = new WebSocketConnection[configuration.LimitConnections];
+
+    _httpListener = new HttpListener();
+    _httpListener.Prefixes.Add($"http://+:{configuration.Port}/");
+    _httpListener.Start();
+
+    _executor.Run(() => StartAccept(_cancellationTokenSource.Token));
+
+    var protocolDecoded = RagonVersion.Parse(configuration.Protocol);
+    _logger.Info($"Listen at http://0.0.0.0:{configuration.Port}/");
+    _logger.Info($"Protocol: {protocolDecoded}");
+  }
+
+  public void Stop()
+  {
+    _cancellationTokenSource.Cancel();
+    _httpListener.Stop();
+  }
+}