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

feat(runtime): OpcUaPublishActor batched value updates

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:16:22 -04:00
parent ba55a5b48e
commit 04e6a4690c
2 changed files with 173 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+24
View File
@@ -51,6 +51,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// hands it to the SDK verbatim.</para></summary>
public sealed record AttributeValueUpdate(string NodeId, object? Value, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
/// <summary>Batched form of <see cref="AttributeValueUpdate"/>: every fan-out write of one driver
/// production burst in one message — the sink takes its lock once for the whole batch. Items carry
/// their realm each (<see cref="ValueWrite"/>); status codes ride verbatim (#519).</summary>
public sealed record AttributeValueBatchUpdate(IReadOnlyList<ValueWrite> Items);
/// <summary>Carries the full Part 9 condition state for a scripted alarm to the sink. The
/// <paramref name="State"/> snapshot is the Commons projection the Runtime host maps from the engine's
/// Core <c>AlarmConditionState</c> + severity/message — the actor stays decoupled from
@@ -286,6 +291,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
_healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);
Receive<AttributeValueUpdate>(HandleAttributeUpdate);
Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);
Receive<AlarmStateUpdate>(HandleAlarmUpdate);
Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
Receive<RebuildAddressSpace>(HandleRebuild);
@@ -330,6 +336,24 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
/// <summary>Mirrors <see cref="HandleAttributeUpdate"/> for a whole burst: ONE
/// <see cref="IOpcUaAddressSpaceSink.WriteValues"/> call, but the write tally and the
/// <c>otopcua.opcua.sink.write</c> meter still advance PER ITEM — batching is a transport change, so
/// dashboards must not drop by the batch factor.</summary>
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
{
try
{
_sink.WriteValues(msg.Items);
Interlocked.Add(ref _writes, msg.Items.Count);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "value"));
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count);
}
}
private void HandleAlarmUpdate(AlarmStateUpdate msg)
{
try
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs
+149
View File
@@ -0,0 +1,149 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
/// <summary>
/// Task 7 of the value-pipeline batching work: <see cref="OpcUaPublishActor.AttributeValueBatchUpdate"/>
/// must reach the sink as exactly ONE <see cref="IOpcUaAddressSpaceSink.WriteValues"/> call — that single
/// call is the whole point (the SDK-backed sink takes its lock once per batch instead of once per value),
/// so a handler that looped <c>WriteValue</c> would pass a "the values arrived" assertion while delivering
/// none of the win. Hence the fake counts per-item <c>WriteValue</c> calls separately and the assertions
/// require that count to be zero.
/// <para>The per-item <see cref="OpcUaPublishActor.AttributeValueUpdate"/> receive stays — VirtualTagHostActor
/// and the alarm paths still send it — so it is pinned here too.</para>
/// </summary>
public sealed class OpcUaPublishActorBatchTests : RuntimeActorTestBase
{
/// <summary>A 3-item batch lands as one WriteValues call carrying all three items verbatim (field-exact,
/// realm included), and never decomposes into per-item WriteValue calls.</summary>
[Fact]
public void Batch_message_reaches_sink_as_one_WriteValues_call()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
var items = new[]
{
new ValueWrite("Plant/Modbus/dev1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
new ValueWrite("AreaA/Line1/Eq1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns),
new ValueWrite("Plant/Modbus/dev1/Temp", 7.5d, OpcUaStatus.BadNoCommunication,
t0.AddSeconds(1), AddressSpaceRealm.Raw),
};
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(items));
AwaitAssert(() => sink.Batches.Count.ShouldBe(1), PresenceBudget);
var batch = sink.Batches[0];
batch.Count.ShouldBe(3);
batch[0].ShouldBe(items[0]);
batch[1].ShouldBe(items[1]);
batch[2].ShouldBe(items[2]);
// The load-bearing half: one batched call, NOT three per-item calls.
sink.SingleWrites.Count.ShouldBe(0);
}
/// <summary>A sink that throws out of WriteValues is swallowed and logged — the actor survives and keeps
/// processing (proved by a follow-up message landing), rather than restarting and dropping traffic.</summary>
[Fact]
public void Sink_throw_is_swallowed_and_logged()
{
var sink = new ThrowOnWriteValuesSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(new[]
{
new ValueWrite("Plant/Modbus/dev1/Speed", 1, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
}));
// Follow-up on the untouched per-item path: it lands only if the actor is still alive and pumping.
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate(
"Plant/Modbus/dev1/Temp", 2, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw));
AwaitAssert(() => sink.SingleWrites.Count.ShouldBe(1), PresenceBudget);
sink.Threw.ShouldBeTrue();
}
/// <summary>The pre-existing single-item AttributeValueUpdate path is untouched by the batch addition.</summary>
[Fact]
public void Single_item_message_still_works()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate(
"AreaA/Line1/Eq1/Speed", 99, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.SingleWrites.Count.ShouldBe(1), PresenceBudget);
sink.SingleWrites[0].ShouldBe(new ValueWrite("AreaA/Line1/Eq1/Speed", 99, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns));
sink.Batches.Count.ShouldBe(0);
}
private void AwaitAssert(Action assertion, TimeSpan duration)
{
var deadline = DateTime.UtcNow + duration;
Exception? last = null;
while (DateTime.UtcNow < deadline)
{
try { assertion(); return; }
catch (Exception ex) { last = ex; Thread.Sleep(25); }
}
if (last is not null) throw last;
}
/// <summary>Records batched and per-item value writes separately, so a handler that looped WriteValue
/// instead of calling WriteValues once is distinguishable from one that batched.</summary>
private class RecordingSink : IOpcUaAddressSpaceSink
{
private readonly object _gate = new();
private readonly List<IReadOnlyList<ValueWrite>> _batches = new();
private readonly List<ValueWrite> _singles = new();
public IReadOnlyList<IReadOnlyList<ValueWrite>> Batches { get { lock (_gate) return _batches.ToArray(); } }
public IReadOnlyList<ValueWrite> SingleWrites { get { lock (_gate) return _singles.ToArray(); } }
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns)
{
lock (_gate) _singles.Add(new ValueWrite(nodeId, value, statusCode, sourceTimestampUtc, realm));
}
// Overrides the interface's default WriteValues so the batch is observed as one call, not N.
public virtual void WriteValues(IReadOnlyList<ValueWrite> values)
{
lock (_gate) _batches.Add(values.ToArray());
}
public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
public void RebuildAddressSpace() { }
public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
}
/// <summary>A sink whose batched write throws — drives the handler's catch.</summary>
private sealed class ThrowOnWriteValuesSink : RecordingSink
{
private int _threw;
public bool Threw => Volatile.Read(ref _threw) != 0;
public override void WriteValues(IReadOnlyList<ValueWrite> values)
{
Interlocked.Exchange(ref _threw, 1);
throw new InvalidOperationException("simulated batched-write fault");
}
}
}