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 _logger; public CameraDeviceWebSocketHandler(ICameraDeviceSocketRegistry registry, ILogger 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(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); } } }