using System.Buffers; using System.Net.WebSockets; using System.Text; using System.Text.Json; using Serilog; using YZWater.Core.Models; namespace YZWater.Core.Services; /// /// 中控客户端 - 连接到服务端,接收数据,发送命令,自动重连 /// public class HubClient : IDisposable { private ClientWebSocket? _ws; private CancellationTokenSource? _cts; private Task? _connectionTask; private bool _disposed; private volatile bool _isConnected; private volatile bool _shouldReconnect = true; /// /// 服务端地址 /// public string ServerUrl { get; set; } = "ws://localhost:8765"; /// /// 认证令牌 /// public string Token { get; set; } = string.Empty; /// /// 是否忽略证书验证(仅用于自签名证书的测试环境) /// public bool IgnoreCertificateErrors { get; set; } = false; /// /// 是否已连接 /// public bool IsConnected => _isConnected; /// /// 是否有操作锁 /// public bool HasLock { get; private set; } /// /// 当前操作员 /// public string? CurrentOperator { get; private set; } public event Action? DataReceived; public event Action? AlarmReceived; public event Action? LockStatusChanged; public event Action? ConnectionChanged; public event Action? ClientCountChanged; public event Action? CommandResult; /// /// 连接到服务端(带自动重连) /// public async Task ConnectAsync() { if (_isConnected) return; _cts = new CancellationTokenSource(); _shouldReconnect = true; await ConnectInternalAsync(); // 启动接收 + 重连循环 _connectionTask = Task.Run(() => ConnectionLoopAsync(_cts.Token)); } /// /// 断开连接 /// public async Task DisconnectAsync() { _shouldReconnect = false; _cts?.Cancel(); if (_ws?.State == WebSocketState.Open) { try { await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "客户端断开", CancellationToken.None); } catch { /* 断开清理中,忽略关闭异常 */ } } _isConnected = false; HasLock = false; CurrentOperator = null; ConnectionChanged?.Invoke(false); Log.Information("已断开中控服务端连接"); } /// /// 连接 + 接收 + 自动重连主循环 /// private async Task ConnectionLoopAsync(CancellationToken ct) { var reconnectDelay = 1000; const int maxDelay = 30000; while (!ct.IsCancellationRequested && _shouldReconnect) { try { if (!_isConnected) { await ConnectInternalAsync(); reconnectDelay = 1000; // 连接成功,重置延迟 } // 进入接收循环 await ReceiveLoopAsync(ct); } catch (OperationCanceledException) { break; } catch (Exception ex) { Log.Warning(ex, "连接异常"); } // 连接断开,自动重连 if (!ct.IsCancellationRequested && _shouldReconnect) { _isConnected = false; HasLock = false; ConnectionChanged?.Invoke(false); Log.Information("将在 {Delay}ms 后重连...", reconnectDelay); try { await Task.Delay(reconnectDelay, ct); } catch (OperationCanceledException) { break; } reconnectDelay = Math.Min(reconnectDelay * 2, maxDelay); } } } /// /// 内部连接方法 /// private async Task ConnectInternalAsync() { _ws?.Dispose(); _ws = new ClientWebSocket(); // 如果启用忽略证书验证(自签名证书场景) if (IgnoreCertificateErrors && ServerUrl.StartsWith("wss://")) { _ws.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) => true; Log.Warning("已忽略 TLS 证书验证(仅用于测试环境)"); } try { await _ws.ConnectAsync(new Uri(ServerUrl), _cts?.Token ?? CancellationToken.None); // 发送认证令牌 if (!string.IsNullOrEmpty(Token)) { await SendMessageAsync(HubMessage.Auth(Token)); // 等待认证结果 var authResult = await ReceiveOneMessageAsync(_cts?.Token ?? CancellationToken.None); if (authResult?.Type != "auth_result" || authResult.GetPayload() != true) { Log.Warning("中控服务端认证失败"); _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "认证失败", CancellationToken.None).Wait(1000); throw new Exception("认证失败"); } Log.Information("中控服务端认证成功"); } _isConnected = true; ConnectionChanged?.Invoke(true); Log.Information("已连接到中控服务端: {Url}", ServerUrl); } catch (Exception ex) { Log.Warning(ex, "连接中控服务端失败: {Url}", ServerUrl); _isConnected = false; throw; } } /// /// 请求操作锁 /// public async Task RequestLockAsync(string operatorName) { await SendMessageAsync(new HubMessage { Type = "command", Payload = JsonSerializer.SerializeToElement(new CommandPayload { Action = "request_lock", Operator = operatorName }) }); } /// /// 释放操作锁 /// public async Task ReleaseLockAsync() { await SendMessageAsync(new HubMessage { Type = "command", Payload = JsonSerializer.SerializeToElement(new CommandPayload { Action = "release_lock", Operator = CurrentOperator ?? "" }) }); } /// /// 发送控制命令 /// public async Task SendCommandAsync(string action, string? target = null, object? value = null, string? operatorName = null) { await SendMessageAsync(new HubMessage { Type = "command", Payload = JsonSerializer.SerializeToElement(new CommandPayload { Action = action, Target = target, Value = value, Operator = operatorName ?? CurrentOperator ?? "unknown" }) }); } /// /// 发送消息 /// private async Task SendMessageAsync(HubMessage message) { if (_ws?.State != WebSocketState.Open) { Log.Warning("未连接到服务端,无法发送消息"); return; } var bytes = Encoding.UTF8.GetBytes(message.ToJson()); await _ws.SendAsync(new ArraySegment(bytes), WebSocketMessageType.Text, true, CancellationToken.None); } /// /// 接收单条消息(用于认证等握手场景) /// private async Task ReceiveOneMessageAsync(CancellationToken ct) { if (_ws?.State != WebSocketState.Open) return null; var buffer = new byte[65536]; var messageBuffer = new MemoryStream(); WebSocketReceiveResult result; do { result = await _ws.ReceiveAsync(new ArraySegment(buffer), ct); if (result.MessageType == WebSocketMessageType.Close) return null; messageBuffer.Write(buffer, 0, result.Count); } while (!result.EndOfMessage); if (result.MessageType != WebSocketMessageType.Text) return null; var json = Encoding.UTF8.GetString(messageBuffer.ToArray()); return HubMessage.FromJson(json); } /// /// 接收消息循环(带分片处理,使用 ArrayPool 复用缓冲区) /// private async Task ReceiveLoopAsync(CancellationToken ct) { var buffer = ArrayPool.Shared.Rent(65536); var messageBuffer = new MemoryStream(); try { while (_ws?.State == WebSocketState.Open && !ct.IsCancellationRequested) { messageBuffer.SetLength(0); // 复用 MemoryStream WebSocketReceiveResult result; do { result = await _ws.ReceiveAsync(new ArraySegment(buffer), ct); if (result.MessageType == WebSocketMessageType.Close) { return; // 触发重连 } messageBuffer.Write(buffer, 0, result.Count); } while (!result.EndOfMessage); if (result.MessageType == WebSocketMessageType.Text) { var json = Encoding.UTF8.GetString(messageBuffer.ToArray()); var message = HubMessage.FromJson(json); if (message != null) { HandleMessage(message); } } } } finally { ArrayPool.Shared.Return(buffer); messageBuffer.Dispose(); } } /// /// 处理接收到的消息 /// private void HandleMessage(HubMessage message) { switch (message.Type) { case "data": var data = message.GetPayload(); if (data != null) DataReceived?.Invoke(data); break; case "alarm": var alarm = message.GetPayload(); if (alarm != null) AlarmReceived?.Invoke(alarm); break; case "lock_result": var lockResult = message.GetPayload(); if (lockResult != null) { HasLock = lockResult.Success; CurrentOperator = lockResult.CurrentOperator; LockStatusChanged?.Invoke(lockResult.Success, lockResult.CurrentOperator); } break; case "lock_status": var lockStatus = message.GetPayload(); if (lockStatus != null) { LockStatusChanged?.Invoke(lockStatus.IsLocked, lockStatus.CurrentOperator); } break; case "command_result": var cmdResult = message.GetPayload(); if (cmdResult != null) { CommandResult?.Invoke(cmdResult.Success, cmdResult.Message); } break; case "client_count": var count = message.Payload?.GetInt32(); if (count.HasValue) ClientCountChanged?.Invoke(count.Value); break; } } public void Dispose() { if (_disposed) return; _disposed = true; Log.Information("HubClient 正在释放... _shouldReconnect={Reconnect}, _connectionTask={HasTask}", _shouldReconnect, _connectionTask != null); _shouldReconnect = false; _cts?.Cancel(); if (_ws?.State == WebSocketState.Open) { try { _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "关闭", CancellationToken.None).Wait(1000); } catch { /* 断开清理中,忽略关闭异常 */ } } _ws?.Dispose(); try { var completed = _connectionTask?.Wait(TimeSpan.FromSeconds(2)); if (completed == true) Log.Information("HubClient 连接任务已完成"); else Log.Warning("HubClient 连接任务在2秒内未完成"); } catch (AggregateException ex) when (ex.InnerExceptions.All(e => e is OperationCanceledException)) { // 取消是正常退出路径。 Log.Information("HubClient 连接任务已取消(正常)"); } _connectionTask = null; _cts?.Dispose(); Log.Information("HubClient 已释放"); } }