using Fleet.Core;
using Fleet.Server;
using Microsoft.Extensions.DependencyInjection;
using MoShiro.Backend.Application;
using MoShiro.Protocol;
namespace MoShiro.Backend.Gateway;
///
/// Application 通往 Agent 连接的端口的实现,建立在 Fleet.Server.Link 的在线表之上。
/// Fleet 的连接宿主依赖节点事件处理器,而事件处理器依赖 Application,Application 又依赖这个端口:
/// 这里在调用时才解析 Fleet 的服务,打断构造期的循环依赖。
///
public sealed class FleetGateway : IAgentGateway
{
private readonly IServiceProvider _services;
public FleetGateway(IServiceProvider services)
{
_services = services;
}
private IFleetConnections Connections => _services.GetRequiredService();
private IStateNotifier Notifier => _services.GetRequiredService();
public IReadOnlyCollection OnlineAgents => [.. Connections.OnlineNodes.Where(n => n.Value.StartsWith(AgentIds.Prefix, StringComparison.Ordinal))];
public IReadOnlyCollection OnlineNodes => Connections.OnlineNodes;
public bool IsOnline(NodeId agent) => Connections.IsOnline(agent);
public void NotifyTaskSetChanged(NodeId agent) => Notifier.NotifyChanged(agent);
public void NotifyAllTaskSetsChanged() => Notifier.NotifyAllChanged();
public async ValueTask> RunTestAsync(NodeId agent, TestRunParams run, CancellationToken ct)
{
Result result = await Connections.RequestAsync(
agent, MoShiroMethods.TestRun, run, MoShiroJsonContext.Default.TestRunParams, MoShiroJsonContext.Default.TestRunResult, ct);
if (!result.TryGet(out TestRunResult? accepted, out FleetError? error))
{
return Result.Fail(error.Code == FleetErrors.NodeOffline ? MoShiroErrors.AgentOffline : error.Code, error.Message);
}
return accepted.Accepted ? Result.Ok(true) : Result.Fail(MoShiroErrors.TestBusy, "the agent did not accept the test");
}
public async ValueTask CancelTestAsync(NodeId agent, TestId test, CancellationToken ct) =>
await Connections.RequestAsync(
agent, MoShiroMethods.TestCancel, new TestCancelParams { TestId = test }, MoShiroJsonContext.Default.TestCancelParams, MoShiroJsonContext.Default.TestRunResult, ct);
public async ValueTask> DisconnectDialerAsync(NodeId listener, NodeId dialer, CancellationToken ct)
{
Result result = await Connections.RequestAsync(
listener, MoShiroMethods.DialerDisconnect, new DialerDisconnectParams { DialerId = dialer }, MoShiroJsonContext.Default.DialerDisconnectParams, MoShiroJsonContext.Default.TestRunResult, ct);
if (!result.TryGet(out TestRunResult? done, out FleetError? error))
{
return Result.Fail(error.Code == FleetErrors.NodeOffline ? MoShiroErrors.AgentOffline : error.Code, error.Message);
}
return Result.Ok(done.Accepted);
}
public ValueTask DisconnectAsync(NodeId agent, string reason, CancellationToken ct) => Connections.DisconnectAsync(agent, reason, ct);
public ValueTask NotifyLinkRetryAsync(NodeId dialer, NodeId listener, CancellationToken ct) =>
Connections.NotifyAsync(dialer, MoShiroMethods.LinkRetry, new LinkRetryParams { ListenerId = listener }, MoShiroJsonContext.Default.LinkRetryParams, ct);
public async ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct) =>
await Connections.NotifyAsync(node, MoShiroMethods.UpdateOffer, new UpdateOfferParams { Manifest = manifest, Signature = signature }, MoShiroJsonContext.Default.UpdateOfferParams, ct);
public async ValueTask NotifyScheduleAsync(NodeId node, string? restartAt, CancellationToken ct) =>
await Connections.NotifyAsync(node, MoShiroMethods.Schedule, new ScheduleParams { RestartAt = restartAt }, MoShiroJsonContext.Default.ScheduleParams, ct);
public async ValueTask NotifyUpdateApplyAsync(NodeId node, CancellationToken ct) =>
await Connections.NotifyAsync(node, MoShiroMethods.UpdateApply, new UpdateOfferParams(), MoShiroJsonContext.Default.UpdateOfferParams, ct);
}