using CloudPrint.Client.Models; using CloudPrint.Client.Services.Abstractions; using System.IO; using System.Net.Http; using System.Net.WebSockets; using System.Text; namespace CloudPrint.Client.Services; /// /// WebSocket 备用通信通道。 /// 当前后端主链路仍使用 gRPC 订阅;当后端提供 /ws/print 时,本服务可保持备用长连接并记录推送消息。 /// public sealed class WebSocketService : IWebSocketService { private const int ReceiveBufferSize = 8 * 1024; private readonly SemaphoreSlim _connectionLock = new(1, 1); private ClientWebSocket? _socket; private CancellationTokenSource? _receiveCts; public event EventHandler? LogReceived; public bool IsConnected { get; private set; } public Uri? Endpoint { get; private set; } public async Task ConnectAsync(Uri endpoint, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(endpoint); if (endpoint.Scheme is not "ws" and not "wss") { throw new ArgumentException("WebSocket 地址必须以 ws:// 或 wss:// 开头", nameof(endpoint)); } await _connectionLock.WaitAsync(cancellationToken).ConfigureAwait(false); try { await DisconnectCoreAsync(CancellationToken.None).ConfigureAwait(false); Endpoint = endpoint; var socket = new ClientWebSocket(); _receiveCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); try { await socket.ConnectAsync(endpoint, cancellationToken).ConfigureAwait(false); } catch (Exception ex) when (ex is WebSocketException or HttpRequestException or IOException) { socket.Dispose(); _receiveCts.Dispose(); _receiveCts = null; IsConnected = false; Log(AppLogLevel.Warning, $"WebSocket 备用通道不可用:{ex.Message}"); return; } _socket = socket; IsConnected = true; Log(AppLogLevel.Success, $"WebSocket 备用通道已连接:{endpoint}"); _ = Task.Run(() => ReceiveLoopAsync(socket, _receiveCts.Token), CancellationToken.None); } finally { _connectionLock.Release(); } } public async Task DisconnectAsync(CancellationToken cancellationToken = default) { await _connectionLock.WaitAsync(cancellationToken).ConfigureAwait(false); try { await DisconnectCoreAsync(cancellationToken).ConfigureAwait(false); } finally { _connectionLock.Release(); } } private async Task ReceiveLoopAsync(ClientWebSocket socket, CancellationToken cancellationToken) { var buffer = new byte[ReceiveBufferSize]; try { while (!cancellationToken.IsCancellationRequested && socket.State == WebSocketState.Open) { using var message = new MemoryStream(); WebSocketReceiveResult result; do { result = await socket.ReceiveAsync(buffer, cancellationToken).ConfigureAwait(false); if (result.MessageType == WebSocketMessageType.Close) { await CloseSocketAsync(socket, cancellationToken).ConfigureAwait(false); MarkDisconnected(socket, "WebSocket 备用通道被服务端关闭"); return; } message.Write(buffer, 0, result.Count); } while (!result.EndOfMessage); if (result.MessageType == WebSocketMessageType.Text) { var payload = Encoding.UTF8.GetString(message.ToArray()); Log(AppLogLevel.Info, $"WebSocket 推送:{TrimPayload(payload)}"); } else if (result.MessageType == WebSocketMessageType.Binary) { Log(AppLogLevel.Info, $"WebSocket 收到二进制消息:{message.Length} 字节"); } } } catch (OperationCanceledException) { // Normal shutdown path. } catch (Exception ex) when (ex is WebSocketException or IOException) { MarkDisconnected(socket, $"WebSocket 备用通道中断:{ex.Message}"); } } private async Task DisconnectCoreAsync(CancellationToken cancellationToken) { _receiveCts?.Cancel(); var socket = _socket; _socket = null; IsConnected = false; if (socket is not null) { await CloseSocketAsync(socket, cancellationToken).ConfigureAwait(false); socket.Dispose(); Log(AppLogLevel.Warning, "WebSocket 备用通道已断开"); } _receiveCts?.Dispose(); _receiveCts = null; } private static async Task CloseSocketAsync(ClientWebSocket socket, CancellationToken cancellationToken) { if (socket.State is WebSocketState.Open or WebSocketState.CloseReceived) { await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "client closing", cancellationToken).ConfigureAwait(false); } } private void MarkDisconnected(ClientWebSocket socket, string message) { if (ReferenceEquals(_socket, socket)) { _socket = null; IsConnected = false; _receiveCts?.Dispose(); _receiveCts = null; } socket.Dispose(); Log(AppLogLevel.Warning, message); } private void Log(AppLogLevel level, string message) { LogReceived?.Invoke(this, new AppLogEntry { Level = level, Message = message }); } private static string TrimPayload(string payload) { const int maxLength = 500; return payload.Length <= maxLength ? payload : payload[..maxLength] + "..."; } }