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

feat(commons): batched WriteValues on the address-space sink

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:09:27 -04:00
parent 1320f3fa08
commit bb3ccf264b
4 changed files with 155 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs
+6
View File
@@ -26,6 +26,12 @@ public sealed class DeferredAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgical
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _inner.WriteValue(nodeId, value, statusCode, sourceTimestampUtc, realm);
/// <inheritdoc />
// Forward the batch to the inner sink. Without this the interface's default per-item loop runs here and
// the one-lock-per-batch win is silently degraded back to one lock per value on every driver-role host
// (actors inject THIS wrapper, not the inner SdkAddressSpaceSink) — same trap as WireAlarmNotifiers above.
public void WriteValues(IReadOnlyList<ValueWrite> values) => _inner.WriteValues(values);
/// <inheritdoc />
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _inner.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm);
src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs
+20
View File
@@ -22,6 +22,21 @@ public interface IOpcUaAddressSpaceSink
/// calls this once per NodeId (raw + every referencing UNS node) with the matching realm.</param>
void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm);
/// <summary>Write a batch of Variable values. Semantically identical to calling
/// <see cref="WriteValue"/> per item; an SDK-backed sink overrides this to take its internal
/// lock ONCE per batch instead of once per value (the hot-path win). The default implementation
/// loops <see cref="WriteValue"/> so existing sinks and fakes behave unchanged.</summary>
/// <param name="values">The value writes to apply, in order.</param>
void WriteValues(IReadOnlyList<ValueWrite> values)
{
ArgumentNullException.ThrowIfNull(values);
for (var i = 0; i < values.Count; i++)
{
var v = values[i];
WriteValue(v.NodeId, v.Value, v.StatusCode, v.SourceTimestampUtc, v.Realm);
}
}
/// <summary>Write an alarm-condition's full Part 9 state. When a real condition node has been
/// materialised for <paramref name="alarmNodeId"/> via <see cref="MaterialiseAlarmCondition"/>,
/// this projects the whole <see cref="AlarmConditionSnapshot"/>
@@ -165,6 +180,11 @@ public interface IOpcUaAddressSpaceSink
string referenceType = "Organizes");
}
/// <summary>One value write within a <see cref="IOpcUaAddressSpaceSink.WriteValues"/> batch.
/// Same semantics per field as <see cref="IOpcUaAddressSpaceSink.WriteValue"/> (#519: status verbatim).</summary>
public readonly record struct ValueWrite(
string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm);
/// <summary>No-op sink. Bound by default so the actors are safe to run in dev / Mac /
/// integration tests without a real SDK behind them.</summary>
public sealed class NullOpcUaAddressSpaceSink : IOpcUaAddressSpaceSink
tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs
+112
View File
@@ -0,0 +1,112 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Commons.Tests;
/// <summary>
/// The batched <see cref="IOpcUaAddressSpaceSink.WriteValues"/> seam. It exists so an SDK-backed sink
/// can take its internal lock ONCE per batch instead of once per value, and it ships as a <b>default
/// interface method</b> so every existing sink and test fake keeps compiling unchanged — this suite
/// pins both halves of that bargain: the default loops <see cref="IOpcUaAddressSpaceSink.WriteValue"/>
/// verbatim (#519: the status code rides through untouched), and
/// <see cref="DeferredAddressSpaceSink"/> forwards the <i>batch</i> to an overriding inner rather than
/// silently degrading it to per-item calls on every driver-role host.
/// </summary>
public class AddressSpaceSinkBatchTests
{
[Fact]
public void Default_WriteValues_loops_WriteValue()
{
// A sink that implements ONLY the pre-existing interface members — exactly the shape of every
// sink and fake that predates the batch seam. The default implementation must fan the batch out
// to WriteValue, in order, with every field carried through unchanged.
var fake = new PerItemOnlySink();
var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
((IOpcUaAddressSpaceSink)fake).WriteValues(new[]
{
new ValueWrite("Plant/Modbus/dev1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
new ValueWrite("Area/Line/Eq/Speed", 43.5, OpcUaStatus.BadNoCommunication, t0.AddSeconds(1), AddressSpaceRealm.Uns),
new ValueWrite("Plant/Modbus/dev1/Temp", null, OpcUaStatus.Bad, t0.AddSeconds(2), AddressSpaceRealm.Raw),
});
fake.Calls.Count.ShouldBe(3);
fake.Calls[0].ShouldBe(("Plant/Modbus/dev1/Speed", (object?)42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw));
fake.Calls[1].ShouldBe(("Area/Line/Eq/Speed", (object?)43.5, OpcUaStatus.BadNoCommunication, t0.AddSeconds(1), AddressSpaceRealm.Uns));
fake.Calls[2].ShouldBe(("Plant/Modbus/dev1/Temp", (object?)null, OpcUaStatus.Bad, t0.AddSeconds(2), AddressSpaceRealm.Raw));
}
[Fact]
public void Deferred_sink_forwards_WriteValues_to_inner_once()
{
// The whole point of the seam is the ONE-lock-per-batch win, and actors inject the deferred
// wrapper — not the inner SdkAddressSpaceSink. A missing forward here would fall back to the
// interface default and re-explode the batch into per-item WriteValue calls on every
// driver-role host: the batching would ship inert, exactly the F10b / PR#423 trap class.
var inner = new BatchingSpySink();
var sink = new DeferredAddressSpaceSink();
sink.SetSink(inner);
var t0 = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
var batch = new[]
{
new ValueWrite("a", 1, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
new ValueWrite("b", 2, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns),
new ValueWrite("c", 3, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
};
sink.WriteValues(batch);
inner.Batches.Count.ShouldBe(1, "the batch must reach the inner sink as ONE call, not be re-looped");
inner.Batches[0].Count.ShouldBe(3);
inner.Batches[0].Select(v => v.NodeId).ShouldBe(new[] { "a", "b", "c" });
inner.PerItemCalls.ShouldBe(0, "the inner sink's per-item WriteValue must not be reached at all");
}
// ---- test doubles ----
/// <summary>Implements only the members that existed before the batch seam, so the interface's
/// default <c>WriteValues</c> is what runs.</summary>
private sealed class PerItemOnlySink : IOpcUaAddressSpaceSink
{
public List<(string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm)> Calls { get; } = new();
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> Calls.Add((nodeId, value, statusCode, sourceTimestampUtc, realm));
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) { }
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) { }
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
}
/// <summary>Overrides <c>WriteValues</c> the way a real SDK-backed sink does — records the batch and
/// does NOT loop — so a wrapper that dropped the forward is distinguishable from one that kept it.</summary>
private sealed class BatchingSpySink : IOpcUaAddressSpaceSink
{
public List<IReadOnlyList<ValueWrite>> Batches { get; } = new();
public int PerItemCalls { get; private set; }
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> PerItemCalls++;
public void WriteValues(IReadOnlyList<ValueWrite> values) => Batches.Add(values);
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) { }
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) { }
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
}
}
tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredSinkForwardingReflectionTests.cs
+17 -1
View File
@@ -86,12 +86,28 @@ public class DeferredSinkForwardingReflectionTests
{
if (method.Name == nameof(IOpcUaAddressSpaceSink.RebuildAddressSpace)) continue;
method.GetParameters().Any(p => p.ParameterType == typeof(AddressSpaceRealm)).ShouldBeTrue(
method.GetParameters().Any(p => CarriesRealm(p.ParameterType)).ShouldBeTrue(
$"{method.DeclaringType!.Name}.{method.Name} names a node but has no AddressSpaceRealm parameter — " +
"a bare node id is ambiguous across the Raw and UNS namespaces (B4-WP2). Add the realm discriminator.");
}
}
/// <summary>Whether a parameter type supplies the B4-WP2 realm discriminator — either directly, or
/// carried per element by a batch parameter (<c>WriteValues(IReadOnlyList&lt;ValueWrite&gt;)</c> is
/// realm-qualified per item, since one batch legitimately spans both namespaces). Looking THROUGH the
/// element type rather than exempting the method by name keeps the guard mechanical: a future batch
/// item struct that forgets its realm still trips it.</summary>
private static bool CarriesRealm(Type t)
{
if (t == typeof(AddressSpaceRealm)) return true;
if (t.IsGenericType && t.GetGenericArguments().Any(CarriesRealm)) return true;
return t.Namespace == typeof(IOpcUaAddressSpaceSink).Namespace
&& (t.GetProperties().Any(p => p.PropertyType == typeof(AddressSpaceRealm))
|| t.GetFields().Any(f => f.FieldType == typeof(AddressSpaceRealm)));
}
[Fact]
public void Every_IServiceLevelPublisher_method_reaches_the_inner_publisher()
{