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

feat(mqtt): Sparkplug topic parse/format + datatype map

Browse Source 
SparkplugTopic (Driver.Mqtt/Sparkplug/) parses/formats spBv1.0 topics for
every message type (NBIRTH/DBIRTH/NDATA/DDATA/NDEATH/DDEATH/NCMD/DCMD/STATE).
TryParse never throws -- it is fed every topic a live spBv1.0/{group}/#
subscription delivers -- and rejects device/node-scope mismatches and MQTT
wildcard chars in a segment. STATE is handled honestly rather than force-fit
into the {group}/{type}/{node} mould: it parses both the v3.0
spBv1.0/STATE/{hostId} form and the legacy no-prefix STATE/{hostId} form
(tolerated on receive only -- Format/FormatState always emit v3.0). Format
gives Task 20's NCMD/DCMD write path a builder instead of hand-concatenation.

SparkplugDataType is a `global using` alias for the vendored proto's
generated Org.Eclipse.Tahu.Protobuf.DataType, not a second hand-duplicated
enum -- Metric.Datatype is a raw wire uint32 (no enum-typed field forces a
second CLR type to exist), and SparkplugCodec (Task 16, landed concurrently)
already casts straight to the generated type. A duplicate enum would be the
same enum-drift hazard this repo already names systemic (CLAUDE.md's driver
enum-serialization bug) and would force every downstream task to cast
between two value-compatible-but-nominally-different enums. ToDriverDataType()
maps per design doc SS3.5: Int8/UInt8 widen to Int16/UInt16 (no 8-bit
DriverDataType member), Float/Double to Float32/Float64 (there is no
DriverDataType.Double), Text/UUID/Bytes/File to String, *Array variants to
their element type (IsSparkplugArray carries the array bit separately), and
DataSet/Template/PropertySet/PropertySetList/Unknown return null -- an
explicit "unsupported, caller must skip+warn" rather than a guessed String.

