feat(runtime): #113 DependencyMuxActor — drivers → virtual-tag fan-out
Some checks failed
v2-ci / build (push) Failing after 36s
v2-ci / unit-tests (tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Security.Tests) (push) Has been skipped
v2-ci / integration (push) Has been skipped
Some checks failed
v2-ci / build (push) Failing after 36s
v2-ci / unit-tests (tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Security.Tests) (push) Has been skipped
v2-ci / integration (push) Has been skipped
End-to-end data path is now wired on the read side: driver subscriptions
fire AttributeValuePublished → DriverHostActor → DependencyMuxActor →
DependencyValueChanged to every interested VirtualTagActor. Previously
the publish hit a dead-letter at the host.
DependencyMuxActor:
- Per-node fan-out router. Maintains tagRef → Set<IActorRef> with a
reverse subscriber → refs index so unregister/replace are O(refs).
- Watches subscribers; Terminated triggers automatic unregister so
dead virtual-tag actors stop receiving publishes.
- Re-register replaces the prior interest set — no stale-ref leaks
on actor restart.
- Drops publishes for refs with no interested subscribers.
VirtualTagActor:
- New Props params: dependencyRefs + mux ActorRef.
- PreStart sends RegisterInterest to the mux; PostStop sends
UnregisterInterest. Default both null so older callers stay quiet.
DriverHostActor:
- New dependencyMux Props param. Steady + Applying states now
receive AttributeValuePublished from their DriverInstance children
and forward to the mux. Null mux is a no-op (dev/Mac).
ServiceCollectionExtensions:
- WithOtOpcUaRuntimeActors spawns DependencyMuxActor before
DriverHostActor and threads its ActorRef into the host's Props.
New DependencyMuxActorKey + DependencyMuxActorName.
Tests: Runtime 57 -> 63 (+6):
- Mux forwards to only subscribers interested in each ref
- Publish for unregistered ref is dropped silently
- Unregister stops forwarding
- Re-register replaces prior interest set
- VirtualTagActor PreStart registration drives end-to-end eval
(uses AwaitAssert to race-safely settle the PreStart Tell)
- DriverHostActor forwards AttributeValuePublished through to mux
All 6 v2 test suites green: 163 tests passing.
F8 (#79) state updated — dep subscribe seam shipped, Core.VirtualTags
production engine binding (compile + ITagUpstreamSource subscribe) is
the residual.
This commit is contained in:
@@ -0,0 +1,155 @@
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
|
||||
|
||||
public sealed class DependencyMuxActorTests : RuntimeActorTestBase
|
||||
{
|
||||
[Fact]
|
||||
public void AttributeValuePublished_is_forwarded_only_to_subscribers_interested_in_that_ref()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var subA = CreateTestProbe();
|
||||
var subB = CreateTestProbe();
|
||||
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1", "tag-2" }, subA.Ref));
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-2", "tag-3" }, subB.Ref));
|
||||
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-1", 10, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-3", 30, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-2", 20, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
|
||||
// subA hears tag-1 + tag-2.
|
||||
var aMsgs = new[]
|
||||
{
|
||||
subA.ExpectMsg<VirtualTagActor.DependencyValueChanged>(),
|
||||
subA.ExpectMsg<VirtualTagActor.DependencyValueChanged>(),
|
||||
}.OrderBy(m => m.TagId).ToList();
|
||||
aMsgs.Select(m => m.TagId).ShouldBe(new[] { "tag-1", "tag-2" });
|
||||
|
||||
// subB hears tag-3 + tag-2.
|
||||
var bMsgs = new[]
|
||||
{
|
||||
subB.ExpectMsg<VirtualTagActor.DependencyValueChanged>(),
|
||||
subB.ExpectMsg<VirtualTagActor.DependencyValueChanged>(),
|
||||
}.OrderBy(m => m.TagId).ToList();
|
||||
bMsgs.Select(m => m.TagId).ShouldBe(new[] { "tag-2", "tag-3" });
|
||||
|
||||
subA.ExpectNoMsg(TimeSpan.FromMilliseconds(100));
|
||||
subB.ExpectNoMsg(TimeSpan.FromMilliseconds(100));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Publish_for_unregistered_ref_is_silently_dropped()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var sub = CreateTestProbe();
|
||||
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref));
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("nobody-cares", 99, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
|
||||
sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void UnregisterInterest_stops_forwarding()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var sub = CreateTestProbe();
|
||||
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref));
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-1", 10, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
sub.ExpectMsg<VirtualTagActor.DependencyValueChanged>();
|
||||
|
||||
mux.Tell(new DependencyMuxActor.UnregisterInterest(sub.Ref));
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-1", 20, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Re_register_replaces_prior_interest_set()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var sub = CreateTestProbe();
|
||||
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-1" }, sub.Ref));
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "tag-2" }, sub.Ref)); // replaces tag-1
|
||||
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-1", 10, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
sub.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("tag-2", 20, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
sub.ExpectMsg<VirtualTagActor.DependencyValueChanged>().TagId.ShouldBe("tag-2");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void VirtualTagActor_PreStart_registers_deps_with_mux_and_eval_fires_end_to_end()
|
||||
{
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var parent = CreateTestProbe();
|
||||
var evaluator = new EchoSumEvaluator();
|
||||
var actor = parent.ChildActorOf(VirtualTagActor.Props(
|
||||
"vt-1", "a+b",
|
||||
evaluator: evaluator,
|
||||
dependencyRefs: new[] { "ref-a", "ref-b" },
|
||||
mux: mux));
|
||||
|
||||
// Race-safe end-to-end check: AwaitAssert retries until the PreStart RegisterInterest
|
||||
// has actually landed at the mux + the publish has fanned out. Until then the publish
|
||||
// gets dropped (no subscriber for "ref-a" yet), so we re-publish each pass.
|
||||
AwaitAssert(() =>
|
||||
{
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished(
|
||||
"ref-a", 10, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
parent.ExpectMsg<VirtualTagActor.EvaluationResult>(TimeSpan.FromMilliseconds(200))
|
||||
.Value.ShouldBe(10);
|
||||
}, duration: TimeSpan.FromSeconds(3));
|
||||
|
||||
// From here the actor is wired — second publish drives the sum.
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("ref-b", 32, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
parent.ExpectMsg<VirtualTagActor.EvaluationResult>(TimeSpan.FromSeconds(2)).Value.ShouldBe(42);
|
||||
|
||||
// Unrelated ref shouldn't fire eval.
|
||||
mux.Tell(new DriverInstanceActor.AttributeValuePublished("ref-unrelated", 99, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DriverHostActor_forwards_AttributeValuePublished_through_to_mux()
|
||||
{
|
||||
// Spin a mux + a stand-in for DriverHostActor that wraps the real DriverHostActor's
|
||||
// forward path. We feed it AttributeValuePublished directly to verify the routing —
|
||||
// exercising the actual DriverHost spawn would require a deployment artifact + DI.
|
||||
var mux = Sys.ActorOf(DependencyMuxActor.Props());
|
||||
var subscriber = CreateTestProbe();
|
||||
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "ref-1" }, subscriber.Ref));
|
||||
|
||||
var hostProbe = CreateTestProbe();
|
||||
var hostActor = Sys.ActorOf(DriverHostActor.Props(
|
||||
dbFactory: NewInMemoryDbFactory(),
|
||||
localNode: ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId.Parse("host-1"),
|
||||
coordinator: hostProbe.Ref,
|
||||
dependencyMux: mux));
|
||||
|
||||
// Tell the host an AttributeValuePublished — it should fan out to the mux + subscriber.
|
||||
hostActor.Tell(new DriverInstanceActor.AttributeValuePublished(
|
||||
"ref-1", 42, OpcUaQuality.Good, DateTime.UtcNow));
|
||||
|
||||
subscriber.ExpectMsg<VirtualTagActor.DependencyValueChanged>().TagId.ShouldBe("ref-1");
|
||||
}
|
||||
|
||||
private sealed class EchoSumEvaluator : ZB.MOM.WW.OtOpcUa.Commons.Engines.IVirtualTagEvaluator
|
||||
{
|
||||
public ZB.MOM.WW.OtOpcUa.Commons.Engines.VirtualTagEvalResult Evaluate(
|
||||
string id, string expression, IReadOnlyDictionary<string, object?> deps)
|
||||
{
|
||||
var sum = deps.Values.OfType<int>().Sum();
|
||||
return ZB.MOM.WW.OtOpcUa.Commons.Engines.VirtualTagEvalResult.Ok(sum);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user