test(mqtt): Sparkplug array metrics in the simulator + live gate (lmxopcua#770)
The simulator now packs Int32/Boolean/Float/Double/String arrays per the Sparkplug B spec (independent of the driver codec) and seeds Profile, Flags and Curve in the standalone catalog. SparkplugLiveTests gains a live test that asserts typed int[]/bool[]/float[] values from the birth, a changed vector (including a different Boolean length) from alias-only DATA, the read path, and BadTypeMismatch for an array metric bound without isArray. Run against a throwaway local Mosquitto; mutation-checked (scalar binding / seed removed). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
@@ -51,6 +51,9 @@ public sealed class SparkplugLiveTests(MqttFixture fixture)
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private const uint BadCommunicationError = 0x80050000u;
|
private const uint BadCommunicationError = 0x80050000u;
|
||||||
|
|
||||||
|
/// <summary>OPC UA <c>BadTypeMismatch</c> — an array metric on a scalar tag (or the reverse).</summary>
|
||||||
|
private const uint BadTypeMismatch = 0x80740000u;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Bound on every wait for broker traffic. Generous enough to absorb a TLS handshake plus a
|
/// Bound on every wait for broker traffic. Generous enough to absorb a TLS handshake plus a
|
||||||
/// rebirth round-trip; short enough that a genuinely dark path fails rather than hangs.
|
/// rebirth round-trip; short enough that a genuinely dark path fails rather than hangs.
|
||||||
@@ -154,6 +157,96 @@ public sealed class SparkplugLiveTests(MqttFixture fixture)
|
|||||||
read[1].Value.ShouldBe(-1234);
|
read[1].Value.ShouldBe(-1234);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// -----------------------------------------------------------------------
|
||||||
|
// array metrics (lmxopcua#770 / driver support #754)
|
||||||
|
// -----------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Sparkplug <c>*Array</c> metrics over a real broker: the edge node packs the vectors itself
|
||||||
|
/// (spec encoding, independent of the driver's codec), the birth carries them, the driver
|
||||||
|
/// publishes typed 1-D arrays under the RawPath, and a changed DATA vector arrives as a change.
|
||||||
|
/// <para>
|
||||||
|
/// Covers Int32Array (a negative element and one wider than 16 bits), BooleanArray (nine
|
||||||
|
/// elements, so the bit-packing crosses a byte boundary) and FloatArray. A shape mismatch
|
||||||
|
/// is also exercised: the same array metric bound WITHOUT <c>isArray</c> must be
|
||||||
|
/// BadTypeMismatch, never a packed buffer served as a scalar.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task Array_metrics_arrive_as_typed_arrays_and_a_changed_vector_is_a_change()
|
||||||
|
{
|
||||||
|
SkipIfUnconfigured();
|
||||||
|
|
||||||
|
var ct = TestContext.Current.CancellationToken;
|
||||||
|
var group = NewGroup("array");
|
||||||
|
const string node = "EdgeA";
|
||||||
|
const string profilePath = "Fixture/Sparkplug/EdgeA/Profile";
|
||||||
|
const string flagsPath = "Fixture/Sparkplug/EdgeA/Flags";
|
||||||
|
const string curvePath = "Fixture/Sparkplug/EdgeA/Curve";
|
||||||
|
const string scalarViewPath = "Fixture/Sparkplug/EdgeA/ProfileAsScalar";
|
||||||
|
|
||||||
|
bool[] flags = [true, false, true, true, false, false, true, false, true];
|
||||||
|
|
||||||
|
await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
|
||||||
|
sim.SetNodeMetrics(
|
||||||
|
new SparkplugMetricSeed("Profile", 1, DataType.Int32Array, new[] { 1, -2, 300000 }),
|
||||||
|
new SparkplugMetricSeed("Flags", 2, DataType.BooleanArray, flags),
|
||||||
|
new SparkplugMetricSeed("Curve", 3, DataType.FloatArray, new[] { 0.5f, 1.25f, -2.5f }));
|
||||||
|
|
||||||
|
await sim.ConnectAsync(ct);
|
||||||
|
await sim.BirthAsync(ct);
|
||||||
|
|
||||||
|
await using var driver = new MqttDriver(
|
||||||
|
SparkplugOptions(
|
||||||
|
group,
|
||||||
|
ArrayTag(profilePath, group, node, "Profile", isArray: true),
|
||||||
|
ArrayTag(flagsPath, group, node, "Flags", isArray: true),
|
||||||
|
ArrayTag(curvePath, group, node, "Curve", isArray: true),
|
||||||
|
ArrayTag(scalarViewPath, group, node, "Profile", isArray: false)),
|
||||||
|
"mqtt-live-sp-array");
|
||||||
|
var recorder = new SparkplugRecorder(driver);
|
||||||
|
|
||||||
|
await driver.InitializeAsync(string.Empty, ct);
|
||||||
|
await driver.SubscribeAsync([profilePath, flagsPath, curvePath, scalarViewPath], TimeSpan.FromSeconds(1), ct);
|
||||||
|
|
||||||
|
var birthProfile = await recorder.WaitAsync(profilePath, v => v.Value is int[], Timeout, ct);
|
||||||
|
birthProfile.Snapshot.StatusCode.ShouldBe(Good);
|
||||||
|
((int[])birthProfile.Snapshot.Value!).ShouldBe([1, -2, 300000]);
|
||||||
|
|
||||||
|
var birthFlags = await recorder.WaitAsync(flagsPath, v => v.Value is bool[], Timeout, ct);
|
||||||
|
birthFlags.Snapshot.StatusCode.ShouldBe(Good);
|
||||||
|
((bool[])birthFlags.Snapshot.Value!).ShouldBe(flags);
|
||||||
|
|
||||||
|
var birthCurve = await recorder.WaitAsync(curvePath, v => v.Value is float[], Timeout, ct);
|
||||||
|
birthCurve.Snapshot.StatusCode.ShouldBe(Good);
|
||||||
|
((float[])birthCurve.Snapshot.Value!).ShouldBe([0.5f, 1.25f, -2.5f]);
|
||||||
|
|
||||||
|
var mismatch = await recorder.WaitAsync(scalarViewPath, v => v.StatusCode != Good, Timeout, ct);
|
||||||
|
mismatch.Snapshot.StatusCode.ShouldBe(BadTypeMismatch);
|
||||||
|
|
||||||
|
// The value-change: alias-only DATA with new vectors (and a different Boolean length).
|
||||||
|
bool[] newFlags = [false, true, true];
|
||||||
|
await sim.PublishNodeDataAsync(
|
||||||
|
[("Profile", new[] { 7, 8, 9, -10 }), ("Flags", newFlags), ("Curve", new[] { 9.75f })],
|
||||||
|
ct);
|
||||||
|
|
||||||
|
var changedProfile = await recorder.WaitAsync(
|
||||||
|
profilePath, v => v.Value is int[] a && a.Length == 4, Timeout, ct);
|
||||||
|
((int[])changedProfile.Snapshot.Value!).ShouldBe([7, 8, 9, -10]);
|
||||||
|
changedProfile.Snapshot.StatusCode.ShouldBe(Good);
|
||||||
|
|
||||||
|
var changedFlags = await recorder.WaitAsync(flagsPath, v => v.Value is bool[] { Length: 3 }, Timeout, ct);
|
||||||
|
((bool[])changedFlags.Snapshot.Value!).ShouldBe(newFlags);
|
||||||
|
|
||||||
|
var changedCurve = await recorder.WaitAsync(curvePath, v => v.Value is float[] { Length: 1 }, Timeout, ct);
|
||||||
|
((float[])changedCurve.Snapshot.Value!).ShouldBe([9.75f]);
|
||||||
|
|
||||||
|
// The read path serves the latest vector.
|
||||||
|
var read = await driver.ReadAsync([profilePath], ct);
|
||||||
|
read[0].StatusCode.ShouldBe(Good);
|
||||||
|
((int[])read[0].Value!).ShouldBe([7, 8, 9, -10]);
|
||||||
|
}
|
||||||
|
|
||||||
// -----------------------------------------------------------------------
|
// -----------------------------------------------------------------------
|
||||||
// rebirth recovery (§3.6 #5)
|
// rebirth recovery (§3.6 #5)
|
||||||
// -----------------------------------------------------------------------
|
// -----------------------------------------------------------------------
|
||||||
@@ -697,6 +790,15 @@ public sealed class SparkplugLiveTests(MqttFixture fixture)
|
|||||||
WriteIdempotent: false);
|
WriteIdempotent: false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>An authored Sparkplug node-level raw tag, optionally declared an array (<c>isArray</c>).</summary>
|
||||||
|
private static RawTagEntry ArrayTag(string rawPath, string groupId, string edgeNodeId, string metricName, bool isArray) =>
|
||||||
|
new(
|
||||||
|
rawPath,
|
||||||
|
$$"""
|
||||||
|
{"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}","metricName":"{{metricName}}"{{(isArray ? ",\"isArray\":true" : "")}}}
|
||||||
|
""",
|
||||||
|
WriteIdempotent: false);
|
||||||
|
|
||||||
/// <summary>The browse configuration blob, parsed through the driver's ONE shared JSON options.</summary>
|
/// <summary>The browse configuration blob, parsed through the driver's ONE shared JSON options.</summary>
|
||||||
private string BrowseConfigJson(string groupId) =>
|
private string BrowseConfigJson(string groupId) =>
|
||||||
JsonSerializer.Serialize(SparkplugOptions(groupId), MqttJson.Options);
|
JsonSerializer.Serialize(SparkplugOptions(groupId), MqttJson.Options);
|
||||||
|
|||||||
+6
-1
@@ -75,7 +75,12 @@ foreach (var node in nodes)
|
|||||||
new SparkplugMetricSeed("Pressure", 2, DataType.Double, 101.325d),
|
new SparkplugMetricSeed("Pressure", 2, DataType.Double, 101.325d),
|
||||||
new SparkplugMetricSeed("Count", 3, DataType.Int32, -42),
|
new SparkplugMetricSeed("Count", 3, DataType.Int32, -42),
|
||||||
new SparkplugMetricSeed("Running", 4, DataType.Boolean, true),
|
new SparkplugMetricSeed("Running", 4, DataType.Boolean, true),
|
||||||
new SparkplugMetricSeed("Serial", 5, DataType.String, $"{node}-001"));
|
new SparkplugMetricSeed("Serial", 5, DataType.String, $"{node}-001"),
|
||||||
|
|
||||||
|
// Array metrics (lmxopcua#770): bind with "isArray": true.
|
||||||
|
new SparkplugMetricSeed("Profile", 6, DataType.Int32Array, new[] { 1, -2, 300000 }),
|
||||||
|
new SparkplugMetricSeed("Flags", 7, DataType.BooleanArray, new[] { true, false, true, true, false, false, true, false, true }),
|
||||||
|
new SparkplugMetricSeed("Curve", 8, DataType.FloatArray, new[] { 0.5f, 1.25f, -2.5f }));
|
||||||
|
|
||||||
edge.SetDeviceMetrics(
|
edge.SetDeviceMetrics(
|
||||||
"Filler1",
|
"Filler1",
|
||||||
|
|||||||
+84
@@ -757,6 +757,17 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable
|
|||||||
metric.BytesValue = ByteString.CopyFrom((byte[])value);
|
metric.BytesValue = ByteString.CopyFrom((byte[])value);
|
||||||
break;
|
break;
|
||||||
|
|
||||||
|
// Array metrics (lmxopcua#770, driver side #754): the whole vector rides packed in bytes_value.
|
||||||
|
// Packed HERE, from the spec text, not through the driver's codec — an encoder that shared
|
||||||
|
// the decoder's assumptions could not falsify them.
|
||||||
|
case DataType.Int32Array:
|
||||||
|
case DataType.BooleanArray:
|
||||||
|
case DataType.FloatArray:
|
||||||
|
case DataType.DoubleArray:
|
||||||
|
case DataType.StringArray:
|
||||||
|
metric.BytesValue = ByteString.CopyFrom(PackArray(dataType, value));
|
||||||
|
break;
|
||||||
|
|
||||||
// Structured values (Gitea #515). The seed carries the already-built protobuf message
|
// Structured values (Gitea #515). The seed carries the already-built protobuf message
|
||||||
// rather than a scalar, because a Template's shape IS its value — there is nothing simpler
|
// rather than a scalar, because a Template's shape IS its value — there is nothing simpler
|
||||||
// to convert from, and inventing a mini-DSL here would put a second, divergent notion of
|
// to convert from, and inventing a mini-DSL here would put a second, divergent notion of
|
||||||
@@ -775,6 +786,74 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Packs an array per Sparkplug B 3.0 "Datatype details": fixed-width elements little-endian, no
|
||||||
|
/// header; BooleanArray = uint32 LE count then bits MSB-first (last byte zero padded);
|
||||||
|
/// StringArray = each UTF-8 string NUL-terminated.
|
||||||
|
/// </summary>
|
||||||
|
private static byte[] PackArray(DataType dataType, object value)
|
||||||
|
{
|
||||||
|
using var ms = new MemoryStream();
|
||||||
|
switch (dataType)
|
||||||
|
{
|
||||||
|
case DataType.Int32Array:
|
||||||
|
foreach (var v in (int[])value)
|
||||||
|
{
|
||||||
|
ms.Write(BitConverter.GetBytes(v));
|
||||||
|
}
|
||||||
|
|
||||||
|
break;
|
||||||
|
case DataType.FloatArray:
|
||||||
|
foreach (var v in (float[])value)
|
||||||
|
{
|
||||||
|
ms.Write(BitConverter.GetBytes(v));
|
||||||
|
}
|
||||||
|
|
||||||
|
break;
|
||||||
|
case DataType.DoubleArray:
|
||||||
|
foreach (var v in (double[])value)
|
||||||
|
{
|
||||||
|
ms.Write(BitConverter.GetBytes(v));
|
||||||
|
}
|
||||||
|
|
||||||
|
break;
|
||||||
|
case DataType.BooleanArray:
|
||||||
|
{
|
||||||
|
var bools = (bool[])value;
|
||||||
|
ms.Write(BitConverter.GetBytes((uint)bools.Length));
|
||||||
|
var bits = new byte[(bools.Length + 7) / 8];
|
||||||
|
for (var i = 0; i < bools.Length; i++)
|
||||||
|
{
|
||||||
|
if (bools[i])
|
||||||
|
{
|
||||||
|
bits[i >> 3] |= (byte)(0x80 >> (i & 7));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
ms.Write(bits);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
case DataType.StringArray:
|
||||||
|
foreach (var v in (string[])value)
|
||||||
|
{
|
||||||
|
ms.Write(System.Text.Encoding.UTF8.GetBytes(v));
|
||||||
|
ms.WriteByte(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
throw new NotSupportedException($"Cannot pack {dataType}.");
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!BitConverter.IsLittleEndian)
|
||||||
|
{
|
||||||
|
throw new PlatformNotSupportedException("The simulator packs arrays with host byte order = little-endian.");
|
||||||
|
}
|
||||||
|
|
||||||
|
return ms.ToArray();
|
||||||
|
}
|
||||||
|
|
||||||
private Payload NewSequencedPayload()
|
private Payload NewSequencedPayload()
|
||||||
{
|
{
|
||||||
var payload = new Payload { Timestamp = NowMs(), Seq = NextSeq() };
|
var payload = new Payload { Timestamp = NowMs(), Seq = NextSeq() };
|
||||||
@@ -837,6 +916,11 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable
|
|||||||
DataType.Boolean => tick % 2 == 0,
|
DataType.Boolean => tick % 2 == 0,
|
||||||
DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}",
|
DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}",
|
||||||
|
|
||||||
|
// Arrays move too (a fixed length, so a tag declaring arrayLength stays valid).
|
||||||
|
DataType.Int32Array => new[] { tick, -tick, tick * 2 },
|
||||||
|
DataType.BooleanArray => new[] { tick % 2 == 0, true, false, tick % 2 != 0, true },
|
||||||
|
DataType.FloatArray => new[] { tick + 0.5f, tick + 1.5f },
|
||||||
|
|
||||||
// Structured values move too, or the live gate could not tell a flattened member that is being
|
// Structured values move too, or the live gate could not tell a flattened member that is being
|
||||||
// fed from one that was merely published once at birth and then went quiet.
|
// fed from one that was merely published once at birth and then went quiet.
|
||||||
DataType.Template => AdvanceTemplate((Payload.Types.Template)seed.Value!, tick),
|
DataType.Template => AdvanceTemplate((Payload.Types.Template)seed.Value!, tick),
|
||||||
|
|||||||
Reference in New Issue
Block a user