lmxopcua/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs

using System.Collections.Concurrent;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Core.Tests;

/// <summary>
///     Guards for <c>PollGroupEngine</c>'s optional per-cycle batch callback: one raise per poll
///     cycle carrying exactly the items that changed in that cycle, raised after the per-tag
///     callbacks of the same cycle and only when non-empty. Drivers forward it to
///     <see cref="IBatchSubscribable.OnDataChangeBatch"/>.
/// </summary>
[Trait("Category", "Unit")]
public sealed class PollGroupEngineBatchTests
{
    private static readonly TimeSpan Interval = TimeSpan.FromMilliseconds(30);
    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5);

    /// <summary>
    ///     A reader over a mutable value map. Snapshots are returned in the requested reference
    ///     order (the engine's reader contract); <see cref="PollCount"/> lets a test await poll
    ///     cycles instead of sleeping a fixed wall-clock span.
    /// </summary>
    private sealed class FakeReader
    {
        private readonly Lock _gate = new();
        private readonly Dictionary<string, object?> _values = new(StringComparer.Ordinal);
        private int _pollCount;

        public FakeReader(params (string Reference, object? Value)[] seed)
        {
            foreach (var (reference, value) in seed) _values[reference] = value;
        }

        /// <summary>Gets the number of completed reader invocations.</summary>
        public int PollCount => Volatile.Read(ref _pollCount);

        /// <summary>
        ///     Overwrites the values a subsequent poll will observe. Applied under the same lock the
        ///     reader holds, so a multi-reference change is observed by ONE poll cycle rather than
        ///     being torn across two.
        /// </summary>
        public void Set(params (string Reference, object? Value)[] changes)
        {
            lock (_gate)
            {
                foreach (var (reference, value) in changes) _values[reference] = value;
            }
        }

        /// <summary>The reader delegate handed to the engine.</summary>
        public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IReadOnlyList<string> references, CancellationToken ct)
        {
            var now = DateTime.UtcNow;
            var snapshots = new List<DataValueSnapshot>(references.Count);
            lock (_gate)
            {
                foreach (var reference in references)
                    snapshots.Add(new DataValueSnapshot(_values.GetValueOrDefault(reference), 0u, now, now));
            }
            Interlocked.Increment(ref _pollCount);
            return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(snapshots);
        }
    }

    private static async Task WaitForAsync(Func<bool> condition, string because)
    {
        var deadline = DateTime.UtcNow + WaitTimeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(10);
        }
        throw new TimeoutException($"Timed out after {WaitTimeout.TotalSeconds}s waiting for: {because}");
    }

    /// <summary>
    ///     The initial-data cycle batches every subscribed tag; a later cycle in which exactly one
    ///     value moved batches exactly that one item — and the per-tag callback fired for the same
    ///     items, so both callbacks see identical content.
    /// </summary>
    [Fact]
    public async Task Batch_fires_once_per_cycle_with_only_changed_items()
    {
        var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3));
        var perTag = new ConcurrentQueue<(string Reference, DataValueSnapshot Snapshot)>();
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, snapshot) => perTag.Enqueue((reference, snapshot)),
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["a", "b", "c"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        batches.TryDequeue(out var initial).ShouldBeTrue();
        initial!.Count.ShouldBe(3);
        initial.Select(i => i.FullReference).ShouldBe(["a", "b", "c"]);
        perTag.Count.ShouldBe(3);

        perTag.Clear();
        batches.Clear();
        reader.Set(("b", 99));

        await WaitForAsync(() => batches.Count >= 1, "the batch for the single changed tag");
        batches.TryDequeue(out var changed).ShouldBeTrue();
        changed!.Count.ShouldBe(1);
        changed[0].FullReference.ShouldBe("b");
        changed[0].Snapshot.Value.ShouldBe(99);

        // The per-tag callback saw the same single item, with the same snapshot instance.
        perTag.Count.ShouldBe(1);
        perTag.TryDequeue(out var perTagItem).ShouldBeTrue();
        perTagItem.Reference.ShouldBe("b");
        perTagItem.Snapshot.ShouldBeSameAs(changed[0].Snapshot);
    }

    /// <summary>A cycle in which no value moved raises neither the batch nor any per-tag callback.</summary>
    [Fact]
    public async Task No_batch_when_nothing_changed()
    {
        var reader = new FakeReader(("a", 1), ("b", 2));
        var perTag = new ConcurrentQueue<string>();
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, _) => perTag.Enqueue(reference),
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["a", "b"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        var pollsAfterInitial = reader.PollCount;
        perTag.Clear();
        batches.Clear();

        // Several further cycles over identical snapshots, plus a grace window for the callbacks
        // of the last counted poll to have run.
        await WaitForAsync(() => reader.PollCount >= pollsAfterInitial + 5, "five further poll cycles");
        await Task.Delay(50, TestContext.Current.CancellationToken);

        batches.ShouldBeEmpty();
        perTag.ShouldBeEmpty();
    }

    /// <summary>
    ///     Constructed without <c>onBatch</c>, the engine behaves exactly as before: per-tag
    ///     callbacks only, with no batch collection and no change to the raise pattern.
    /// </summary>
    [Fact]
    public async Task Batch_is_not_invoked_when_callback_absent()
    {
        var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3));
        var perTag = new ConcurrentQueue<string>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, _) => perTag.Enqueue(reference),
            minInterval: Interval);

        engine.Subscribe(["a", "b", "c"], Interval);

        await WaitForAsync(() => perTag.Count >= 3, "the initial-data per-tag callbacks");
        perTag.Count.ShouldBe(3);

        reader.Set(("c", 42));
        await WaitForAsync(() => perTag.Count >= 4, "the per-tag callback for the changed tag");

        // Let further cycles run: nothing else changed, so the count must stay at 4.
        var polls = reader.PollCount;
        await WaitForAsync(() => reader.PollCount >= polls + 3, "three further poll cycles");
        await Task.Delay(50, TestContext.Current.CancellationToken);
        perTag.Count.ShouldBe(4);
        perTag.ToArray()[3].ShouldBe("c");
    }

    /// <summary>Batch items follow the reader's reference order, not any incidental sort order.</summary>
    [Fact]
    public async Task Batch_items_are_in_reader_order()
    {
        // Deliberately non-alphabetical subscription order: an alphabetical (or insertion-ordered
        // dictionary) collection would produce z/a/m in a different sequence.
        var reader = new FakeReader(("z", 1), ("a", 2), ("m", 3));
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, _, _) => { },
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["z", "a", "m"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        batches.TryDequeue(out var initial).ShouldBeTrue();
        initial!.Select(i => i.FullReference).ShouldBe(["z", "a", "m"]);

        // A cycle in which two of the three moved keeps the reader's relative order.
        batches.Clear();
        reader.Set(("m", 30), ("z", 10));

        await WaitForAsync(() => batches.Count >= 1, "the batch for the two changed tags");
        batches.TryDequeue(out var changed).ShouldBeTrue();
        changed!.Select(i => i.FullReference).ShouldBe(["z", "m"]);
    }
}