using System.Collections.Concurrent;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;
///
/// lmxopcua#780 (design §4.5) — the virtual-tag seam's opportunistic batching: an evaluation on an idle seam is
/// submitted at once by its own caller (no timer, no added latency), and evaluations submitted while a submission
/// is in progress go to the runner together, in arrival order, each caller getting its own outcome.
///
[Trait("Category", "Unit")]
public sealed class ScriptBatchCoalescerTests
{
private static readonly IReadOnlyDictionary NoInputs = new Dictionary();
private static ScriptInvocation Vt(string key) => new(key, ScriptContextKind.VirtualTag, "return 1;", NoInputs, TimeSpan.FromSeconds(1));
/// Answers each key with itself; the first blocks until released.
private sealed class BlockingRunner : IScriptRunner
{
private int _calls;
public ManualResetEventSlim Entered { get; } = new();
public ManualResetEventSlim Release { get; } = new();
public ConcurrentQueue Batches { get; } = new();
public bool BlockFirst { get; init; } = true;
public Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
=> throw new InvalidOperationException("the coalescer must submit through RunBatch");
public IReadOnlyList> RunBatch(IReadOnlyList invocations, CancellationToken ct = default)
{
Batches.Enqueue(invocations.Select(i => i.Key).ToArray());
if (BlockFirst && Interlocked.Increment(ref _calls) == 1)
{
Entered.Set();
Release.Wait(TimeSpan.FromSeconds(10)).ShouldBeTrue();
}
return invocations.Select(i => Task.FromResult(ScriptRunOutcome.Ok(i.Key))).ToArray();
}
public void Reset(ScriptContextKind kind) { }
public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => [];
}
[Fact]
public async Task An_evaluation_on_an_idle_seam_is_submitted_at_once_by_its_own_caller()
{
var runner = new BlockingRunner { BlockFirst = false };
var coalescer = new ScriptBatchCoalescer(runner);
var outcome = coalescer.RunAsync(Vt("only"));
outcome.IsCompletedSuccessfully.ShouldBeTrue("submitted synchronously, no timer");
(await outcome).Value.ShouldBe("only");
runner.Batches.ToArray().ShouldBe([["only"]]);
}
[Fact]
public async Task Evaluations_submitted_while_a_submission_is_in_progress_go_together_in_arrival_order()
{
var runner = new BlockingRunner();
var coalescer = new ScriptBatchCoalescer(runner);
var first = Task.Run(() => coalescer.RunAsync(Vt("first")), TestContext.Current.CancellationToken);
runner.Entered.Wait(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken).ShouldBeTrue();
var queued = Enumerable.Range(0, 10).Select(i => coalescer.RunAsync(Vt($"k{i}"))).ToArray();
queued.ShouldAllBe(t => !t.IsCompleted, "queued behind the submission in progress");
runner.Release.Set();
(await first).Value.ShouldBe("first");
var outcomes = await Task.WhenAll(queued);
outcomes.Select(o => o.Value).ShouldBe(Enumerable.Range(0, 10).Select(i => (object?)$"k{i}"));
runner.Batches.ToArray().ShouldBe([["first"], [.. Enumerable.Range(0, 10).Select(i => $"k{i}")]]);
}
[Fact]
public async Task A_runner_that_throws_does_not_strand_its_callers()
{
var coalescer = new ScriptBatchCoalescer(new ThrowingRunner());
var outcome = await coalescer.RunAsync(Vt("x"));
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
outcome.Reason!.ShouldContain("boom");
(await coalescer.RunAsync(Vt("y"))).Status.ShouldBe(ScriptRunStatus.Failed, "the flush flag was released");
}
private sealed class ThrowingRunner : IScriptRunner
{
public Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default) => throw new NotSupportedException();
public IReadOnlyList> RunBatch(IReadOnlyList invocations, CancellationToken ct = default)
=> throw new InvalidOperationException("boom");
public void Reset(ScriptContextKind kind) { }
public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => [];
}
}