dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): PollGroupEngine optional per-cycle batch callback

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:11:12 -04:00
parent bb3ccf264b
commit adebad711f
2 changed files with 237 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs
+26 -1
View File
@@ -24,6 +24,15 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// can record the failure. Without an <c>onError</c> callback the exception is silently
/// swallowed (preserves the original behaviour for drivers that have not opted in yet).</para>
///
/// <para>An optional <c>onBatch</c> 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 <c>onChange</c> callback of that same cycle. Drivers
/// forward it to <see cref="IBatchSubscribable.OnDataChangeBatch"/> 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 <c>onBatch</c> callback the engine
/// behaves byte-identically to before: no batch is collected and no per-cycle allocation is
/// made.</para>
///
/// <para>Programmer errors and obviously-fatal exceptions (<see cref="OutOfMemoryException"/>,
/// <see cref="ThreadAbortException"/>, <see cref="StackOverflowException"/>,
/// <see cref="AccessViolationException"/>) are NOT caught — they propagate and tear the poll
@@ -34,6 +43,7 @@ public sealed class PollGroupEngine : IAsyncDisposable
private readonly Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> _reader;
private readonly Action<ISubscriptionHandle, string, DataValueSnapshot> _onChange;
private readonly Action<Exception>? _onError;
private readonly Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? _onBatch;
private readonly TimeSpan _minInterval;
private readonly TimeSpan? _backoffCap;
private readonly ConcurrentDictionary<long, SubscriptionState> _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. <c>null</c> (default) keeps the
/// byte-identical fixed-interval cadence for consumers that have not opted in.</param>
/// <param name="onBatch">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 <paramref name="onChange"/> callbacks. The
/// driver forwards it to its own <see cref="IBatchSubscribable.OnDataChangeBatch"/> event.
/// <c>null</c> (default) collects nothing and allocates nothing per cycle.</param>
public PollGroupEngine(
Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> reader,
Action<ISubscriptionHandle, string, DataValueSnapshot> onChange,
TimeSpan? minInterval = null,
Action<Exception>? onError = null,
TimeSpan? backoffCap = null)
TimeSpan? backoffCap = null,
Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? 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;
}
/// <summary>Register a new polled subscription and start its background loop.</summary>
@@ -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<DataChangeItem>? 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<DataChangeItem>()).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);
}
/// <summary>
tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs
+211
View File
@@ -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;
/// <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"]);
}
}