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] + "...";
}
}