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

feat(runtime): DriverInstanceActor forwards batched data changes

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:16:49 -04:00
parent 04e6a4690c
commit e04ff2488c
5 changed files with 332 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs
+226
View File
@@ -0,0 +1,226 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Covers the ingress seam of the value-pipeline batching work: <see cref="DriverInstanceActor"/>
/// forwards data changes to its parent as <see cref="DriverInstanceActor.AttributeValueBatchPublished"/>
/// — ONE message per driver production burst.
/// <para>A driver implementing <see cref="IBatchSubscribable"/> gets the batch handler attached and the
/// per-tag <see cref="ISubscribable.OnDataChange"/> handler NOT attached: such a driver raises BOTH
/// events per the contract, so attaching both would double-deliver every value. A per-tag-only
/// <see cref="ISubscribable"/> driver keeps working unchanged — each raise is wrapped as a
/// single-item batch, so the parent sees exactly one message shape either way.</para>
/// <para>Status codes ride verbatim (#519); timestamps come from the driver's snapshot
/// (<c>SourceTimestampUtc ?? ServerTimestampUtc</c>), never from the actor's clock.</para>
/// </summary>
public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase
{
/// <summary>
/// A batch-capable driver's burst of 3 changed tags reaches the parent as exactly ONE
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> carrying all 3 items, each mapped
/// verbatim (reference / value / status code) with the timestamp derived from the snapshot.
/// </summary>
[Fact]
public async Task Batch_driver_forwards_one_batch_message()
{
var driver = new BatchSubscribableStubDriver();
var parent = CreateTestProbe();
// #477: DriverInstanceActor Tells the parent ConnectivityChanged on each connect/disconnect;
// this suite asserts on the value forwards only, so ignore it.
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b", "tag-c" }, TimeSpan.FromMilliseconds(100)),
TimeSpan.FromSeconds(3));
var sourceTs = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
var serverTs = new DateTime(2026, 8, 15, 10, 0, 5, DateTimeKind.Utc);
driver.FireBatch(includePerTagRaises: false,
new DataChangeItem("tag-a", new DataValueSnapshot(3.14, OpcUaStatus.Good, sourceTs, serverTs)),
new DataChangeItem("tag-b", new DataValueSnapshot(7, 0x80740000u /* BadTypeMismatch */, sourceTs.AddSeconds(1), serverTs)),
// Null source timestamp — the server timestamp is the fallback, per OnDataChangeForward.
new DataChangeItem("tag-c", new DataValueSnapshot(null, OpcUaStatus.BadNoCommunication, null, serverTs)));
var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
batch.DriverInstanceId.ShouldBe(driver.DriverInstanceId);
batch.Items.Count.ShouldBe(3);
batch.Items[0].FullReference.ShouldBe("tag-a");
batch.Items[0].Value.ShouldBe(3.14);
batch.Items[0].StatusCode.ShouldBe(OpcUaStatus.Good);
batch.Items[0].TimestampUtc.ShouldBe(sourceTs);
batch.Items[1].FullReference.ShouldBe("tag-b");
batch.Items[1].Value.ShouldBe(7);
// #519: the sub-code rides through, not a collapsed OpcUaStatus.Bad.
batch.Items[1].StatusCode.ShouldBe(0x80740000u);
batch.Items[1].TimestampUtc.ShouldBe(sourceTs.AddSeconds(1));
batch.Items[2].FullReference.ShouldBe("tag-c");
batch.Items[2].Value.ShouldBeNull();
batch.Items[2].StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
batch.Items[2].TimestampUtc.ShouldBe(serverTs, "a null SourceTimestampUtc falls back to the server timestamp");
// Exactly ONE message for the burst — no per-item trailer.
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>
/// An <see cref="IBatchSubscribable"/> driver raises BOTH the batch event and the per-tag
/// <see cref="ISubscribable.OnDataChange"/> for every item (that is the documented driver contract,
/// kept for legacy consumers). The actor must attach ONE handler only, so the parent sees the batch
/// and nothing else — attaching both would publish every value twice.
/// </summary>
[Fact]
public async Task Batch_driver_per_tag_event_not_double_forwarded()
{
var driver = new BatchSubscribableStubDriver();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100)),
TimeSpan.FromSeconds(3));
var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
driver.FireBatch(includePerTagRaises: true,
new DataChangeItem("tag-a", new DataValueSnapshot(1, OpcUaStatus.Good, ts, ts)),
new DataChangeItem("tag-b", new DataValueSnapshot(2, OpcUaStatus.Good, ts, ts)));
var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
batch.Items.Count.ShouldBe(2);
// The per-tag raises for the SAME burst produced nothing extra — no second batch, no
// AttributeValuePublished. Anything here means both handlers were attached.
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
}
/// <summary>
/// A driver that implements only <see cref="ISubscribable"/> keeps its per-tag handler and each raise
/// arrives at the parent as a single-item batch — the whole fleet minus the batch-capable drivers
/// stays on one message shape.
/// </summary>
[Fact]
public async Task Per_tag_driver_forwards_single_item_batches()
{
var driver = new SubscribableStubDriver();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
TimeSpan.FromSeconds(3));
driver.FireDataChange("tag-a", value: 3.14, statusCode: OpcUaStatus.Good);
var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
batch.DriverInstanceId.ShouldBe(driver.DriverInstanceId);
batch.Items.Count.ShouldBe(1);
batch.Items[0].FullReference.ShouldBe("tag-a");
batch.Items[0].Value.ShouldBe(3.14);
batch.Items[0].StatusCode.ShouldBe(OpcUaStatus.Good);
}
/// <summary>
/// The detach is load-bearing — the <see cref="IDriver"/> object outlives the actor across respawns,
/// so a batch handler left attached would push into a dead/stale actor. After Unsubscribe, a burst
/// produces nothing at the parent.
/// </summary>
[Fact]
public async Task Detach_removes_batch_handler()
{
var driver = new BatchSubscribableStubDriver();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
TimeSpan.FromSeconds(3));
driver.OnDataChangeBatchSubscriberCount.ShouldBe(1, "the batch handler is attached for an IBatchSubscribable driver");
actor.Tell(new DriverInstanceActor.Unsubscribe());
AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget);
var ts = new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc);
driver.FireBatch(includePerTagRaises: true,
new DataChangeItem("tag-a", new DataValueSnapshot(42, OpcUaStatus.Good, ts, ts)));
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
}
/// <summary>
/// A <see cref="StubDriver"/> that implements <see cref="IBatchSubscribable"/> — it raises
/// <see cref="OnDataChangeBatch"/> once per burst AND (optionally, per the documented driver
/// contract) the per-tag <see cref="OnDataChange"/> for each item of that same burst.
/// </summary>
private sealed class BatchSubscribableStubDriver : StubDriver, IBatchSubscribable
{
private readonly StubHandle _handle = new();
/// <summary>Occurs when a single attribute changes (legacy per-tag contract).</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>Occurs once per production burst with every attribute that changed in it.</summary>
public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
/// <summary>Gets the number of handlers attached to <see cref="OnDataChange"/>.</summary>
public int OnDataChangeSubscriberCount => OnDataChange?.GetInvocationList().Length ?? 0;
/// <summary>Gets the number of handlers attached to <see cref="OnDataChangeBatch"/>.</summary>
public int OnDataChangeBatchSubscriberCount => OnDataChangeBatch?.GetInvocationList().Length ?? 0;
/// <summary>Subscribes to the specified full references.</summary>
/// <param name="fullReferences">The full references to subscribe to.</param>
/// <param name="publishingInterval">The publishing interval.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
Task.FromResult<ISubscriptionHandle>(_handle);
/// <summary>Unsubscribes from the specified subscription handle.</summary>
/// <param name="handle">The subscription handle.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) =>
Task.CompletedTask;
/// <summary>Raises one production burst.</summary>
/// <param name="includePerTagRaises">
/// When true, also raises the per-tag <see cref="OnDataChange"/> for every item — the documented
/// contract a real batch-capable driver honours for legacy consumers.
/// </param>
/// <param name="items">The changed attributes of this burst.</param>
public void FireBatch(bool includePerTagRaises, params DataChangeItem[] items)
{
OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(_handle, items));
if (!includePerTagRaises) return;
foreach (var item in items)
{
OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, item.FullReference, item.Snapshot));
}
}
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs
+5 -5
View File
@@ -15,7 +15,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <see cref="IAlarmSource.OnAlarmEvent"/> (when the driver is an <see cref="IAlarmSource"/>)
/// and forwards every transition to its parent as
/// <see cref="DriverInstanceActor.AttributeAlarmPublished"/> — mirroring the OnDataChange →
/// <see cref="DriverInstanceActor.AttributeValuePublished"/> forward. The driver fires
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> forward. The driver fires
/// <c>OnAlarmEvent</c> on its OWN thread; the actor marshals it onto the actor thread via Self.
/// </summary>
public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
@@ -63,8 +63,8 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
// The same driver's OnDataChange still flows independently — alarm + value events coexist.
driver.FireDataChange("tag-z", value: 9.0, statusCode: 0u);
parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(2))
.FullReference.ShouldBe("tag-z");
parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2))
.Items[0].FullReference.ShouldBe("tag-z");
}
/// <summary>
@@ -131,8 +131,8 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
driver.FireDataChange("tag-a", value: 1.5, statusCode: 0u);
// Data-change forwarding still works (no crash on AttachAlarmSource for a non-IAlarmSource driver).
var dc = parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(2));
dc.FullReference.ShouldBe("tag-a");
var dc = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
dc.Items[0].FullReference.ShouldBe("tag-a");
// An AttributeAlarmPublished can never be produced for a non-IAlarmSource driver.
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs
+17 -11
View File
@@ -121,14 +121,16 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(250)),
TimeSpan.FromSeconds(3));
// Driver fires an OnDataChange — actor should forward it to its parent as
// AttributeValuePublished carrying the driver's status code.
// Driver fires an OnDataChange — actor should forward it to its parent as an
// AttributeValueBatchPublished (single-item, since this driver is per-tag-only) carrying the
// driver's status code.
driver.FireDataChange("tag-a", value: 3.14, statusCode: 0u);
var published = parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(2));
published.FullReference.ShouldBe("tag-a");
published.Value.ShouldBe(3.14);
published.StatusCode.ShouldBe(OpcUaStatus.Good);
var published = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
published.Items.Count.ShouldBe(1);
published.Items[0].FullReference.ShouldBe("tag-a");
published.Items[0].Value.ShouldBe(3.14);
published.Items[0].StatusCode.ShouldBe(OpcUaStatus.Good);
}
/// <summary>
@@ -162,17 +164,19 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
// Uncertain — severity bits 01 (top 2 bits = 01).
driver.FireDataChange("tag-1", value: 1, statusCode: 0x40000000u);
parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>().StatusCode.ShouldBe(OpcUaStatus.Uncertain);
parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>().Items[0].StatusCode
.ShouldBe(OpcUaStatus.Uncertain);
// Bad — severity bits 10.
driver.FireDataChange("tag-1", value: 2, statusCode: 0x80000000u);
parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>().StatusCode.ShouldBe(OpcUaStatus.Bad);
parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>().Items[0].StatusCode
.ShouldBe(OpcUaStatus.Bad);
// The regression itself: a Bad SUB-code must arrive intact, not flattened to OpcUaStatus.Bad.
// This is the exact code MTConnect maps the Agent's UNAVAILABLE sentinel to, and the live gate
// that opened #519 observed it reaching a client as 0x80000000.
driver.FireDataChange("tag-1", value: null, statusCode: OpcUaStatus.BadNoCommunication);
var noComms = parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>();
var noComms = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>().Items[0];
noComms.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
noComms.StatusCode.ShouldNotBe(OpcUaStatus.Bad, "the sub-code was collapsed to a generic Bad");
OpcUaStatus.IsBad(noComms.StatusCode).ShouldBeTrue("it must still classify as Bad");
@@ -180,7 +184,8 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
// A second, DIFFERENT Bad sub-code — the two were indistinguishable before the fix, so asserting
// one alone would not prove the sub-code is carried rather than hard-coded.
driver.FireDataChange("tag-1", value: null, statusCode: 0x80740000u /* BadTypeMismatch */);
parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>().StatusCode.ShouldBe(0x80740000u);
parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>().Items[0].StatusCode
.ShouldBe(0x80740000u);
}
/// <summary>
@@ -209,7 +214,8 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
// The auto-subscription is live — a data change reaches the parent.
driver.FireDataChange("tag-a", value: 7, statusCode: 0u);
parent.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(2)).Value.ShouldBe(7);
parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2))
.Items[0].Value.ShouldBe(7);
// Reconnect → the desired set is re-established without any new host message.
actor.Tell(new DriverInstanceActor.DisconnectObserved("backend blip"));