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

merge: MQTT browser+contracts review findings resolved (Driver.Mqtt.Browser-002..-005, Driver.Mqtt.Contracts-004..-007)

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-10 08:23:16 -04:00
parent a54661ded8 2dc2075d05
commit acea65eb96
15 changed files with 859 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs
+79 -13
View File
@@ -55,7 +55,27 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
/// </summary>
internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
{
/// <summary>Default cap on recorded nodes; a busy broker's <c>#</c> firehose is unbounded.</summary>
/// <summary>
/// Default cap on recorded nodes; a busy broker's <c>#</c> firehose is unbounded.
/// <para>
/// <b>Read this as a memory bound, not just a node count.</b> Each node retains its full
/// <c>Path</c> (up to <see cref="MaxTopicChars"/>), its <c>Segment</c> (up to
/// <see cref="MaxSegmentChars"/>) and, in plain mode, a snippet (up to
/// <see cref="PayloadSnippetMaxChars"/>), plus its own child dictionary — and every node is
/// held twice, once in its parent's children and once in the flat path index. So a session
/// near these ceilings costs on the order of 100 MB, not the few MB "50 000 nodes" suggests,
/// and a plain-mode open with no <c>TopicPrefix</c> subscribes <c>#</c> — the whole broker —
/// which is exactly the shape that approaches them.
/// </para>
/// <para>
/// <b>The aggregate across sessions is bounded by LIFETIME, not by a count.</b>
/// <c>BrowseSessionRegistry</c> caps neither concurrent sessions nor sessions per user;
/// <c>BrowseSessionReaper</c> evicts and disposes anything untouched for its 2-minute idle
/// TTL on a 30-second tick, and <c>DisposeAsync</c> clears both maps. If a per-session bound
/// is ever needed, cap the number of concurrent sessions there rather than lowering this
/// constant, which would make the picker less useful on a genuinely large broker.
/// </para>
/// </summary>
internal const int DefaultNodeCap = 50_000;
/// <summary>Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet.</summary>
@@ -198,8 +218,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
internal bool Truncated => Volatile.Read(ref _truncated) != 0;
/// <summary>
/// True once a topic was rejected for breaching <see cref="MaxTopicChars"/> /
/// <see cref="MaxSegmentChars"/>; surfaced as its own marker node from
/// True once a topic or metric was rejected as unusable — for breaching
/// <see cref="MaxTopicChars"/> / <see cref="MaxSegmentChars"/>, or for carrying a control
/// character in a segment / identifier; surfaced as its own marker node from
/// <see cref="RootAsync"/>. Kept distinct from <see cref="Truncated"/> because the operator's
/// remedy differs — narrow the prefix, versus this broker publishes absurd topic names.
/// </summary>
@@ -245,9 +266,11 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
{
nodes.Add(new BrowseNode(
OversizedNodeId,
// Both modes word it the same way: since the plain path rejects a control-character
// segment as well as an oversized one, "too long" would name only half the cause.
Mode == MqttMode.SparkplugB
? "⚠ Some metrics were unusable (oversized or malformed names) — use manual entry for those"
: "⚠ Some topics were too long to show — use manual entry for those",
: "⚠ Some topics were unusable (oversized or malformed names) — use manual entry for those",
BrowseNodeKind.Folder,
HasChildrenHint: false));
}
@@ -477,9 +500,21 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
var segments = topic.Split('/');
// Checked as a pre-pass so a rejected topic leaves no half-built path behind.
//
// The control-character half is the same rule IsUsableLabel applies to a Sparkplug identifier,
// and for the same reason: a segment becomes the picker's displayed label, MQTT forbids only
// U+0000 in a topic NAME, and an embedded newline or ANSI escape breaks a single-line row. It is
// a REJECTION rather than a scrub because a scrubbed segment is a different topic, and the
// picker would commit it as the tag's address.
//
// An EMPTY segment is deliberately NOT rejected, so this is not simply IsUsableLabel: `a//b` and
// a trailing '/' are legal MQTT, and such a topic's terminal leaf is a real committable address
// (the commit reads AddressFields[topic], which carries the node's full path). Refusing the whole
// topic would cost the operator a legal address to tidy a blank row. The Sparkplug path has no
// such case — every id it labels came off SparkplugTopic.TryParse, which rejects an empty segment.
foreach (var segment in segments)
{
if (segment.Length > MaxSegmentChars)
if (segment.Length > MaxSegmentChars || HasControlCharacter(segment))
{
Volatile.Write(ref _oversized, 1);
return;
@@ -682,8 +717,14 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
// different nodes, and keying by label would silently drop whichever arrived second.
if (level.TryGetValue(path, out var existing)) return existing;
if (Volatile.Read(ref _nodeCount) >= _nodeCap)
// The slot is RESERVED before the node is created, in one interlocked step, and given back on
// either path that does not consume it. Reading the count and then incrementing it would let N
// observers each see `cap - 1` and all insert, overshooting the cap by up to N-1 — the cap is
// the only bound between a `#` firehose and unbounded memory, so it must not depend on MQTTnet
// happening to deliver on a single dispatcher thread.
if (Interlocked.Increment(ref _nodeCount) > _nodeCap)
{
Interlocked.Decrement(ref _nodeCount);
Volatile.Write(ref _truncated, 1);
return null;
}
@@ -692,9 +733,12 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
var child = level.GetOrAdd(path, created);
if (ReferenceEquals(child, created))
{
Interlocked.Increment(ref _nodeCount);
_byPath[path] = child;
}
else
{
Interlocked.Decrement(ref _nodeCount); // lost the insert race; the reservation was not consumed
}
return child;
}
@@ -709,15 +753,24 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
/// </summary>
/// <param name="label">The candidate label.</param>
/// <returns><see langword="true"/> when the label may be recorded.</returns>
private static bool IsUsableLabel(string label)
private static bool IsUsableLabel(string label) =>
label.Length != 0 && label.Length <= MaxSegmentChars && !HasControlCharacter(label);
/// <summary>
/// Whether <paramref name="text"/> carries a character that would break the picker's single-line
/// rows. Single-sourced so the plain and Sparkplug ingest paths cannot drift on what may become a
/// tree label — see <see cref="ObservePlainTopic"/> for the one rule they deliberately differ on.
/// </summary>
/// <param name="text">The candidate label.</param>
/// <returns><see langword="true"/> when a control character is present.</returns>
private static bool HasControlCharacter(string text)
{
if (label.Length == 0 || label.Length > MaxSegmentChars) return false;
foreach (var c in label)
foreach (var c in text)
{
if (char.IsControl(c)) return false;
if (char.IsControl(c)) return true;
}
return true;
return false;
}
/// <summary>
@@ -1095,10 +1148,23 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
/// <summary>Projects one level of the tree into browse nodes, ordered so the picker is stable.</summary>
/// <param name="level">The child map to project.</param>
/// <returns>The projected nodes, sorted by segment.</returns>
/// <returns>The projected nodes, sorted by segment and then by path.</returns>
/// <remarks>
/// <b>The sort must be TOTAL, which is why the path is the tiebreaker.</b> Sorting by segment
/// alone is only a stable order when labels are unique, and the whole reason this map is keyed by
/// path (see <see cref="GetOrCreate"/>) is that in Sparkplug mode two siblings CAN share one —
/// a device folder <c>Plant1/EdgeA/Temp</c> and an edge-node-level metric
/// <c>Plant1/EdgeA::Temp</c>. <see cref="Enumerable.OrderBy{TSource, TKey}(IEnumerable{TSource},
/// Func{TSource, TKey}, IComparer{TKey})"/> is stable with respect to its input, but the input is
/// <see cref="ConcurrentDictionary{TKey, TValue}.Values"/>, whose enumeration order is not
/// guaranteed and can change as the tree grows — so a tie left to it lets two visually identical
/// rows swap places between successive <see cref="ExpandAsync"/> calls, and the operator cannot
/// see which one they are clicking.
/// </remarks>
private static List<BrowseNode> Project(ConcurrentDictionary<string, TopicNode> level) =>
level.Values
.OrderBy(n => n.Segment, StringComparer.Ordinal)
.ThenBy(n => n.Path, StringComparer.Ordinal)
.Select(n =>
{
// Plain mode infers: a childless node is always the terminal segment of a topic that
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs
+33 -15
View File
@@ -1,4 +1,3 @@
using System.ComponentModel.DataAnnotations;
using System.Text;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -15,14 +14,30 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// <c>OpcUaClientDriverOptions</c>. <see cref="Mode"/> selects the ingest shape;
/// <see cref="Sparkplug"/> and <see cref="Plain"/> are nullable sub-objects — only the one
/// matching <see cref="Mode"/> is populated.
/// <para>
/// ⚠️ <b>There are deliberately no <c>DataAnnotations</c> range attributes on this DTO, and adding
/// any would be documentation that reads like enforcement.</b> The options are bound by
/// <c>JsonSerializer.Deserialize</c> through <see cref="MqttJson.Options"/>, never by
/// <c>IOptions</c> + <c>ValidateDataAnnotations</c>, and nothing in the four MQTT projects calls
/// <c>Validator.TryValidateObject</c> — so the seven <c>[Range]</c>s this type used to carry never
/// ran once. Worse, two of them <i>contradicted</i> the real authority: <c>[Range(1, …)]</c> on
/// <see cref="KeepAliveSeconds"/> forbade the zero that <c>MqttDriver.Normalize</c> deliberately
/// admits as legal MQTT, and the same on <see cref="MaxPayloadBytes"/> forbade the non-positive
/// value this type's own remarks document as "falls back to the 1 MiB default". Wiring them up
/// would therefore have broken deliberate behaviour, so the bounds live where they are actually
/// applied — <c>MqttDriver.Normalize</c>, <c>MqttSubscriptionManager</c>'s constructor and
/// <c>MqttConnection.NextBackoff</c> — and each property below names its own.
/// </para>
/// </remarks>
public sealed record MqttDriverOptions
{
/// <summary>Broker hostname or IP address.</summary>
public string Host { get; init; } = "localhost";
/// <summary>Broker TCP port.</summary>
[Range(1, 65535)]
/// <summary>
/// Broker TCP port. Range-checked by <c>MqttDriverProbe</c> and the AdminUI driver form; there is
/// no <c>[Range]</c> attribute here because nothing would ever run it — see the type remarks.
/// </summary>
public int Port { get; init; } = 8883;
/// <summary>
@@ -67,20 +82,22 @@ public sealed record MqttDriverOptions
/// <summary>Whether to request a clean session (v3.1.1) / clean start (v5.0) at CONNECT.</summary>
public bool CleanSession { get; init; } = true;
/// <summary>Keep-alive interval sent at CONNECT.</summary>
[Range(1, int.MaxValue)]
/// <summary>
/// Keep-alive interval sent at CONNECT. Bounded by <c>MqttDriver.Normalize</c>, which admits
/// <b>zero</b> — legal MQTT for "no keep-alive" — and only replaces a negative value.
/// </summary>
public int KeepAliveSeconds { get; init; } = 30;
/// <summary>Bounded connect deadline (see design §8) — the driver never hangs past this.</summary>
[Range(1, int.MaxValue)]
/// <summary>
/// Bounded connect deadline (see design §8) — the driver never hangs past this. A non-positive
/// value is replaced by <c>MqttDriver.Normalize</c>, which is the enforcement point.
/// </summary>
public int ConnectTimeoutSeconds { get; init; } = 15;
/// <summary>Initial reconnect backoff after a connection drop.</summary>
[Range(1, int.MaxValue)]
/// <summary>Initial reconnect backoff after a connection drop; clamped by <c>MqttConnection.NextBackoff</c>.</summary>
public int ReconnectMinBackoffSeconds { get; init; } = 1;
/// <summary>Cap on the exponential reconnect backoff.</summary>
[Range(1, int.MaxValue)]
/// <summary>Cap on the exponential reconnect backoff; clamped by <c>MqttConnection.NextBackoff</c>.</summary>
public int ReconnectMaxBackoffSeconds { get; init; } = 30;
/// <summary>
@@ -96,7 +113,6 @@ public sealed record MqttDriverOptions
/// because this <c>.Contracts</c> assembly is <i>referenced by</i> the driver assembly and
/// cannot reference it back; the manager's constructor is the single enforcement point.
/// </remarks>
[Range(1, int.MaxValue)]
public int MaxPayloadBytes { get; init; } = 1024 * 1024;
/// <summary>Selects the ingest shape — Plain topics or Sparkplug B.</summary>
@@ -208,7 +224,6 @@ public sealed record MqttSparkplugOptions
/// How long, in seconds, browse/discovery collects NBIRTH/DBIRTH traffic before
/// considering the observed metric set stable.
/// </summary>
[Range(1, int.MaxValue)]
public int BirthObservationWindowSeconds { get; init; } = 15;
}
@@ -225,7 +240,10 @@ public sealed record MqttPlainOptions
/// </summary>
public string TopicPrefix { get; init; } = "";
/// <summary>Default QoS used when a tag's own <c>qos</c> is unset.</summary>
[Range(0, 2)]
/// <summary>
/// Default QoS used when a tag's own <c>qos</c> is unset. MQTT defines exactly three levels;
/// <c>MqttSubscriptionManager</c>'s constructor clamps this to 0–2 before it can reach a
/// SUBSCRIBE, matching what the AdminUI driver form persists.
/// </summary>
public int DefaultQos { get; init; } = 1;
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs
+62 -11
View File
@@ -61,10 +61,11 @@ public static class MqttTagDefinitionFactory
/// and framing is decided solely by <c>JsonDocument.Parse</c> plus the
/// root-<see cref="JsonValueKind.Object"/> check — so leading whitespace is skipped exactly as the JSON
/// spec says, and a padded blob reaches the same verdict its unpadded form does (Gitea #559).
/// Enum fields (<c>payloadFormat</c> / <c>dataType</c>) and <c>qos</c> are read STRICTLY — a
/// present-but-invalid (typo'd) value rejects the whole tag (returns <see langword="false"/> ⇒ the
/// driver surfaces <c>BadNodeIdUnknown</c>) rather than silently defaulting to a wrong-typed Good or
/// a downgraded delivery guarantee. Never throws.
/// Enum fields (<c>payloadFormat</c> / <c>dataType</c>), <c>qos</c> and <c>retainSeed</c> are read
/// STRICTLY — a present-but-invalid (typo'd) value rejects the whole tag (returns
/// <see langword="false"/> ⇒ the driver surfaces <c>BadNodeIdUnknown</c>) rather than silently
/// defaulting to a wrong-typed Good, a downgraded delivery guarantee, or the inverse of the
/// authored seeding behaviour. Never throws.
/// <para>
/// A <b>wildcard</b> topic (<c>+</c> / <c>#</c>) is deliberately NOT rejected here: it is
/// ambiguous rather than unparseable, and <see cref="Inspect"/> is the operator-visible surface
@@ -119,6 +120,12 @@ public static class MqttTagDefinitionFactory
// WEAKER delivery guarantee than the one they authored, with nothing to show for it.
if (!TryReadQosStrict(root, out var qos)) return false;
// …and so is retainSeed, the last behavioural key that was not. A lenient read turned
// "retainSeed":"false" — which a CSV import or a hand-edited blob produces naturally — into
// the default TRUE, the OPPOSITE of what was authored, and the tag then seeded its initial
// value from the broker's retained message the author asked it to ignore.
if (!TryReadBoolStrict(root, "retainSeed", defaultValue: true, out var retainSeed)) return false;
var jsonPath = ReadString(root, "jsonPath");
if (string.IsNullOrEmpty(jsonPath)) jsonPath = RootJsonPath;
@@ -129,7 +136,7 @@ public static class MqttTagDefinitionFactory
JsonPath: jsonPath,
DataType: dataType,
Qos: qos,
RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true),
RetainSeed: retainSeed,
DataTypeAuthored: dataTypeAuthored,
// Optional, and absence is meaningful: no writeTopic ⇒ not writable (Gitea #508).
// Blank normalises to null so "" and absent cannot mean different things.
@@ -230,6 +237,10 @@ public static class MqttTagDefinitionFactory
if (!TryReadQosStrict(root, out var qos)) return false;
// Same key, same strictness — retainSeed is read identically by both shapes, so a blob's
// verdict on it must not depend on which reader took it.
if (!TryReadBoolStrict(root, "retainSeed", defaultValue: true, out var retainSeed)) return false;
def = new MqttTagDefinition(
Name: rawPath,
// Plain-shape fields carried through verbatim; the Sparkplug ingest path reads none of
@@ -239,7 +250,7 @@ public static class MqttTagDefinitionFactory
JsonPath: RootJsonPath,
DataType: dataType,
Qos: qos,
RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true),
RetainSeed: retainSeed,
DataTypeAuthored: dataTypeAuthored,
GroupId: groupId,
EdgeNodeId: edgeNodeId,
@@ -267,7 +278,7 @@ public static class MqttTagDefinitionFactory
/// <item>
/// <b>Rejections</b> — a blob NEITHER runtime mapper would read: an unparseable or
/// non-object document, present-but-invalid <c>payloadFormat</c> / <c>dataType</c> /
/// <c>qos</c> values (all read strictly), or an address that names neither a Plain
/// <c>qos</c> / <c>retainSeed</c> values (all read strictly), or an address that names neither a Plain
/// <c>topic</c> nor a complete Sparkplug <c>groupId</c>/<c>edgeNodeId</c>/<c>metricName</c>
/// tuple. Such a tag is skipped by the driver and reads <c>BadNodeIdUnknown</c> forever.
/// </item>
@@ -446,6 +457,9 @@ public static class MqttTagDefinitionFactory
TagConfigJson.DescribeInvalidEnum<MqttPayloadFormat>(root, "payloadFormat"),
TagConfigJson.DescribeInvalidEnum<DriverDataType>(root, "dataType"),
DescribeInvalidQos(root),
// Read strictly by BOTH mappers, so — unlike payloadFormat — it can never land on an
// accepted blob and is a rejection cause outright.
DescribeInvalidBool(root, "retainSeed"),
})
{
if (w is not null) rejections.Add(w);
@@ -515,8 +529,45 @@ public static class MqttTagDefinitionFactory
private static string? NullIfBlank(string value)
=> string.IsNullOrWhiteSpace(value) ? null : value.Trim();
private static bool ReadBoolOrDefault(JsonElement o, string name, bool defaultValue)
=> o.TryGetProperty(name, out var e) && e.ValueKind is JsonValueKind.True or JsonValueKind.False
? e.GetBoolean()
: defaultValue;
/// <summary>
/// Strict boolean read, mirroring <see cref="TryReadQosStrict"/>'s absent / valid /
/// present-but-invalid split: absent ⇒ <paramref name="defaultValue"/>; a JSON <c>true</c>/
/// <c>false</c> ⇒ that value; anything else present (a string, a number, an explicit
/// <c>null</c>) ⇒ the read FAILS.
/// </summary>
/// <param name="o">The TagConfig root object.</param>
/// <param name="name">The property name to read.</param>
/// <param name="defaultValue">The value an ABSENT key takes.</param>
/// <param name="value">The parsed flag.</param>
/// <returns><see langword="false"/> only when the key is present but not a JSON boolean.</returns>
/// <remarks>
/// Lenient-with-a-default is the wrong shape for a BEHAVIOURAL flag, which is the whole
/// <c>qos</c> argument again: absorbing a malformed value into the default silently hands the
/// operator the opposite of what they authored, with nothing to show for it. It is worse here
/// than for <c>qos</c>, in fact — a defaulted <c>qos</c> is a weaker delivery guarantee, whereas
/// a defaulted <c>retainSeed</c> is the INVERSE of the authored intent whenever the author wrote
/// <c>false</c>, which is the only value worth writing at all.
/// </remarks>
private static bool TryReadBoolStrict(JsonElement o, string name, bool defaultValue, out bool value)
{
value = defaultValue;
if (!o.TryGetProperty(name, out var e)) return true; // absent
if (e.ValueKind is not (JsonValueKind.True or JsonValueKind.False)) return false;
value = e.GetBoolean();
return true;
}
/// <summary>
/// A human-readable warning for a present-but-invalid boolean field, or <see langword="null"/>
/// when it is absent or valid — the boolean counterpart of <see cref="DescribeInvalidQos"/>.
/// </summary>
/// <param name="o">The TagConfig root object.</param>
/// <param name="name">The property name to describe.</param>
/// <returns>The warning text, or <see langword="null"/>.</returns>
private static string? DescribeInvalidBool(JsonElement o, string name)
{
if (!o.TryGetProperty(name, out var e)) return null;
if (e.ValueKind is JsonValueKind.True or JsonValueKind.False) return null;
return $"value '{e.GetRawText()}' for '{name}' is not a JSON boolean; valid: true, false";
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
+21 -3
View File
@@ -304,11 +304,29 @@ public static class MqttWriteEncoder
return MqttWriteEncodeError.None;
case DriverDataType.DateTime:
{
var authored = Convert.ToDateTime(value, CultureInfo.InvariantCulture);
// An Unspecified kind is read as UTC, not passed to ToUniversalTime() — that treats
// it as LOCAL, so the identical authored instant would encode differently on the two
// nodes of a redundant pair, which are not required to share a timezone. Reading it
// as UTC rather than refusing it keeps an ordinary authored "2026-08-10T12:00:00"
// writable: Convert.ToDateTime produces exactly that kind for one.
var utc = authored.Kind == DateTimeKind.Unspecified
? DateTime.SpecifyKind(authored, DateTimeKind.Utc)
: authored.ToUniversalTime();
// Sparkplug's DateTime is an UNSIGNED milliseconds-since-epoch, so a pre-1970 instant
// has no representation at all — the cast below would wrap a negative long into a
// timestamp ~584 million years in the future. Refuse it: unrepresentable values are
// refused, never coerced.
var unixMs = new DateTimeOffset(utc).ToUnixTimeMilliseconds();
if (unixMs < 0) return MqttWriteEncodeError.ValueNotRepresentable;
metric.Datatype = (uint)TahuDataType.DateTime;
metric.LongValue = (ulong)new DateTimeOffset(
Convert.ToDateTime(value, CultureInfo.InvariantCulture).ToUniversalTime())
.ToUnixTimeMilliseconds();
metric.LongValue = (ulong)unixMs;
return MqttWriteEncodeError.None;
}
default:
// DriverDataType.Reference and anything added later: refuse rather than guess a
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugTopic.cs
+49 -1
View File
@@ -195,10 +195,24 @@ public sealed record SparkplugTopic(
/// 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="groupId"/>/<paramref name="edgeNodeId"/> is null/empty, any supplied id is
/// not a legal single topic segment (see <see cref="IsValidSegment"/>), or
/// <paramref name="type"/> is <see cref="SparkplugMessageType.STATE"/> (use
/// <see cref="FormatState"/> instead — STATE has no group/edge-node/device segments).
/// </exception>
/// <remarks>
/// <b>The per-segment rule is enforced here, not left to the caller</b>, so the type is
/// round-trip-closed: everything this emits re-parses through <see cref="TryParse"/> to the
/// instance it was built from. Without it, a <c>+</c>/<c>#</c> id produced a topic MQTT-3.3.2-2
/// forbids in a PUBLISH — a protocol violation a conformant broker answers by dropping the
/// connection — and a <c>deviceId</c> containing <c>/</c> silently added a segment, i.e. a
/// well-formed topic addressing a different device than the caller named.
/// <para>
/// ⚠️ This THROWS where it previously succeeded, so a caller that must not throw keeps its own
/// guard: <c>MqttWriteEncoder.TryEncode</c> returns <c>NoWriteDestination</c> for these ids
/// rather than propagating, and that contract does not change.
/// </para>
/// </remarks>
public static string Format(string groupId, SparkplugMessageType type, string edgeNodeId, string? deviceId = null)
{
ArgumentException.ThrowIfNullOrEmpty(groupId);
@@ -208,15 +222,26 @@ public sealed record SparkplugTopic(
throw new ArgumentException("Use FormatState to build a STATE topic.", nameof(type));
}
ThrowIfNotASegment(groupId, nameof(groupId));
ThrowIfNotASegment(edgeNodeId, nameof(edgeNodeId));
if (deviceId is not null) ThrowIfNotASegment(deviceId, nameof(deviceId));
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>
/// <param name="hostId">The primary-host id; must be a legal single topic segment.</param>
/// <returns>The STATE topic string.</returns>
/// <exception cref="ArgumentException">
/// <paramref name="hostId"/> is null/empty or is not a legal single topic segment — see the
/// round-trip-closure remarks on <see cref="Format"/>.
/// </exception>
public static string FormatState(string hostId)
{
ArgumentException.ThrowIfNullOrEmpty(hostId);
ThrowIfNotASegment(hostId, nameof(hostId));
return $"{Namespace}/STATE/{hostId}";
}
@@ -288,4 +313,27 @@ public sealed record SparkplugTopic(
private static bool IsValidSegment(string segment) =>
segment.Length > 0 && segment.IndexOf('+') < 0 && segment.IndexOf('#') < 0;
/// <summary>
/// Applies <see cref="IsValidSegment"/> on the FORMAT side, so the parser and the formatter judge
/// an id by one rule — plus <c>/</c>, which only the format side can see.
/// </summary>
/// <param name="segment">The candidate segment.</param>
/// <param name="paramName">The argument name to report.</param>
/// <exception cref="ArgumentException">The segment is empty, carries an MQTT wildcard, or carries <c>/</c>.</exception>
/// <remarks>
/// <c>/</c> cannot appear in a parsed id — <see cref="TryParse"/> has already split on it — so
/// <see cref="IsValidSegment"/> has no reason to test for one and does not. On this side it is
/// the difference between an id and two ids: a formatted value carrying one produces a topic with
/// an extra segment, which addresses something other than what the caller named and no longer
/// re-parses at all. Rejecting it is what makes the round trip total.
/// </remarks>
private static void ThrowIfNotASegment(string segment, string paramName)
{
if (IsValidSegment(segment) && segment.IndexOf('/') < 0) return;
throw new ArgumentException(
$"'{segment}' is not a legal Sparkplug topic segment: it must be non-empty and carry no '/' "
+ "and no MQTT wildcard (+ or #).",
paramName);
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
+15 -1
View File
@@ -65,6 +65,14 @@ public sealed class MqttSubscriptionManager
/// </summary>
public const int DefaultMaxPayloadBytes = 1024 * 1024;
/// <summary>
/// Subscription QoS applied when neither the tag nor the driver's <c>Plain</c> section authored
/// one. Mirrors <see cref="MqttPlainOptions.DefaultQos"/>' own default — duplicated rather than
/// referenced because that value is an instance property, and this is the fallback for the case
/// where the whole section is absent.
/// </summary>
public const int DefaultQos = 1;
// OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled).
private const uint StatusGood = 0x00000000u;
@@ -137,7 +145,13 @@ public sealed class MqttSubscriptionManager
MaxPayloadBytes = maxPayloadBytes > 0 ? maxPayloadBytes : DefaultMaxPayloadBytes;
_mode = options.Mode;
_defaultQos = options.Plain?.DefaultQos ?? 1;
// Clamped here for the same reason MaxPayloadBytes is: this constructor is the single
// enforcement point for the driver options the manager captures, and MQTT defines exactly three
// QoS levels — an out-of-range value would otherwise ride into BuildFilters and out onto the
// SUBSCRIBE verbatim. Clamp rather than fall back to the default, matching what
// MqttDriverFormModel.Normalize already persists, so the authoring surface and the driver cannot
// reach different answers for the same authored number.
_defaultQos = Math.Clamp(options.Plain?.DefaultQos ?? DefaultQos, 0, 2);
_driverId = driverId;
_transport = transport;
_logger = logger;