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();

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

                    switch (request)
                    {
                        case "station:live":
                            if (!string.IsNullOrEmpty(serial))
                            {
                                boundSerial.Value = serial;
                                _registry.Register(serial, webSocket);
                            }
                            break;

                        case "machine:heartbeat":
                        case "station:heartbeat":
                            {
                                var msgId = root["msgId"]?.ToString();

                                if (!string.IsNullOrEmpty(serial))
                                {
                                    // 注册 or 更新心跳
                                    if (string.IsNullOrEmpty(boundSerial.Value))
                                    {
                                        boundSerial.Value = serial;
                                        _registry.Register(serial, webSocket);
                                    }

                                    _registry.UpdateHeartbeat(serial);   // ⭐关键
                                }

                                var pong = new JObject
                                {
                                    ["serialNumber"] = serial,
                                    ["androidId"] = root["androidId"] ?? "",
                                    ["imei"] = root["imei"] ?? "",
                                    ["imsi"] = root["imsi"] ?? "",
                                    ["param"] = "pong",
                                    ["request"] = "station:heartbeat",
                                    ["msgId"] = msgId
                                }.ToString();

                                await SendAsync(webSocket, pong, ct);

                                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);
        }
    }
}