From d7edc709cf646850f4b5b773ef03e620591a7f2c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:44:11 -0400 Subject: [PATCH] feat(runtime): dependency mux accepts batched publishes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DependencyMuxActor now receives DriverInstanceActor.AttributeValueBatchPublished — the primary value ingress since DriverHostActor's batched fan-out — and filters it per item, emitting one DependencyValueChanged per genuinely interested (item, subscriber) pair with that item's own value/timestamp/status sub-code (#519). The per-item AttributeValuePublished receive stays: DriverHostActor's comms-loss fan fabricates it, and per-tag drivers still take that path. Indexed for-loop over Items to avoid an interface-enumerator allocation per burst. --- .../VirtualTags/DependencyMuxActor.cs | 28 +++++++++++++++-- .../VirtualTags/DependencyMuxActorTests.cs | 30 +++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs index 2820df66..69fb5bb4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs @@ -12,9 +12,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// forwards a to each interested /// subscriber. /// -/// DriverHostActor forwards every AttributeValuePublished it receives from its -/// DriverInstanceActor children to this mux (one mux per driver-role node). The mux is -/// deliberately not a DPS subscriber — virtual-tag evaluation is local to each node and +/// DriverHostActor forwards driver-child value publishes to this mux (one mux per driver-role +/// node). The PRIMARY ingress is the batched +/// — one envelope per driver +/// production burst, which the mux filters per item — while the per-value +/// receive stays for the comms-loss fan +/// (DriverHostActor.FanCommsLossToDependencyMux fabricates it) and per-tag drivers. Both +/// produce the same per interested subscriber. +/// The mux is deliberately not a DPS subscriber — virtual-tag evaluation is local to each node and /// would over-emit if it spanned the cluster. /// public sealed class DependencyMuxActor : ReceiveActor @@ -43,6 +48,7 @@ public sealed class DependencyMuxActor : ReceiveActor Receive(OnRegister); Receive(msg => RemoveSubscriber(msg.Subscriber)); Receive(OnAttributeValuePublished); + Receive(OnAttributeValueBatchPublished); Receive(msg => RemoveSubscriber(msg.ActorRef)); } @@ -107,4 +113,20 @@ public sealed class DependencyMuxActor : ReceiveActor } } + /// Filters one batched driver production burst per item — the batched analogue of + /// . Items nobody registered for are dropped (the common case: + /// only a fraction of an address space feeds virtual-tag expressions), so a burst of N values costs one + /// message plus one DependencyValueChanged per genuinely interested (item, subscriber) pair. + private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg) + { + // Indexed for-loop, not foreach: this is the hot value ingress and an IReadOnlyList enumerator + // would allocate once per burst. + for (var i = 0; i < msg.Items.Count; i++) + { + var item = msg.Items[i]; + if (!_byRef.TryGetValue(item.FullReference, out var subscribers) || subscribers.Count == 0) continue; + var dep = new VirtualTagActor.DependencyValueChanged(item.FullReference, item.Value, item.TimestampUtc, item.StatusCode); + foreach (var sub in subscribers) sub.Tell(dep); + } + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs index c5960c68..fccaebc4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs @@ -149,6 +149,36 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase subscriber.ExpectMsg().TagId.ShouldBe("ref-1"); } + /// A batched driver burst (the primary ingress from DriverHostActor since the value-pipeline + /// batching work) is filtered per item: a subscriber hears exactly one DependencyValueChanged for the + /// ref it registered, carrying that item's own value/timestamp/quality, and nothing for the rest of the + /// burst. + [Fact] + public void AttributeValueBatchPublished_is_filtered_per_item_to_interested_subscribers() + { + var mux = Sys.ActorOf(DependencyMuxActor.Props()); + var sub = CreateTestProbe(); + mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-x" }, sub.Ref)); + + var tsX = new DateTime(2026, 8, 15, 9, 0, 0, DateTimeKind.Utc); + var tsY = new DateTime(2026, 8, 15, 9, 0, 1, DateTimeKind.Utc); + mux.Tell(new DriverInstanceActor.AttributeValueBatchPublished("driver-1", new[] + { + new DriverInstanceActor.AttributeValueItem("tag-y", 99, OpcUaStatus.Good, tsY), + new DriverInstanceActor.AttributeValueItem("tag-x", 10, OpcUaStatus.BadNoCommunication, tsX), + })); + + var dep = sub.ExpectMsg(TimeSpan.FromSeconds(3)); + dep.TagId.ShouldBe("tag-x"); + dep.Value.ShouldBe(10); + dep.TimestampUtc.ShouldBe(tsX); + // #519: the item's own status sub-code rides through, not a Good/Bad bucket. + dep.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication); + + // Nothing for tag-y — no subscriber registered for it. + sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + /// #478 — the mux forwards the published source quality onto DependencyValueChanged so the /// scripted-alarm host can derive a condition's worst-of-input quality. [Fact]