client/Services/WebSocketService.cs

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;

/// <summary>
/// WebSocket 备用通信通道。
/// 当前后端主链路仍使用 gRPC 订阅；当后端提供 /ws/print 时，本服务可保持备用长连接并记录推送消息。
/// </summary>
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<AppLogEntry>? 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] + "...";
    }
}