diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 4b865dab..6310b9da 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -924,6 +924,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Receive(HandleDispatchFromSteady); Receive(HandleGetDiagnostics); Receive(ForwardToMux); + Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleDeltaApplied); @@ -958,6 +959,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Receive(HandleFetchedForApply); Receive(HandleGetDiagnostics); Receive(ForwardToMux); + Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleDeltaApplied); @@ -973,6 +975,22 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Receive(_ => { /* PubSub ack */ }); } + /// + /// Temporary bridge: unwrap a batched driver publish back into the per-value + /// path, so the batching ingress seam lands without changing any + /// downstream behaviour. + /// + // TODO(Task 8 plan 2026-08-15): replace with batched fan-out — one mux hop and one OPC UA publish + // hop per burst instead of one per value, which is the whole point of the batching work. + private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished batch) + { + foreach (var i in batch.Items) + { + ForwardToMux(new DriverInstanceActor.AttributeValuePublished( + batch.DriverInstanceId, i.FullReference, i.Value, i.StatusCode, i.TimestampUtc)); + } + } + private void ForwardToMux(DriverInstanceActor.AttributeValuePublished msg) { // Pass driver-published values to the dependency mux when one is wired (VirtualTag inputs, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 4ab6af32..4bb49819 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -101,13 +101,25 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// state — a no-op when already Reconnecting or Connecting. /// public sealed record ForceReconnect; - /// Published to the actor's parent whenever the subscribed IDriver fires - /// . The parent forwards to OpcUaPublishActor. + /// One driver-published value. NOT sent to the parent by the subscription path any more — the + /// actor batches every data change into an . Still live as the + /// per-value message the parent fans onward (the dependency mux consumes it, and + /// DriverHostActor.FanCommsLossToDependencyMux fabricates it). /// is the driver's full OPC UA status code including its sub-code /// (#519) — not a severity bucket. Classify it with and friends rather /// than comparing it to . public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); + /// One value within an . + public sealed record AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); + /// Batched form of — every changed attribute of ONE + /// driver production burst in one message. The parent fans the whole batch to the mux and the OPC UA + /// publish path in one hop each. Status codes ride verbatim (#519). + public sealed record AttributeValueBatchPublished(string DriverInstanceId, IReadOnlyList Items); private sealed record DataChangeForward(string FullReference, DataValueSnapshot Snapshot); + /// Self-sent when a batch-capable driver () raises + /// — marshals the whole burst off the driver's + /// thread onto the actor thread in one hop. Parallels . + private sealed record DataChangeBatchForward(IReadOnlyList Items); /// Published to the parent whenever the subscribed driver (an ) fires /// . The parent () projects + routes it /// to the materialised Part 9 condition. Parallels . @@ -205,6 +217,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// do not need to re-send subscription requests after a reconnect. private ISubscriptionHandle? _subscriptionHandle; private EventHandler? _dataChangeHandler; + /// The batch data-change handler, attached INSTEAD of when the + /// driver is an . Exactly one of the two is ever non-null. + private EventHandler? _batchChangeHandler; private EventHandler? _alarmEventHandler; private EventHandler? _rediscoveryHandler; private EventHandler? _hostStatusHandler; @@ -478,8 +493,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers }); ReceiveAsync(HandleSubscribeAlarmsAsync); Receive(OnDataChangeForward); + Receive(OnDataChangeBatchForward); // Native alarm transition marshaled onto the actor thread from the driver's OnAlarmEvent; - // project it to the parent the same way DataChangeForward projects AttributeValuePublished. + // project it to the parent the same way DataChangeForward projects AttributeValueBatchPublished. Receive(m => Context.Parent.Tell(new AttributeAlarmPublished(_driverInstanceId, m.Args))); // ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the // sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter. @@ -877,8 +893,21 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers try { - _dataChangeHandler = (_, args) => self.Tell(new DataChangeForward(args.FullReference, args.Snapshot)); - subscribable.OnDataChange += _dataChangeHandler; + // EXACTLY ONE of the two data-change events is subscribed. An IBatchSubscribable driver raises + // OnDataChangeBatch once per production burst AND the per-tag OnDataChange for every item in + // that same burst (the per-tag event is kept for legacy consumers — Galaxy cold reads, driver + // CLIs). Attaching both here would forward every value twice. The batch event is preferred: one + // Tell, one parent hop, one mux pass per burst instead of per tag. + if (subscribable is IBatchSubscribable batchable) + { + _batchChangeHandler = (_, args) => self.Tell(new DataChangeBatchForward(args.Items)); + batchable.OnDataChangeBatch += _batchChangeHandler; + } + else + { + _dataChangeHandler = (_, args) => self.Tell(new DataChangeForward(args.FullReference, args.Snapshot)); + subscribable.OnDataChange += _dataChangeHandler; + } using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); // A bulk subscribe spans potentially many hosts, so key the pipeline on the driver @@ -928,9 +957,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } } - /// Tear down the data-change + native-alarm event handlers + null the handle. Called from the - /// Unsubscribe path, on PostStop, and on Connected → Reconnecting transitions so a stale handler doesn't - /// push data-change / alarm events to an actor that has lost its driver connection. + /// Tear down the data-change (per-tag AND batch) + native-alarm event handlers + null the handle. + /// Called from the Unsubscribe path, on PostStop, and on Connected → Reconnecting transitions so a stale + /// handler doesn't push data-change / alarm events to an actor that has lost its driver connection. + /// Both data-change fields are cleared unconditionally even though only one is ever attached — the + /// IDriver object outlives the actor when the host respawns a child around it, so a handler left on the + /// driver's event would keep Telling a dead actor. private void DetachSubscription() { if (_driver is ISubscribable subscribable && _dataChangeHandler is not null) @@ -938,6 +970,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers subscribable.OnDataChange -= _dataChangeHandler; } _dataChangeHandler = null; + if (_driver is IBatchSubscribable batchable && _batchChangeHandler is not null) + { + batchable.OnDataChangeBatch -= _batchChangeHandler; + } + _batchChangeHandler = null; _subscriptionHandle = null; DetachAlarmSource(); } @@ -1130,7 +1167,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } } - /// Forward a driver's data change to the parent for OPC UA publication. + /// Forward a per-tag-only driver's data change to the parent for OPC UA publication, as a + /// single-item batch — so the parent sees ONE message shape whether or not the driver batches. /// The driver's status code rides through UNCHANGED (#519). This used to project it onto a /// 3-state Good/Uncertain/Bad enum here, which the node manager then re-expanded to a bare /// 0x80000000 — so no driver's sub-code ever reached a client, and the status constants each @@ -1138,8 +1176,25 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private void OnDataChangeForward(DataChangeForward msg) { var ts = msg.Snapshot.SourceTimestampUtc ?? msg.Snapshot.ServerTimestampUtc; - Context.Parent.Tell(new AttributeValuePublished( - _driverInstanceId, msg.FullReference, msg.Snapshot.Value, msg.Snapshot.StatusCode, ts)); + Context.Parent.Tell(new AttributeValueBatchPublished( + _driverInstanceId, + new[] { new AttributeValueItem(msg.FullReference, msg.Snapshot.Value, msg.Snapshot.StatusCode, ts) })); + } + + /// Forward one batch-capable driver production burst to the parent as a single message. + /// Every item's status code and timestamp ride through exactly as + /// carries them for a per-tag driver (#519; timestamps come from the + /// driver's snapshot, never from this actor's clock). + private void OnDataChangeBatchForward(DataChangeBatchForward msg) + { + var items = new AttributeValueItem[msg.Items.Count]; + for (var i = 0; i < msg.Items.Count; i++) + { + var item = msg.Items[i]; + var ts = item.Snapshot.SourceTimestampUtc ?? item.Snapshot.ServerTimestampUtc; + items[i] = new AttributeValueItem(item.FullReference, item.Snapshot.Value, item.Snapshot.StatusCode, ts); + } + Context.Parent.Tell(new AttributeValueBatchPublished(_driverInstanceId, items)); } private static bool IsGoodStatus(uint statusCode) => OpcUaStatus.IsGood(statusCode); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs new file mode 100644 index 00000000..71a45b29 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs @@ -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; + +/// +/// Covers the ingress seam of the value-pipeline batching work: +/// forwards data changes to its parent as +/// — ONE message per driver production burst. +/// A driver implementing gets the batch handler attached and the +/// per-tag handler NOT attached: such a driver raises BOTH +/// events per the contract, so attaching both would double-deliver every value. A per-tag-only +/// driver keeps working unchanged — each raise is wrapped as a +/// single-item batch, so the parent sees exactly one message shape either way. +/// Status codes ride verbatim (#519); timestamps come from the driver's snapshot +/// (SourceTimestampUtc ?? ServerTimestampUtc), never from the actor's clock. +/// +public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase +{ + /// + /// A batch-capable driver's burst of 3 changed tags reaches the parent as exactly ONE + /// carrying all 3 items, each mapped + /// verbatim (reference / value / status code) with the timestamp derived from the snapshot. + /// + [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( + 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(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)); + } + + /// + /// An driver raises BOTH the batch event and the per-tag + /// 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. + /// + [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( + 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(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)); + } + + /// + /// A driver that implements only 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. + /// + [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( + 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(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); + } + + /// + /// The detach is load-bearing — the 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. + /// + [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( + 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)); + } + + /// + /// A that implements — it raises + /// once per burst AND (optionally, per the documented driver + /// contract) the per-tag for each item of that same burst. + /// + private sealed class BatchSubscribableStubDriver : StubDriver, IBatchSubscribable + { + private readonly StubHandle _handle = new(); + + /// Occurs when a single attribute changes (legacy per-tag contract). + public event EventHandler? OnDataChange; + + /// Occurs once per production burst with every attribute that changed in it. + public event EventHandler? OnDataChangeBatch; + + /// Gets the number of handlers attached to . + public int OnDataChangeSubscriberCount => OnDataChange?.GetInvocationList().Length ?? 0; + + /// Gets the number of handlers attached to . + public int OnDataChangeBatchSubscriberCount => OnDataChangeBatch?.GetInvocationList().Length ?? 0; + + /// Subscribes to the specified full references. + /// The full references to subscribe to. + /// The publishing interval. + /// Cancellation token for the operation. + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => + Task.FromResult(_handle); + + /// Unsubscribes from the specified subscription handle. + /// The subscription handle. + /// Cancellation token for the operation. + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => + Task.CompletedTask; + + /// Raises one production burst. + /// + /// When true, also raises the per-tag for every item — the documented + /// contract a real batch-capable driver honours for legacy consumers. + /// + /// The changed attributes of this burst. + 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)); + } + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs index e1f22d7c..1e1ce406 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs @@ -15,7 +15,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// (when the driver is an ) /// and forwards every transition to its parent as /// — mirroring the OnDataChange → -/// forward. The driver fires +/// forward. The driver fires /// OnAlarmEvent on its OWN thread; the actor marshals it onto the actor thread via Self. /// 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(TimeSpan.FromSeconds(2)) - .FullReference.ShouldBe("tag-z"); + parent.ExpectMsg(TimeSpan.FromSeconds(2)) + .Items[0].FullReference.ShouldBe("tag-z"); } /// @@ -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(TimeSpan.FromSeconds(2)); - dc.FullReference.ShouldBe("tag-a"); + var dc = parent.ExpectMsg(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)); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs index 7cc4e9b3..6f1c4958 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs @@ -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(TimeSpan.FromSeconds(2)); - published.FullReference.ShouldBe("tag-a"); - published.Value.ShouldBe(3.14); - published.StatusCode.ShouldBe(OpcUaStatus.Good); + var published = parent.ExpectMsg(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); } /// @@ -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().StatusCode.ShouldBe(OpcUaStatus.Uncertain); + parent.ExpectMsg().Items[0].StatusCode + .ShouldBe(OpcUaStatus.Uncertain); // Bad — severity bits 10. driver.FireDataChange("tag-1", value: 2, statusCode: 0x80000000u); - parent.ExpectMsg().StatusCode.ShouldBe(OpcUaStatus.Bad); + parent.ExpectMsg().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(); + var noComms = parent.ExpectMsg().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().StatusCode.ShouldBe(0x80740000u); + parent.ExpectMsg().Items[0].StatusCode + .ShouldBe(0x80740000u); } /// @@ -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(TimeSpan.FromSeconds(2)).Value.ShouldBe(7); + parent.ExpectMsg(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"));