feat(ablegacy): implement IBatchSubscribable

Review 02 P1 / Wave 2 rule (d): AbLegacy runs PollGroupEngine but raised only
the per-tag OnDataChange, so every changed tag paid its own trip through the
actor chain. The engine's per-cycle batch is now raised as OnDataChangeBatch
(one per poll cycle), alongside the per-tag event for legacy consumers.

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:42:23 -04:00
parent 530f4bdcd4
commit 582e82899b
2 changed files with 59 additions and 1 deletions
@@ -12,7 +12,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy;
/// <see cref="IDriver"/> only at PR 1 time; read / write / discovery / subscribe / probe /
/// host-resolver capabilities ship in PRs 2 and 3.
/// </summary>
public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable,
public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscovery, IBatchSubscribable,
IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table.</summary>
@@ -64,6 +64,12 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// </summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>
/// One raise per poll cycle carrying every tag that changed in it (review 02 P1). Raised alongside the per-tag
/// <see cref="OnDataChange"/>; a consumer subscribes to exactly one of the two.
/// </summary>
public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
/// <summary>
/// Occurs when host status changes.
/// </summary>
@@ -99,6 +105,8 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
onError: HandlePollError,
backoffCap: PollBackoffCap,
onBatch: (handle, items) =>
OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items)),
// The engine owns no Meter (Core.Abstractions takes no metrics dependency), so this
// closure IS the production consumer of its overrun signal.
onOverrun: _ => OtOpcUaTelemetry.PollOverruns.Add(1,
@@ -0,0 +1,50 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
/// <summary>
/// Review 02 P1: AbLegacy ran <see cref="PollGroupEngine"/> but did not implement <see cref="IBatchSubscribable"/>,
/// so every changed tag paid its own trip through the actor chain. It now raises one
/// <see cref="IBatchSubscribable.OnDataChangeBatch"/> per poll cycle, alongside the per-tag event.
/// </summary>
[Trait("Category", "Unit")]
public sealed class AbLegacyBatchSubscribeTests
{
private const string Host = "ab://10.0.0.5/1,0";
[Fact]
public async Task One_batch_per_poll_carries_every_changed_tag_and_per_tag_events_still_fire()
{
var factory = new FakeAbLegacyTagFactory { Customise = p => new FakeAbLegacyTag(p) { Value = p.TagName == "N7:0" ? 1 : 2 } };
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host)],
RawTags = AbLegacyRawTags.Entries(
new AbLegacyTagDefinition("A", Host, "N7:0", AbLegacyDataType.Int),
new AbLegacyTagDefinition("B", Host, "N9:0", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-batch", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
IBatchSubscribable batching = drv; // the capability the actor prefers
var batches = new System.Collections.Concurrent.ConcurrentQueue<DataChangeBatchEventArgs>();
var singles = new System.Collections.Concurrent.ConcurrentQueue<DataChangeEventArgs>();
batching.OnDataChangeBatch += (_, e) => batches.Enqueue(e);
drv.OnDataChange += (_, e) => singles.Enqueue(e);
var handle = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
var deadline = DateTime.UtcNow.AddSeconds(5);
while (batches.IsEmpty && DateTime.UtcNow < deadline) await Task.Delay(10);
batches.TryPeek(out var first).ShouldBeTrue();
first!.SubscriptionHandle.ShouldBe(handle);
first.Items.Select(i => i.FullReference).ShouldBe(new[] { "A", "B" }, ignoreOrder: true);
first.Items.Single(i => i.FullReference == "A").Snapshot.Value.ShouldBe(1);
singles.Count.ShouldBeGreaterThanOrEqualTo(2); // positive control: the legacy per-tag event still fires
await drv.UnsubscribeAsync(handle, CancellationToken.None);
await drv.ShutdownAsync(CancellationToken.None);
}
}