From adebad711fb9a154d8ae8f6b458b317a30beb1f7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:11:12 -0400 Subject: [PATCH] feat(core): PollGroupEngine optional per-cycle batch callback --- .../PollGroupEngine.cs | 27 ++- .../PollGroupEngineBatchTests.cs | 211 ++++++++++++++++++ 2 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs index e0d29342..0d8e3e20 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs @@ -24,6 +24,15 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; /// can record the failure. Without an onError callback the exception is silently /// swallowed (preserves the original behaviour for drivers that have not opted in yet). /// +/// An optional onBatch callback delivers the SAME changes collected per poll +/// cycle: it is raised at most once per cycle, only when that cycle produced at least one +/// change, and always AFTER every per-tag onChange callback of that same cycle. Drivers +/// forward it to so downstream consumers can +/// take one message per cycle instead of one per tag. Both callbacks fire by design — a +/// consumer subscribes to exactly one of the two. Without an onBatch callback the engine +/// behaves byte-identically to before: no batch is collected and no per-cycle allocation is +/// made. +/// /// Programmer errors and obviously-fatal exceptions (, /// , , /// ) are NOT caught — they propagate and tear the poll @@ -34,6 +43,7 @@ public sealed class PollGroupEngine : IAsyncDisposable private readonly Func, CancellationToken, Task>> _reader; private readonly Action _onChange; private readonly Action? _onError; + private readonly Action>? _onBatch; private readonly TimeSpan _minInterval; private readonly TimeSpan? _backoffCap; private readonly ConcurrentDictionary _subscriptions = new(); @@ -58,12 +68,18 @@ public sealed class PollGroupEngine : IAsyncDisposable /// (interval, 2×, 4×, … saturating at this cap) instead of hammering a dead device every /// tick; a successful poll resets to the base interval. null (default) keeps the /// byte-identical fixed-interval cadence for consumers that have not opted in. + /// Optional per-cycle batch callback — invoked once per poll cycle that + /// produced at least one change, carrying exactly that cycle's changed items in reader + /// order, after all of the cycle's per-tag callbacks. The + /// driver forwards it to its own event. + /// null (default) collects nothing and allocates nothing per cycle. public PollGroupEngine( Func, CancellationToken, Task>> reader, Action onChange, TimeSpan? minInterval = null, Action? onError = null, - TimeSpan? backoffCap = null) + TimeSpan? backoffCap = null, + Action>? onBatch = null) { ArgumentNullException.ThrowIfNull(reader); ArgumentNullException.ThrowIfNull(onChange); @@ -72,6 +88,7 @@ public sealed class PollGroupEngine : IAsyncDisposable _onError = onError; _minInterval = minInterval ?? DefaultMinInterval; _backoffCap = backoffCap; + _onBatch = onBatch; } /// Register a new polled subscription and start its background loop. @@ -211,6 +228,10 @@ public sealed class PollGroupEngine : IAsyncDisposable $"Reader contract violation: expected {state.TagReferences.Count} snapshots but received {snapshots.Count}. " + "The reader delegate must return one snapshot per input reference in input order."); + // Collected lazily: nothing is allocated for a cycle with no changes, and nothing at all + // when no batch callback is wired. + List? changed = null; + for (var i = 0; i < state.TagReferences.Count; i++) { var tagRef = state.TagReferences[i]; @@ -221,8 +242,12 @@ public sealed class PollGroupEngine : IAsyncDisposable { state.LastValues[tagRef] = current; _onChange(state.Handle, tagRef, current); + if (_onBatch is not null) (changed ??= new List()).Add(new DataChangeItem(tagRef, current)); } } + + // One raise per cycle, after every per-tag callback of that cycle, never empty. + if (changed is { Count: > 0 }) _onBatch!(state.Handle, changed); } /// diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs new file mode 100644 index 00000000..a2f5acfa --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs @@ -0,0 +1,211 @@ +using System.Collections.Concurrent; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Core.Tests; + +/// +/// Guards for PollGroupEngine'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 +/// . +/// +[Trait("Category", "Unit")] +public sealed class PollGroupEngineBatchTests +{ + private static readonly TimeSpan Interval = TimeSpan.FromMilliseconds(30); + private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5); + + /// + /// A reader over a mutable value map. Snapshots are returned in the requested reference + /// order (the engine's reader contract); lets a test await poll + /// cycles instead of sleeping a fixed wall-clock span. + /// + private sealed class FakeReader + { + private readonly Lock _gate = new(); + private readonly Dictionary _values = new(StringComparer.Ordinal); + private int _pollCount; + + public FakeReader(params (string Reference, object? Value)[] seed) + { + foreach (var (reference, value) in seed) _values[reference] = value; + } + + /// Gets the number of completed reader invocations. + public int PollCount => Volatile.Read(ref _pollCount); + + /// + /// 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. + /// + public void Set(params (string Reference, object? Value)[] changes) + { + lock (_gate) + { + foreach (var (reference, value) in changes) _values[reference] = value; + } + } + + /// The reader delegate handed to the engine. + public Task> ReadAsync(IReadOnlyList references, CancellationToken ct) + { + var now = DateTime.UtcNow; + var snapshots = new List(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>(snapshots); + } + } + + private static async Task WaitForAsync(Func 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}"); + } + + /// + /// 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. + /// + [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>(); + + 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); + } + + /// A cycle in which no value moved raises neither the batch nor any per-tag callback. + [Fact] + public async Task No_batch_when_nothing_changed() + { + var reader = new FakeReader(("a", 1), ("b", 2)); + var perTag = new ConcurrentQueue(); + var batches = new ConcurrentQueue>(); + + 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(); + } + + /// + /// Constructed without onBatch, the engine behaves exactly as before: per-tag + /// callbacks only, with no batch collection and no change to the raise pattern. + /// + [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(); + + 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"); + } + + /// Batch items follow the reader's reference order, not any incidental sort order. + [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>(); + + 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"]); + } +}