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"]);
+ }
+}