The completeness test enumerates the live generated DataType member set
(via the alias) and asserts every member is mapped or on the explicit
unsupported list, so a future Tahu proto change is caught automatically
instead of silently falling through a stale duplicate enum's default.
Falsifiability verified by hand for three defect shapes (each reverted after
observing RED): wrong Int8 widening, a dropped mapping falling through
undetected, and inverted node/device topic scoping.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 21:16:02 -04:00
parent c164ec3da1
commit b245f380b1
4 changed files with 725 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugTopic.cs
+291
View File
@@ -0,0 +1,291 @@
using System.Diagnostics.CodeAnalysis;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
/// <summary>
/// The Sparkplug B topic-namespace element that identifies a message's purpose — the third
/// segment of <c>spBv1.0/{group}/{type}/{node}[/{device}]</c>, or the second segment of the
/// differently-shaped <c>spBv1.0/STATE/{hostId}</c>. See design doc §3.1/§3.6 and Sparkplug B
/// v3.0 spec §6 (Topic Namespace Elements).
/// </summary>
public enum SparkplugMessageType
{
/// <summary>Node birth certificate — (re)publishes an edge node's full metric/alias set.</summary>
NBIRTH,
/// <summary>Device birth certificate — (re)publishes a device's full metric/alias set.</summary>
DBIRTH,
/// <summary>Node data — incremental metric updates owned by the edge node itself.</summary>
NDATA,
/// <summary>Device data — incremental metric updates for a device under the edge node.</summary>
DDATA,
/// <summary>Node death certificate (the edge node's MQTT Will) — the node has gone offline.</summary>
NDEATH,
/// <summary>Device death certificate — the device has gone offline (published by its edge node).</summary>
DDEATH,
/// <summary>Node command — a write/rebirth request addressed to the edge node.</summary>
NCMD,
/// <summary>Device command — a write request addressed to a device under the edge node.</summary>
DCMD,
/// <summary>
/// Primary-host online/offline state. Shaped differently from every other message type — see
/// the remarks on <see cref="SparkplugTopic"/> and <see cref="SparkplugTopic.HostId"/>.
/// </summary>
STATE,
}
/// <summary>
/// A parsed Sparkplug B MQTT topic. See design doc §3.1/§3.6 and Sparkplug B v3.0 spec §6.
/// </summary>
/// <param name="Type">The message's purpose.</param>
/// <param name="GroupId">
/// The Sparkplug group id. <see langword="null"/> for <see cref="SparkplugMessageType.STATE"/>,
/// always populated otherwise.
/// </param>
/// <param name="EdgeNodeId">
/// The Sparkplug edge-node id. <see langword="null"/> for <see cref="SparkplugMessageType.STATE"/>,
/// always populated otherwise.
/// </param>
/// <param name="DeviceId">
/// The Sparkplug device id, present only for the device-scoped message types
/// (<see cref="SparkplugMessageType.DBIRTH"/>/<see cref="SparkplugMessageType.DDATA"/>/
/// <see cref="SparkplugMessageType.DDEATH"/>/<see cref="SparkplugMessageType.DCMD"/>);
/// <see langword="null"/> for node-scoped messages and for <see cref="SparkplugMessageType.STATE"/>.
/// </param>
/// <param name="HostId">
/// The primary-host id, populated only for <see cref="SparkplugMessageType.STATE"/>; carries the
/// third topic segment of <c>spBv1.0/STATE/{hostId}</c> (or the second segment of the legacy
/// pre-3.0 <c>STATE/{hostId}</c> form). <see langword="null"/> for every other message type.
/// </param>
/// <remarks>
/// <para>
/// <b>Never throws on arbitrary input.</b> <see cref="TryParse"/> is the primitive everything
/// else is built on — it is fed every topic string the broker delivers under a
/// <c>spBv1.0/{groupId}/#</c> subscription, none of it validated ahead of time, so it returns
/// <see langword="false"/> for anything malformed rather than throwing. <see cref="Parse"/> is
/// a throwing convenience wrapper for call sites (tests, hand-built topics) that already know
/// the string is well-formed.
/// </para>
/// <para>
/// <b>STATE does not fit the <c>{group}/{type}/{node}</c> mould — handled honestly, not
/// force-fit.</b> The Sparkplug v3.0 spec shapes the primary-host state topic as
/// <c>spBv1.0/STATE/{hostId}</c>: the message-type segment sits where a group id would
/// otherwise be, and there is no edge-node or device segment at all. Pre-3.0 peers additionally
/// published a bare <c>STATE/{hostId}</c> with no <c>spBv1.0</c> namespace prefix. This parser
/// targets the v3.0 form and tolerates the legacy one on receive (design §3.1); <see cref="Format"/>
/// and <see cref="FormatState"/> only ever produce the v3.0 form. A parsed STATE topic carries
/// its host id in <see cref="HostId"/> and leaves <see cref="GroupId"/>/<see cref="EdgeNodeId"/>/
/// <see cref="DeviceId"/> <see langword="null"/> — every other message type is the mirror image
/// (<see cref="GroupId"/>/<see cref="EdgeNodeId"/> populated, <see cref="HostId"/> null).
/// </para>
/// <para>
/// Group/edge-node/device/host segments are treated as opaque identifiers: validated only for
/// non-emptiness and the absence of the MQTT wildcard characters <c>+</c>/<c>#</c> (a broker
/// never delivers a PUBLISH on a topic containing either, so a topic string that does is
/// malformed, not a legitimate id worth preserving). No further character-set restriction is
/// applied — Sparkplug does not constrain id charsets beyond "not a topic-level separator".
/// </para>
/// </remarks>
public sealed record SparkplugTopic(
SparkplugMessageType Type,
string? GroupId,
string? EdgeNodeId,
string? DeviceId,
string? HostId)
{
private const string Namespace = "spBv1.0";
/// <summary>
/// Attempts to parse <paramref name="topic"/> as a Sparkplug B topic. Never throws — returns
/// <see langword="false"/> (and a <see langword="null"/> <paramref name="result"/>) for
/// anything that is not a well-formed Sparkplug topic, including <see langword="null"/>/empty
/// input, wrong namespace, an unrecognised message-type segment, a device-scoped message
/// missing its device segment (or vice versa), or a segment containing an MQTT wildcard.
/// </summary>
public static bool TryParse(string? topic, [NotNullWhen(true)] out SparkplugTopic? result)
{
result = null;
if (string.IsNullOrEmpty(topic))
{
return false;
}
var segments = topic.Split('/');
// Legacy pre-3.0 STATE form: `STATE/{hostId}`, no `spBv1.0` namespace prefix.
if (segments.Length == 2 && segments[0] == "STATE")
{
return TryBuildState(segments[1], out result);
}
if (segments.Length < 2 || segments[0] != Namespace)
{
return false;
}
// v3.0 STATE form: `spBv1.0/STATE/{hostId}`.
if (segments[1] == "STATE")
{
return segments.Length == 3 && TryBuildState(segments[2], out result);
}
// Every other message type: `spBv1.0/{group}/{type}/{node}[/{device}]`.
if (segments.Length is not (4 or 5))
{
return false;
}
if (!TryParseMessageType(segments[2], out var type))
{
return false;
}
var groupId = segments[1];
var edgeNodeId = segments[3];
if (!IsValidSegment(groupId) || !IsValidSegment(edgeNodeId))
{
return false;
}
var expectsDevice = IsDeviceScoped(type);
var hasDeviceSegment = segments.Length == 5;
if (expectsDevice != hasDeviceSegment)
{
return false;
}
string? deviceId = null;
if (hasDeviceSegment)
{
deviceId = segments[4];
if (!IsValidSegment(deviceId))
{
return false;
}
}
result = new SparkplugTopic(type, groupId, edgeNodeId, deviceId, null);
return true;
}
/// <summary>
/// Parses <paramref name="topic"/>, throwing <see cref="FormatException"/> if it is not a
/// well-formed Sparkplug B topic. See <see cref="TryParse"/> for the non-throwing form.
/// </summary>
public static SparkplugTopic Parse(string? topic)
{
if (TryParse(topic, out var result))
{
return result;
}
throw new FormatException($"'{topic}' is not a valid Sparkplug B topic.");
}
/// <summary>
/// Builds a Sparkplug B topic string for a non-STATE message — e.g.
/// <c>Format("Plant1", SparkplugMessageType.NCMD, "EdgeA")</c> for the Task 20 write path, so
/// it does not have to hand-concatenate segments itself.
/// </summary>
/// <exception cref="ArgumentException">
/// <paramref name="groupId"/>/<paramref name="edgeNodeId"/> is null/empty, or
/// <paramref name="type"/> is <see cref="SparkplugMessageType.STATE"/> (use
/// <see cref="FormatState"/> instead — STATE has no group/edge-node/device segments).
/// </exception>
public static string Format(string groupId, SparkplugMessageType type, string edgeNodeId, string? deviceId = null)
{
ArgumentException.ThrowIfNullOrEmpty(groupId);
ArgumentException.ThrowIfNullOrEmpty(edgeNodeId);
if (type == SparkplugMessageType.STATE)
{
throw new ArgumentException("Use FormatState to build a STATE topic.", nameof(type));
}
return deviceId is null
? $"{Namespace}/{groupId}/{type}/{edgeNodeId}"
: $"{Namespace}/{groupId}/{type}/{edgeNodeId}/{deviceId}";
}
/// <summary>Builds the v3.0 STATE topic string <c>spBv1.0/STATE/{hostId}</c>.</summary>
public static string FormatState(string hostId)
{
ArgumentException.ThrowIfNullOrEmpty(hostId);
return $"{Namespace}/STATE/{hostId}";
}
/// <summary>
/// Formats this instance back to its topic string (always the v3.0 STATE form for STATE
/// topics, even if this instance was parsed from the legacy no-prefix form).
/// </summary>
/// <exception cref="InvalidOperationException">
/// A required field for this instance's <see cref="Type"/> is <see langword="null"/> — i.e.
/// this instance was not built by <see cref="TryParse"/>/<see cref="Parse"/> and violates the
/// invariants documented on the type.
/// </exception>
public string ToTopicString() => Type == SparkplugMessageType.STATE
? FormatState(HostId ?? throw new InvalidOperationException("STATE topic is missing HostId."))
: Format(
GroupId ?? throw new InvalidOperationException("Non-STATE topic is missing GroupId."),
Type,
EdgeNodeId ?? throw new InvalidOperationException("Non-STATE topic is missing EdgeNodeId."),
DeviceId);
private static bool TryBuildState(string hostId, out SparkplugTopic? result)
{
result = null;
if (!IsValidSegment(hostId))
{
return false;
}
result = new SparkplugTopic(SparkplugMessageType.STATE, null, null, null, hostId);
return true;
}
private static bool TryParseMessageType(string segment, out SparkplugMessageType type)
{
switch (segment)
{
case nameof(SparkplugMessageType.NBIRTH):
type = SparkplugMessageType.NBIRTH;
return true;
case nameof(SparkplugMessageType.DBIRTH):
type = SparkplugMessageType.DBIRTH;
return true;
case nameof(SparkplugMessageType.NDATA):
type = SparkplugMessageType.NDATA;
return true;
case nameof(SparkplugMessageType.DDATA):
type = SparkplugMessageType.DDATA;
return true;
case nameof(SparkplugMessageType.NDEATH):
type = SparkplugMessageType.NDEATH;
return true;
case nameof(SparkplugMessageType.DDEATH):
type = SparkplugMessageType.DDEATH;
return true;
case nameof(SparkplugMessageType.NCMD):
type = SparkplugMessageType.NCMD;
return true;
case nameof(SparkplugMessageType.DCMD):
type = SparkplugMessageType.DCMD;
return true;
default:
type = default;
return false;
}
}
private static bool IsDeviceScoped(SparkplugMessageType type) => type is
SparkplugMessageType.DBIRTH or SparkplugMessageType.DDATA or SparkplugMessageType.DDEATH or SparkplugMessageType.DCMD;
private static bool IsValidSegment(string segment) =>
segment.Length > 0 && segment.IndexOf('+') < 0 && segment.IndexOf('#') < 0;
}