namespace OpenNest.Data;
///
/// Follows one server nest job to its final state for a desktop: the job's event stream
/// while it holds, polling while it does not. A dropped stream is followed by one poll (so
/// nothing is missed), then reconnects with the last event id; a server without the stream is
/// polled throughout. Reports go to sinks, so a UI receives them
/// on its own thread. See docs/server-nesting.md.
///
public sealed class NestJobWatch
{
private readonly RemoteNestJobs _jobs;
public NestJobWatch(RemoteNestJobs jobs)
{
_jobs = jobs ?? throw new ArgumentNullException(nameof(jobs));
}
/// How long to wait between polls, and before reconnecting a dropped stream.
public TimeSpan PollInterval { get; init; } = TimeSpan.FromSeconds(2);
public TimeSpan IdleTimeout { get; init; } = NestJobRules.StreamIdleTimeout;
/// Each new state of the job, the first being the one was given.
public IProgress? Job { get; init; }
/// Each new best plan so far (RemoteNestPlan JSON).
public IProgress? Plan { get; init; }
/// Why the job's updates are late, or empty once they arrive again. The job keeps running meanwhile.
public IProgress? Problem { get; init; }
/// Waits for to finish and returns its final state.
public async Task RunAsync(NestJobRecord job, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(job);
Job?.Report(job);
var lastId = 0L;
var planRevision = 0L;
var streaming = true;
while (!job.IsFinished)
{
cancellationToken.ThrowIfCancellationRequested();
if (streaming)
{
try
{
await foreach (var update in _jobs.WatchAsync(job.Id, lastId, IdleTimeout, cancellationToken))
{
switch (update.Kind)
{
case NestJobEventKind.Open:
Problem?.Report("");
break;
case NestJobEventKind.Gone:
throw Gone();
case NestJobEventKind.Plan:
lastId = System.Math.Max(lastId, update.Id);
planRevision = update.Id;
Plan?.Report(update.Plan);
break;
case NestJobEventKind.Job:
lastId = System.Math.Max(lastId, update.Id);
job = update.Job!;
Job?.Report(job);
if (job.IsFinished)
return job;
break;
}
}
Problem?.Report("The nest server's job updates stopped; reconnecting");
}
catch (ServerTooOldException)
{
streaming = false;
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested
&& ex is IOException or HttpRequestException or TaskCanceledException)
{
Problem?.Report($"Waiting for the nest server: {ex.Message}");
}
}
// Between streams, or instead of them on a server without one: one poll.
try
{
var polled = await _jobs.GetAsync(job.Id, cancellationToken) ?? throw Gone();
if (polled.PlanRevision > planRevision && await _jobs.GetPlanAsync(job.Id, cancellationToken) is { } plan
&& plan.Revision > planRevision)
{
planRevision = plan.Revision;
Plan?.Report(plan.Plan);
}
if (polled.Revision != job.Revision || polled.QueuePosition != job.QueuePosition || polled.IsFinished)
Job?.Report(polled);
job = polled;
if (!streaming)
Problem?.Report("");
if (job.IsFinished)
return job;
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested
&& ex is IOException or HttpRequestException or TaskCanceledException)
{
Problem?.Report($"Waiting for the nest server: {ex.Message}");
}
await Task.Delay(PollInterval, cancellationToken);
}
return job;
}
private static InvalidOperationException Gone() => new("The nest server no longer has this job.");
}