124 lines
4.1 KiB
C#
124 lines
4.1 KiB
C#
using System.Text.Json;
|
|
using Fleet.Core;
|
|
using Fleet.Rpc;
|
|
|
|
namespace Fleet.Server;
|
|
|
|
/// <summary>状态：期望状态推送、在线状态、节点事件。</summary>
|
|
internal sealed partial class NodeConnection
|
|
{
|
|
public void SignalStateChanged() => _stateSignal.Writer.TryWrite(true);
|
|
|
|
|
|
private async Task StatePushLoopAsync(CancellationToken ct)
|
|
{
|
|
while (await _stateSignal.Reader.WaitToReadAsync(ct))
|
|
{
|
|
_stateSignal.Reader.TryRead(out _);
|
|
bool complete = await PushStateAsync(ct);
|
|
if (!complete)
|
|
{
|
|
await Task.Delay(s_statePushRetry, _host.Time, ct);
|
|
SignalStateChanged();
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task<bool> PushStateAsync(CancellationToken ct)
|
|
{
|
|
NodeRecord? fresh = await _host.Store.GetNodeAsync(_node!.Id, ct);
|
|
if (fresh is null)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
_node = fresh with { NodeVersion = _node.NodeVersion ?? fresh.NodeVersion };
|
|
if (fresh.Status == NodeStatus.Disabled)
|
|
{
|
|
await RevokeAsync("disabled", ct);
|
|
return true;
|
|
}
|
|
|
|
if (fresh.Status != NodeStatus.Active || !_helloDone)
|
|
{
|
|
// pending 不下发状态；还没有 hello 的连接等 hello 之后再推送。
|
|
return true;
|
|
}
|
|
|
|
bool complete = true;
|
|
foreach (IStateProvider provider in _host.StateProviders)
|
|
{
|
|
StateDocument? document = await provider.GetAsync(_node, ct);
|
|
if (document is null || (_lastPushedVersion.TryGetValue(provider.Type, out long last) && last == document.Version))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
Result<JsonElement> raw = await RequestAsync(
|
|
FleetMethods.StateSync,
|
|
JsonSerializer.SerializeToElement(
|
|
new StateSyncParams { Type = provider.Type, Version = document.Version, Doc = document.Doc },
|
|
FleetRpcContext.Default.StateSyncParams),
|
|
ct);
|
|
if (!raw.TryGet(out JsonElement value, out FleetError? error))
|
|
{
|
|
FleetLog.StatePushFailed(_host.Logger, _node.Id.Value, provider.Type, error.Code.Value);
|
|
complete = false;
|
|
continue;
|
|
}
|
|
|
|
StateSyncResult? result = null;
|
|
try
|
|
{
|
|
result = value.Deserialize(FleetRpcContext.Default.StateSyncResult);
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 按缺失处理
|
|
}
|
|
|
|
if (result is null)
|
|
{
|
|
complete = false;
|
|
continue;
|
|
}
|
|
|
|
_lastPushedVersion[provider.Type] = document.Version;
|
|
await provider.OnAppliedAsync(_node, document.Version, result, ct);
|
|
}
|
|
|
|
return complete;
|
|
}
|
|
|
|
|
|
private async Task UpdatePresenceAsync(bool online, CancellationToken ct)
|
|
{
|
|
Interlocked.Exchange(ref _lastPresenceTicks, _host.Time.GetTimestamp());
|
|
await _host.Store.UpdatePresenceAsync(
|
|
_node!.Id,
|
|
new Presence(online, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), _session?.NodeVersion, _session?.Capabilities ?? [], _session?.HelloMetaJson),
|
|
ct);
|
|
}
|
|
|
|
private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null)
|
|
{
|
|
var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra, kind == NodeEventKind.Offline ? _closeReason : null, _remoteIp);
|
|
foreach (INodeEventSink sink in _host.Sinks)
|
|
{
|
|
try
|
|
{
|
|
await sink.OnEventAsync(nodeEvent, ct);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
|
{
|
|
throw;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// 产品的事件处理失败不能影响连接。
|
|
FleetLog.SinkFailed(_host.Logger, ex, kind.ToString(), _node!.Id.Value);
|
|
}
|
|
}
|
|
}
|
|
}
|