YD_YiLu.Api/YD_YiLu/WebSockets/CameraDeviceWebSocketHandler.cs

using System.Net.WebSockets;
using System.Text;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json.Linq;

namespace YD_YiLu.Api.WebSockets
{
    internal sealed class BoundSerialHolder
    {
        public string? Value { get; set; }
    }

    public class CameraDeviceWebSocketHandler
    {
        private readonly ICameraDeviceSocketRegistry _registry;
        private readonly ILogger<CameraDeviceWebSocketHandler> _logger;

        public CameraDeviceWebSocketHandler(ICameraDeviceSocketRegistry registry,
            ILogger<CameraDeviceWebSocketHandler> logger)
        {
            _registry = registry;
            _logger = logger;
        }

        public async Task HandleAsync(WebSocket webSocket, CancellationToken ct)
        {
            var boundSerial = new BoundSerialHolder();
            var buffer = new byte[8192];

            _logger.LogInformation("🔌 WebSocket连接建立");

            try
            {
                while (webSocket.State == WebSocketState.Open && !ct.IsCancellationRequested)
                {
                    var (text, close) = await ReceiveAsync(webSocket, buffer, ct);
                    if (close)
                    {
                        if (boundSerial.Value != null)
                            _registry.Unregister(boundSerial.Value);

                        await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closed", ct);
                        return;
                    }

                    if (string.IsNullOrWhiteSpace(text)) continue;

                    _logger.LogInformation("📥 收到: {0}", text);

                    JObject root;
                    try { root = JObject.Parse(text); }
                    catch { continue; }

                    var request = root["request"]?.ToString();

                    switch (request)
                    {
                        case "station:live":
                            var serial = root["serialNumber"]?.ToString();
                            if (!string.IsNullOrEmpty(serial))
                            {
                                boundSerial.Value = serial;
                                _registry.Register(serial, webSocket);
                            }
                            break;

                        case "station:heartbeat":
                            break;

                        case "station:remind":
                            _logger.LogInformation("📢 提醒: {0}", root["data"]);
                            break;

                        case "station:user:push":
                            _logger.LogInformation("👥 用户同步: {0}", root["data"]);
                            break;
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "WebSocket异常");
            }
            finally
            {
                if (boundSerial.Value != null)
                    _registry.Unregister(boundSerial.Value);

                _logger.LogInformation("🔌 连接关闭");
            }
        }

        private static async Task<(string text, bool close)> ReceiveAsync(WebSocket ws, byte[] buffer, CancellationToken ct)
        {
            using var ms = new MemoryStream();
            WebSocketReceiveResult result;

            do
            {
                result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), ct);

                if (result.MessageType == WebSocketMessageType.Close)
                    return ("", true);

                ms.Write(buffer, 0, result.Count);

            } while (!result.EndOfMessage);

            return (Encoding.UTF8.GetString(ms.ToArray()), false);
        }

        public static Task SendAsync(WebSocket ws, string text, CancellationToken ct)
        {
            var bytes = Encoding.UTF8.GetBytes(text);
            return ws.SendAsync(bytes, WebSocketMessageType.Text, true, ct);
        }
    }
}