feat(runtime): dependency mux accepts batched publishes
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.
This commit is contained in:
@@ -12,9 +12,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
/// forwards a <see cref="VirtualTagActor.DependencyValueChanged"/> to each interested
|
||||
/// subscriber.
|
||||
///
|
||||
/// DriverHostActor forwards every <c>AttributeValuePublished</c> 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
|
||||
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> — one envelope per driver
|
||||
/// production burst, which the mux filters per item — while the per-value
|
||||
/// <see cref="DriverInstanceActor.AttributeValuePublished"/> receive stays for the comms-loss fan
|
||||
/// (<c>DriverHostActor.FanCommsLossToDependencyMux</c> fabricates it) and per-tag drivers. Both
|
||||
/// produce the same <see cref="VirtualTagActor.DependencyValueChanged"/> 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.
|
||||
/// </summary>
|
||||
public sealed class DependencyMuxActor : ReceiveActor
|
||||
@@ -43,6 +48,7 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
Receive<RegisterInterest>(OnRegister);
|
||||
Receive<UnregisterInterest>(msg => RemoveSubscriber(msg.Subscriber));
|
||||
Receive<DriverInstanceActor.AttributeValuePublished>(OnAttributeValuePublished);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(OnAttributeValueBatchPublished);
|
||||
Receive<Terminated>(msg => RemoveSubscriber(msg.ActorRef));
|
||||
}
|
||||
|
||||
@@ -107,4 +113,20 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Filters one batched driver production burst per item — the batched analogue of
|
||||
/// <see cref="OnAttributeValuePublished"/>. 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.</summary>
|
||||
private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
// Indexed for-loop, not foreach: this is the hot value ingress and an IReadOnlyList<T> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user