zcbot/windows-node/Zcbot.WindowsNode/Host/NodeSession.cs

using System.Net.WebSockets;
using System.Text.Json;

namespace Zcbot.WindowsNode;

internal sealed class NodeSession
{
    private static readonly TimeSpan[] Backoff =
    [
        TimeSpan.FromSeconds(1),
        TimeSpan.FromSeconds(2),
        TimeSpan.FromSeconds(5),
        TimeSpan.FromSeconds(10),
        TimeSpan.FromSeconds(30),
        TimeSpan.FromSeconds(60),
    ];

    private readonly NodeConfig config;
    private readonly Func<object> runtimePayload;
    private readonly Func<NodeProtocolClient, CancellationToken, Task> connected;
    private readonly Func<NodeProtocolClient, CancellationToken, Task> heartbeat;
    private readonly Func<NodeProtocolClient, NodeProtocolMessage, CancellationToken, Task> received;
    private readonly Action<NodeState, string> report;
    private readonly Func<NodeConfig, CancellationToken, Task<NodeProtocolClient>> connect;
    private readonly Func<TimeSpan, CancellationToken, Task> delay;

    internal NodeSession(
        NodeConfig config,
        Func<object> runtimePayload,
        Func<NodeProtocolClient, CancellationToken, Task> connected,
        Func<NodeProtocolClient, CancellationToken, Task> heartbeat,
        Func<NodeProtocolClient, NodeProtocolMessage, CancellationToken, Task> received,
        Action<NodeState, string> report,
        Func<NodeConfig, CancellationToken, Task<NodeProtocolClient>>? connect = null,
        Func<TimeSpan, CancellationToken, Task>? delay = null)
    {
        this.config = config;
        this.runtimePayload = runtimePayload;
        this.connected = connected;
        this.heartbeat = heartbeat;
        this.received = received;
        this.report = report;
        this.connect = connect ?? NodeProtocolClient.ConnectAsync;
        this.delay = delay ?? Task.Delay;
    }

    internal async Task RunAsync(CancellationToken cancellationToken)
    {
        var attempt = 0;
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                report(NodeState.Connecting, "正在连接 zcbot…");
                await ConnectOnceAsync(cancellationToken);
                attempt = 0;
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                throw;
            }
            catch (WebSocketException exception)
            {
                report(NodeState.Offline, "连接中断，等待重连");
                Console.Error.WriteLine($"[WARN] WebSocket disconnected: {exception.Message}");
            }
            catch (IOException exception)
            {
                report(NodeState.Offline, "网络不可用，等待重连");
                Console.Error.WriteLine($"[WARN] Connection I/O failed: {exception.Message}");
            }
            catch (JsonException exception)
            {
                report(NodeState.Offline, "服务端消息无效，等待重连");
                Console.Error.WriteLine($"[WARN] Invalid server message: {exception.Message}");
            }
            catch (NodeEndpointException exception)
            {
                report(NodeState.Offline, "WebSocket 握手被拒绝，请检查服务端或反向代理");
                Console.Error.WriteLine($"[WARN] WebSocket handshake rejected: {exception.Message}");
            }

            var baseDelay = Backoff[Math.Min(attempt, Backoff.Length - 1)];
            attempt++;
            var jitter = TimeSpan.FromMilliseconds(Random.Shared.Next(0, 750));
            var delay = baseDelay + jitter;
            Console.WriteLine($"[INFO] Reconnecting in {delay.TotalSeconds:F1}s.");
            await this.delay(delay, cancellationToken);
        }
    }

    private async Task ConnectOnceAsync(CancellationToken cancellationToken)
    {
        await using var client = await connect(config, cancellationToken);
        Console.WriteLine("[OK] Node connected.");
        report(NodeState.Online, "已连接");

        await client.SendAsync("hello", runtimePayload(), cancellationToken);
        await connected(client, cancellationToken);
        using var heartbeatStop = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var heartbeatTask = HeartbeatLoopAsync(client, heartbeatStop.Token);
        try
        {
            while (await client.ReceiveAsync(cancellationToken) is { } message)
            {
                Console.WriteLine($"[INFO] Server message: {message.Type}.");
                await received(client, message, cancellationToken);
            }
        }
        finally
        {
            heartbeatStop.Cancel();
            try
            {
                await heartbeatTask;
            }
            catch (OperationCanceledException) when (heartbeatStop.IsCancellationRequested)
            {
            }
        }
    }

    private async Task HeartbeatLoopAsync(
        NodeProtocolClient client,
        CancellationToken cancellationToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(config.HeartbeatSeconds));
        while (await timer.WaitForNextTickAsync(cancellationToken))
        {
            await client.SendAsync("heartbeat", runtimePayload(), cancellationToken);
            await heartbeat(client, cancellationToken);
        }
    }
}