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

fix(mqtt): close #509, #510, #512, #513, #514

Browse Source 
Five MQTT tracker items — four AdminUI/authoring defects plus the .Browser
layering exception. No runtime driver behaviour changes.

#513 payloadFormat in a Sparkplug blob. MqttTagConfigModel.ToJson wrote the key
unconditionally. Sharper than the issue states: PayloadFormat is the one Plain
field with a NON-NULLABLE typed backing, so the write did not preserve a retyped
tag's leftover — it INVENTED "payloadFormat":"Json" in a Sparkplug tag that never
had one. Now Plain-only; preserve-unknown-keys is pinned unchanged.

#509 slashed Sparkplug metric names. The mapper already carried metricName
verbatim; only the tag Name (a RawPath segment) tripped RawPaths.ValidateSegment,
which refused the batch whole — including Node Control/Rebirth, mandatory on
every Sparkplug edge node. ToRawSegment coerces the Name and leaves the address
untouched. Coercion is not injective, so DescribeNameCollisions refuses a
colliding set BY NAME: silently merging two metrics into one tag would be worse
than the refusal it replaces. Applies to every driver, not just MQTT — the
argument is about what a Name is, not about who produced it. The sanitiser is
pinned against RawPaths.ValidateSegment rather than a copied charset.

#510 stale rebirth affordance. _canRebirth was probed once at browse-open and
sessions are reaped for idleness, so a modal left open offered an action whose
token no longer resolved. Re-probed on arm AND on confirm (the reap can land
between the two, and confirm is the one path that publishes to a live broker);
the affordance retires with an explanation instead of vanishing or erroring.

#514 rebirth on a quiet plant. The tree is birth-driven and births are never
retained, so a plant that birthed before the window opened showed an empty tree —
and the scope came only from a tree selection, so the one action that would fill
the tree could not be armed. The session already accepted a bare {group}/{edgeNode}
("the prime rebirth target"); this adds the affordance. Edge-node scope only, not
a group broadcast. The rule is extracted to SparkplugRebirthScope so it is
testable — this repo has no bUnit — and is advisory: the session re-resolves and
refuses before publishing.

#512 .Browser -> .Driver layering. TWO extractions were needed, not the one the
issue anticipated: it costed this as "the client-options builder", but the browse
session decodes births itself, so moving only that would have left the reference
exactly where it was.
  - MqttClientOptionsFactory (TLS, PEM CA-pin validator, credentials, protocol
    mapping) -> new leaf .Transport project. NOT .Contracts: that is deliberately
    transport-free and MqttClientOptions is an MQTTnet type, so .Contracts would
    hand MQTTnet to ControlPlane. This is the fix the .Browser csproj comment had
    already prescribed, against the issue's suggestion.
  - The Sparkplug protocol layer (SparkplugTopic/MessageType, SparkplugCodec,
    RebirthRequester) -> .Contracts, beside the Tahu wire schema it encodes.
Still exactly one copy of the TLS/CA-pin logic; MqttBrowserLayeringTests guards
the graph both ways.

Measured, not assumed: Driver.Mqtt.dll, Polly.Core and Serilog are now absent
from the AdminUI's output. Core.dll remains, because Driver.Sql.Browser has the
SAME violation, undocumented — the old comment blamed this browser for Core and
was wrong. Filed separately rather than fixed in passing.

Full suite green apart from the 7 known pre-existing failures (4 AbLegacy
integration, compose profiles not up; 3 Host.IntegrationTests).

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
2026-07-31 06:17:18 -04:00
parent 3207089f01
commit 7f53bd96ec
23 changed files with 1029 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
+1 -214
View File
@@ -568,7 +568,7 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
MqttClientOptions clientOptions;
try
{
clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
clientOptions = MqttClientOptionsFactory.BuildClientOptions(_options, clientIdSuffix: null, _logger);
}
catch (Exception ex)
{
@@ -1345,217 +1345,4 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
}
}
/// <summary>
/// Assembles the MQTTnet client options from <paramref name="options"/>. Pure — it performs
/// no I/O and touches no network, so it is safe to call on a config-validation path. A
/// configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read lazily by the
/// returned certificate-validation callback at handshake time, not here.
/// </summary>
/// <param name="options">Broker connection settings.</param>
/// <param name="clientIdSuffix">
/// Appended to <see cref="MqttDriverOptions.ClientId"/>, so a browse/probe session can share
/// the configured identity without colliding with the driver's own client id. When both are
/// empty MQTTnet generates a random client id.
/// </param>
/// <param name="logger">Optional logger for the deferred CA-pin load; never receives credentials.</param>
public static MqttClientOptions BuildClientOptions(
MqttDriverOptions options,
string? clientIdSuffix,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
var builder = new MqttClientOptionsBuilder()
.WithTcpServer(options.Host, options.Port)
.WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion))
.WithCleanSession(options.CleanSession)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds))
.WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds));
var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty);
if (!string.IsNullOrWhiteSpace(clientId))
{
builder = builder.WithClientId(clientId);
}
// An empty username means "connect anonymously"; MQTTnet would otherwise send an empty
// CONNECT username, which some brokers treat as a failed auth rather than as anonymous.
if (!string.IsNullOrEmpty(options.Username))
{
builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
}
builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));
return builder.Build();
}
private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger)
{
if (!options.UseTls)
{
tls.UseTls(false);
return;
}
tls.UseTls(true)
// Explicit SNI / hostname-verification target. Left unset, name validation depends on
// how MQTTnet infers the host from the endpoint — pin it to the configured host.
.WithTargetHost(options.Host);
if (options.AllowUntrustedServerCertificate)
{
// Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the
// ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate.
tls.WithAllowUntrustedCertificates(true)
.WithIgnoreCertificateChainErrors(true)
.WithCertificateValidationHandler(static _ => true);
return;
}
if (string.IsNullOrWhiteSpace(options.CaCertificatePath))
{
// No pin: MQTTnet's default handler validates against the OS trust store.
return;
}
tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger));
}
/// <summary>
/// Builds a certificate-validation callback that pins the broker chain to the PEM CA at
/// <paramref name="caCertificatePath"/>. The file is loaded lazily on first validation
/// (i.e. during the TLS handshake) so that this — and therefore
/// <see cref="BuildClientOptions"/> — stays free of I/O. Any failure to load or to chain
/// up to the pinned CA rejects the certificate: fail closed.
/// </summary>
private static Func<MqttClientCertificateValidationEventArgs, bool> CreatePinnedCaValidator(
string caCertificatePath,
ILogger? logger)
{
var trustedRoots = new Lazy<X509Certificate2Collection?>(
() => LoadPemCa(caCertificatePath, logger),
LazyThreadSafetyMode.ExecutionAndPublication);
return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger);
}
private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger)
{
try
{
var collection = new X509Certificate2Collection();
collection.ImportFromPemFile(caCertificatePath);
if (collection.Count != 0)
{
return collection;
}
logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath);
}
catch (Exception ex)
{
logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath);
}
return null;
}
private static bool ValidateAgainstPinnedCa(
MqttClientCertificateValidationEventArgs args,
X509Certificate2Collection? trustedRoots,
string caCertificatePath,
ILogger? logger)
{
if (trustedRoots is null || trustedRoots.Count == 0)
{
// Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store.
return false;
}
// The pin replaces the trust anchor only. Anything else the platform flagged — a hostname
// mismatch, an absent certificate — remains fatal.
if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None)
{
logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors);
return false;
}
if (args.Certificate is null)
{
return false;
}
var presented = args.Certificate as X509Certificate2;
X509Certificate2? owned = null;
if (presented is null)
{
owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert));
presented = owned;
}
try
{
using var chain = new X509Chain();
chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots);
// Revocation is deliberately not checked: the pin exists precisely for self-issued
// dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking
// would fail every such chain. The pin itself is the revocation story — remove the CA.
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
// Seed the intermediates the broker presented during the handshake. A leaf signed by
// an intermediate that chains up to the pinned root is a common broker topology, and
// that intermediate arrives on the wire rather than being installed locally — without
// it in the candidate pool a legitimate trust path simply cannot be built.
// CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue
// self-signed cert smuggled in here cannot become an anchor.
// Both sources are read: platforms differ in whether the peer-supplied intermediates
// land in the incoming chain's elements, its ExtraStore, or both.
if (args.Chain is not null)
{
foreach (var element in args.Chain.ChainElements)
{
chain.ChainPolicy.ExtraStore.Add(element.Certificate);
}
chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore);
}
if (chain.Build(presented))
{
return true;
}
logger?.LogError(
"MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.",
caCertificatePath,
string.Join(", ", chain.ChainStatus.Select(s => s.Status)));
return false;
}
catch (CryptographicException ex)
{
// X509Chain.Build throws on a malformed certificate or an unusable policy. An exception
// escaping a TLS validation callback surfaces as an opaque handshake crash, so classify
// it here and refuse: an unverifiable certificate is a rejected certificate.
logger?.LogError(
ex,
"MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.",
caCertificatePath);
return false;
}
finally
{
owned?.Dispose();
}
}
private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version)
=> version switch
{
MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311,
MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500,
_ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."),
};
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+1 -1
View File
@@ -939,7 +939,7 @@ public sealed class MqttDriver
// ONLINE half without the Will is strictly worse than shipping neither: a crashed node
// would leave a retained ONLINE asserting forever that a dead host is alive, which is the
// exact state the mechanism exists to prevent. The Will belongs in
// MqttConnection.BuildClientOptions; until it is there this flag stays inert and said so.
// MqttClientOptionsFactory.BuildClientOptions; until it is there this flag stays inert and said so.
_logger?.LogWarning(
"MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE "
+ "publishing is not implemented — this driver observes STATE and never publishes one. "
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs
+2 -2
View File
@@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// </summary>
/// <remarks>
/// <para>
/// <b>Reuses <see cref="MqttConnection.BuildClientOptions"/></b> for the connect (TLS,
/// <b>Reuses <see cref="MqttClientOptionsFactory.BuildClientOptions"/></b> for the connect (TLS,
/// CA-pin, credentials, protocol-version mapping) rather than hand-rolling a second
/// connect path — a duplicate would be a security divergence. A distinct client-id suffix
/// (see <see cref="ProbeClientIdPrefix"/>) keeps a probe from colliding with the driver's
@@ -101,7 +101,7 @@ public sealed class MqttDriverProbe : IDriverProbe
MqttClientOptions clientOptions;
try
{
clientOptions = MqttConnection.BuildClientOptions(
clientOptions = MqttClientOptionsFactory.BuildClientOptions(
probeOptions,
ProbeClientIdPrefix + Guid.NewGuid().ToString("N")[..8],
NullLogger.Instance);
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs
-146
View File
@@ -1,146 +0,0 @@
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
/// <summary>
/// The one seam through which a Sparkplug NCMD is published — <see cref="MqttConnection"/> is the
/// production implementation once it grows a publish leg (Task 21); tests substitute a fake so the
/// encode/topic/QoS/retain contract is exercisable without a broker. Deliberately narrower than
/// "publish anything": a Sparkplug driver's only outbound traffic is NCMD/DCMD, so this is not a
/// general MQTT client abstraction in waiting.
/// </summary>
public interface IMqttPublishTransport
{
/// <summary>
/// Publishes one application message. Implementations must be bounded — a broker that accepts
/// PUBLISH and never completes it (QoS &gt; 0 awaiting PUBACK) must fail at a deadline, not
/// hang; <see cref="RebirthRequester.RequestAsync"/> supplies one, but an implementation used
/// directly by a future caller must not rely on that.
/// </summary>
/// <param name="topic">The concrete topic to publish on.</param>
/// <param name="payload">The message body.</param>
/// <param name="qos">Requested QoS, 0–2.</param>
/// <param name="retain">The MQTT <c>retain</c> flag.</param>
/// <param name="cancellationToken">Caller cancellation.</param>
Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken);
}
/// <summary>
/// Encodes and publishes a Sparkplug B rebirth request — an NCMD carrying the well-known
/// <c>Node Control/Rebirth</c> boolean metric, addressed to one edge node's command topic. See
/// Sparkplug B v3.0 spec §6/§7 and design doc §3.6.
/// </summary>
/// <remarks>
/// <para>
/// <b><see cref="Build"/> is pure.</b> It does no I/O and touches no transport, so it is
/// trivially unit-testable and safe to call from anywhere (the ingest state machine on a
/// detected gap, the browser on an operator click) without pulling in a live connection. The
/// bounded, transport-facing publish is the separate <see cref="RequestAsync"/> overload.
/// </para>
/// <para>
/// <b>No <c>seq</c>.</b> Sparkplug's sequence-number stream (§6.4.1) belongs to the messages an
/// edge node itself publishes (NBIRTH/NDATA/DBIRTH/DDATA/NDEATH/DDEATH) — it is how a host
/// detects a gap in <i>that node's</i> outbound stream. An NCMD flows the other direction, host
/// → node, and is not a member of the node's sequence at all: stamping one on here would not be
/// "wrong per spec" so much as meaningless, and a strict edge-node implementation that validated
/// it against its own counter would have grounds to reject the command outright.
/// </para>
/// <para>
/// <b>QoS 0, retain <see langword="false"/> — always.</b> NCMD is ordinary Sparkplug command
/// traffic, published at the same QoS 0 the spec uses for NDATA/DDATA/NBIRTH/DBIRTH (only
/// NDEATH — the broker-issued Will — and STATE use QoS 1). <see langword="retain"/> is the one
/// that matters operationally: a retained NCMD would be replayed by the broker to the edge node
/// on <i>every</i> future subscribe — including the node's own reconnect — driving it into a
/// permanent rebirth loop from a single stale command. See the carry-forward landmine on
/// <c>MqttDriverBrowser</c>/NDEATH-as-Will for the sibling failure shape this type must not
/// introduce on the publish side.
/// </para>
/// <para>
/// <b><c>Datatype</c> and both timestamps are set.</b> <c>Node Control/Rebirth</c> is Boolean;
/// setting <see cref="Payload.Types.Metric.Datatype"/> explicitly (rather than leaving it for the
/// node to infer) and stamping both the metric and payload <c>timestamp</c> matches what a
/// well-behaved command payload carries per spec §6.4.13 and what the Task-25 simulator (and any
/// real Cirrus-Link-derived edge node) expects to validate against. An edge node that treats a
/// missing/ambiguous field as malformed and silently drops the command would otherwise leave the
/// driver retrying forever with no visible error — the whole point of a rebirth request is to
/// recover from exactly that kind of silent desync, so the request itself must not risk being
/// the next cause of one.
/// </para>
/// </remarks>
public static class RebirthRequester
{
/// <summary>The well-known Sparkplug B node-control metric name that triggers a rebirth.</summary>
public const string RebirthMetricName = "Node Control/Rebirth";
/// <summary>
/// Default bounded deadline for <see cref="RequestAsync"/> when the caller does not supply one.
/// Deliberately this type's own value rather than <c>MqttDriverOptions.ConnectTimeoutSeconds</c>
/// — that knob already governs every MQTTnet connect/subscribe operation, and repurposing it
/// would couple the rebirth-publish budget to the connect budget in a way neither side could
/// tune independently (the same reasoning <c>MqttConnection.SubscribeAsync</c> documents for its
/// own deadline).
/// </summary>
public static readonly TimeSpan DefaultRequestTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Builds the wire bytes and target topic for a rebirth-request NCMD, without publishing it.
/// </summary>
/// <param name="groupId">The Sparkplug group id.</param>
/// <param name="edgeNodeId">The Sparkplug edge-node id to address the command to.</param>
/// <returns>The NCMD topic (<c>spBv1.0/{groupId}/NCMD/{edgeNodeId}</c>) and its encoded payload.</returns>
/// <exception cref="ArgumentException"><paramref name="groupId"/>/<paramref name="edgeNodeId"/> is null or empty.</exception>
public static (string Topic, byte[] Bytes) Build(string groupId, string edgeNodeId)
{
// Validates both ids and does the topic assembly — see the "don't hand-concatenate" remark on
// SparkplugTopic.Format itself.
var topic = SparkplugTopic.Format(groupId, SparkplugMessageType.NCMD, edgeNodeId);
var timestampMs = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
var payload = new Payload { Timestamp = timestampMs };
payload.Metrics.Add(new Payload.Types.Metric
{
Name = RebirthMetricName,
Datatype = (uint)TahuDataType.Boolean,
Timestamp = timestampMs,
BooleanValue = true,
});
return (topic, payload.ToByteArray());
}
/// <summary>
/// Builds and publishes a rebirth-request NCMD under a bounded deadline. QoS 0, retain
/// <see langword="false"/> — see the type remarks for why both are fixed rather than
/// configurable.
/// </summary>
/// <param name="transport">The publish seam — the live connection, or a test fake.</param>
/// <param name="groupId">The Sparkplug group id.</param>
/// <param name="edgeNodeId">The Sparkplug edge-node id to address the command to.</param>
/// <param name="cancellationToken">Caller cancellation; linked with the publish deadline.</param>
/// <param name="timeout">
/// The publish deadline. Defaults to <see cref="DefaultRequestTimeout"/> when omitted or
/// non-positive.
/// </param>
/// <exception cref="ArgumentException"><paramref name="groupId"/>/<paramref name="edgeNodeId"/> is null or empty.</exception>
/// <exception cref="ArgumentNullException"><paramref name="transport"/> is null.</exception>
public static async Task RequestAsync(
IMqttPublishTransport transport,
string groupId,
string edgeNodeId,
CancellationToken cancellationToken,
TimeSpan? timeout = null)
{
ArgumentNullException.ThrowIfNull(transport);
var (topic, bytes) = Build(groupId, edgeNodeId);
var deadlineSpan = timeout is { } t && t > TimeSpan.Zero ? t : DefaultRequestTimeout;
using var deadline = new CancellationTokenSource(deadlineSpan);
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, deadline.Token);
await transport.PublishAsync(topic, bytes, qos: 0, retain: false, linked.Token).ConfigureAwait(false);
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugCodec.cs
-313
View File
@@ -1,313 +0,0 @@
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
/// <summary>
/// Decodes Sparkplug-B wire bytes into <see cref="SparkplugPayload"/> — the driver-side projection
/// the ingest state machine consumes. Decode only; the NCMD <i>encode</i> path lives in
/// <c>RebirthRequester</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>Nothing here throws, for any input.</b> This runs on MQTTnet's shared dispatcher thread,
/// behind an unauthenticated firehose the plant's edge nodes publish into. An escaping
/// exception would not degrade one tag — it would stall or kill delivery for every
/// subscription on the connection. Garbage bytes, a truncated body, a zero-length payload, a
/// valid protobuf of some other schema and a pathologically nested Template all resolve to a
/// verdict: <see cref="TryDecode"/> returns <see langword="false"/> and
/// <see cref="Decode"/> returns <see cref="SparkplugPayload.Invalid"/>.
/// </para>
/// <para>
/// <b>Zero-length input is invalid, not empty.</b> Protobuf would happily parse zero bytes as
/// "a Payload with every field defaulted", but in Sparkplug an empty MQTT body is never a
/// legitimate message — treating it as a well-formed payload carrying no metrics is how a
/// truncated-to-nothing body gets mistaken for a real one. It is reported as invalid.
/// </para>
/// <para>
/// <b>Explicit presence is preserved, everywhere.</b> The vendored schema is proto2 precisely
/// so absence and zero stay distinguishable, and every projection below reads
/// <c>Has{Seq,Name,Alias,Datatype,Timestamp,IsNull}</c> rather than testing a value against its
/// default. Two cases make this load-bearing rather than pedantic: a Sparkplug NBIRTH is
/// REQUIRED to carry <c>seq = 0</c>, and every DATA metric after a birth carries an alias with
/// <b>no</b> name and <b>no</b> datatype. A zero-check decoder reports "no sequence" for every
/// birth and "" for every DATA metric name.
/// </para>
/// <para>
/// <b>Values are projected raw — nothing is coerced or reinterpreted here.</b> A metric's value
/// arrives as the CLR type of the wire field that carried it (see
/// <see cref="SparkplugMetric.Value"/>); coercion against the authored tag's declared
/// <c>DriverDataType</c> is the consumer's job, and follows this driver's standing rule that a
/// value which does not fit is refused rather than silently converted. The one wire-level
/// translation this type does offer is <see cref="ReinterpretSigned"/>, because Sparkplug's
/// two's-complement-in-an-unsigned-field encoding of signed integers is a property of the
/// <i>wire</i>, not of the tag.
/// </para>
/// <para>
/// <b><c>DataSet</c> and <c>Template</c> metrics are out of scope for v1.</b> They decode to
/// <see cref="SparkplugValueKind.Unsupported"/> with a null value — deliberately visible, so a
/// consumer can warn about a metric it cannot serve instead of silently treating it as one
/// that was never published.
/// </para>
/// </remarks>
public static class SparkplugCodec
{
/// <summary>Decodes Sparkplug-B wire bytes, reporting success rather than throwing.</summary>
/// <param name="wire">The MQTT message body.</param>
/// <param name="payload">
/// The decoded payload on success; <see cref="SparkplugPayload.Invalid"/> otherwise. Never
/// <see langword="null"/>.
/// </param>
/// <returns>
/// <see langword="true"/> when the bytes parsed as a Sparkplug-B <c>Payload</c>. Protobuf is a
/// permissive, self-describing-only-by-convention format, so this means "parsed" rather than
/// "was genuinely produced by a Sparkplug edge node" — unknown fields are preserved by
/// Google.Protobuf and simply ignored here.
/// </returns>
public static bool TryDecode(ReadOnlySpan<byte> wire, out SparkplugPayload payload)
{
payload = SparkplugPayload.Invalid;
if (wire.IsEmpty)
{
return false;
}
try
{
var proto = Payload.Parser.ParseFrom(wire);
var metrics = proto.Metrics.Count == 0
? []
: new SparkplugMetric[proto.Metrics.Count];
for (var i = 0; i < proto.Metrics.Count; i++)
{
metrics[i] = ProjectMetric(proto.Metrics[i]);
}
payload = new SparkplugPayload(
IsValid: true,
Seq: proto.HasSeq ? proto.Seq : null,
TimestampMs: proto.HasTimestamp ? proto.Timestamp : null,
Metrics: metrics);
return true;
}
catch (Exception)
{
// Deliberately broad, and deliberately spanning the projection as well as the parse.
// InvalidProtocolBufferException covers malformed, truncated and over-nested input, but
// this is the dispatcher-thread boundary and the projection walks an attacker-shaped
// object graph: the contract is a verdict for EVERY input, so a guard that stopped at
// the parse would be a contract with a hole in it. Narrowing the catch would trade the
// guarantee for a taxonomy nobody downstream can act on.
payload = SparkplugPayload.Invalid;
return false;
}
}
/// <summary>
/// Decodes Sparkplug-B wire bytes, returning <see cref="SparkplugPayload.Invalid"/> when they
/// cannot be decoded.
/// </summary>
/// <param name="wire">The MQTT message body.</param>
/// <returns>The decoded payload — never <see langword="null"/>.</returns>
/// <remarks>
/// The convenience shape. <b>Check <see cref="SparkplugPayload.IsValid"/>:</b> an undecodable
/// body and a genuinely metric-less one both present as an empty
/// <see cref="SparkplugPayload.Metrics"/>, and only the flag tells them apart. Prefer
/// <see cref="TryDecode"/> where the verdict drives control flow.
/// </remarks>
public static SparkplugPayload Decode(ReadOnlySpan<byte> wire) =>
TryDecode(wire, out var payload) ? payload : SparkplugPayload.Invalid;
/// <summary>
/// Reinterprets a raw metric value as the signed integer Sparkplug encoded it as, when its
/// datatype is a signed integer type; returns the value unchanged for every other datatype.
/// </summary>
/// <param name="value">
/// A <see cref="SparkplugMetric.Value"/> — a <see cref="uint"/> for the 8/16/32-bit arms, a
/// <see cref="ulong"/> for the 64-bit arm.
/// </param>
/// <param name="datatype">The metric's Sparkplug datatype, from its birth or its own field.</param>
/// <returns>
/// <see cref="sbyte"/> / <see cref="short"/> / <see cref="int"/> / <see cref="long"/> for the
/// signed integer datatypes; <paramref name="value"/> untouched otherwise (including when it is
/// <see langword="null"/> or not the wire type the datatype implies).
/// </returns>
/// <remarks>
/// Sparkplug carries <c>Int8</c>/<c>Int16</c>/<c>Int32</c> in the <b>unsigned</b>
/// <c>int_value</c> field and <c>Int64</c> in the unsigned <c>long_value</c> field, as two's
/// complement. Handing the raw field to a consumer publishes <c>4294967254</c> for a tag whose
/// value is <c>-42</c> — a wrong value that looks entirely plausible, arrives with Good
/// quality, and is invisible until someone reads a gauge. This is the single call that undoes
/// it, and it is separate from the decode itself because a DATA metric carries no datatype of
/// its own: only the consumer, holding the alias table built from the birth, knows which
/// datatype applies.
/// </remarks>
public static object? ReinterpretSigned(object? value, TahuDataType datatype) => datatype switch
{
TahuDataType.Int8 when value is uint raw => unchecked((sbyte)raw),
TahuDataType.Int16 when value is uint raw => unchecked((short)raw),
TahuDataType.Int32 when value is uint raw => unchecked((int)raw),
TahuDataType.Int64 when value is ulong raw => unchecked((long)raw),
_ => value,
};
/// <summary>Projects one generated metric onto the driver-side shape, presence intact.</summary>
/// <param name="metric">The generated metric.</param>
/// <returns>The projection.</returns>
private static SparkplugMetric ProjectMetric(Payload.Types.Metric metric)
{
var (kind, value) = ProjectValue(metric);
return new SparkplugMetric(
Name: metric.HasName ? metric.Name : null,
Alias: metric.HasAlias ? metric.Alias : null,
// Cast, never filter: an index this build's enum does not define is a future Sparkplug
// revision or a misbehaving publisher, and dropping it would deny the mapping layer the
// only evidence it has.
DataType: metric.HasDatatype ? (TahuDataType)metric.Datatype : null,
TimestampMs: metric.HasTimestamp ? metric.Timestamp : null,
ValueKind: kind,
Value: value);
}
/// <summary>Resolves a metric's value <c>oneof</c> to a kind plus a raw CLR value.</summary>
/// <param name="metric">The generated metric.</param>
/// <returns>The value kind and the projected value.</returns>
/// <remarks>
/// <c>is_null</c> wins over the <c>oneof</c>: the Sparkplug spec's whole reason for the field is
/// that some datatypes have no spare sentinel, so a publisher that sets it means "null" even if
/// a value field is also populated.
/// </remarks>
private static (SparkplugValueKind Kind, object? Value) ProjectValue(Payload.Types.Metric metric)
{
if (metric.HasIsNull && metric.IsNull)
{
return (SparkplugValueKind.Null, null);
}
return metric.ValueCase switch
{
Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue),
Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue),
Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue),
Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue),
Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue),
Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue),
// Copied out of the ByteString: the projection must outlive the parsed message.
Payload.Types.Metric.ValueOneofCase.BytesValue =>
(SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray()),
// v1 scope boundary — decoded as a visible refusal, not as an exception or a silent drop.
Payload.Types.Metric.ValueOneofCase.DatasetValue => (SparkplugValueKind.Unsupported, null),
Payload.Types.Metric.ValueOneofCase.TemplateValue => (SparkplugValueKind.Unsupported, null),
Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null),
_ => (SparkplugValueKind.Absent, null),
};
}
}
/// <summary>
/// A decoded Sparkplug-B payload: the driver-side projection of the generated <c>Payload</c>,
/// carrying only what the ingest state machine consumes.
/// </summary>
/// <param name="IsValid">
/// <see langword="false"/> for a payload that could not be decoded. An invalid payload also has an
/// empty <paramref name="Metrics"/> list, so <b>this flag is the only thing distinguishing an
/// undecodable body from a legitimately metric-less one</b>.
/// </param>
/// <param name="Seq">
/// The payload sequence number, or <see langword="null"/> when the message carried none. Kept as
/// the wire's <see cref="ulong"/> rather than narrowed to a byte: the Sparkplug range is 0–255, but
/// a publisher that violates it is reporting a fact the consumer should be able to see and reject,
/// not one this layer should silently truncate into a plausible-looking sequence number.
/// </param>
/// <param name="TimestampMs">The payload timestamp in Sparkplug epoch milliseconds, or null when absent.</param>
/// <param name="Metrics">The payload's metrics, in wire order. Never <see langword="null"/>.</param>
public sealed record SparkplugPayload(
bool IsValid,
ulong? Seq,
ulong? TimestampMs,
IReadOnlyList<SparkplugMetric> Metrics)
{
/// <summary>The shared "could not decode this" result.</summary>
public static readonly SparkplugPayload Invalid = new(false, null, null, []);
}
/// <summary>One decoded Sparkplug metric.</summary>
/// <param name="Name">
/// The metric's stable name, or <see langword="null"/> when the message omitted it — which every
/// real DATA metric after a birth does, carrying only <paramref name="Alias"/>. Distinguishing
/// this from an empty string is the reason the vendored schema is proto2.
/// </param>
/// <param name="Alias">The per-birth alias, or <see langword="null"/> when the metric carried none.</param>
/// <param name="DataType">
/// The metric's declared datatype, or <see langword="null"/> when absent (again, the normal case
/// for a DATA metric — its datatype comes from the birth). An index this build's enum does not
/// define is preserved as an undefined enum value rather than dropped.
/// </param>
/// <param name="TimestampMs">
/// The metric's own acquisition timestamp in Sparkplug epoch milliseconds, or
/// <see langword="null"/> when it carried none — in which case the payload's timestamp applies.
/// </param>
/// <param name="ValueKind">
/// What <paramref name="Value"/> means. Check this before reading the value: a null value is
/// ambiguous between "explicitly null", "absent" and "a kind v1 does not support".
/// </param>
/// <param name="Value">
/// The <b>raw</b> wire value, boxed: <see cref="uint"/> (<c>int_value</c>), <see cref="ulong"/>
/// (<c>long_value</c>), <see cref="float"/>, <see cref="double"/>, <see cref="bool"/>,
/// <see cref="string"/> or <see cref="byte"/><c>[]</c> — and <see langword="null"/> for every
/// <paramref name="ValueKind"/> other than <see cref="SparkplugValueKind.Scalar"/>.
/// <para>
/// <b>Signed integers arrive as their unsigned two's-complement wire value</b> — a
/// <c>DataType.Int32</c> metric holding -42 is a <see cref="uint"/> of 4294967254 here. Run it
/// through <see cref="SparkplugCodec.ReinterpretSigned"/> with the metric's datatype (from the
/// birth, for a DATA metric) before publishing it.
/// </para>
/// <para>
/// Boxing is deliberate: the consumer coerces against the authored tag's declared
/// <c>DriverDataType</c> and hands the result to a <c>DataValueSnapshot</c>, which boxes
/// anyway, so a discriminated-union shape would buy an unboxing hop and cost every consumer a
/// switch over a dozen arms.
/// </para>
/// </param>
public readonly record struct SparkplugMetric(
string? Name,
ulong? Alias,
TahuDataType? DataType,
ulong? TimestampMs,
SparkplugValueKind ValueKind,
object? Value);
/// <summary>What a decoded metric's <see cref="SparkplugMetric.Value"/> represents.</summary>
/// <remarks>
/// Three of the four members have a null <see cref="SparkplugMetric.Value"/> and mean entirely
/// different things, which is exactly why the distinction is carried explicitly rather than left
/// for a consumer to infer from a null.
/// </remarks>
public enum SparkplugValueKind
{
/// <summary>The metric's value <c>oneof</c> carried nothing at all.</summary>
Absent = 0,
/// <summary>The metric set <c>is_null</c>: it exists, and its value is explicitly null.</summary>
Null,
/// <summary><see cref="SparkplugMetric.Value"/> holds the raw wire value.</summary>
Scalar,
/// <summary>
/// A <c>DataSet</c>, <c>Template</c> or extension value — decoded and reported, but not
/// supported in v1. The consumer should warn and skip the metric rather than treat it as
/// missing.
/// </summary>
Unsupported,
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs
+1 -1
View File
@@ -1038,7 +1038,7 @@ public sealed class SparkplugIngestor
/// while their host is offline. Shipping the ONLINE half without the Will is strictly worse
/// than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting
/// forever that a dead host is alive, which is precisely the state the mechanism exists to
/// prevent. The Will has to be set inside <c>MqttConnection.BuildClientOptions</c> — a
/// prevent. The Will has to be set inside <c>MqttClientOptionsFactory.BuildClientOptions</c> — a
/// different file, a different task — so this task implements the observation half and
/// <see cref="MqttDriver"/> warns when the flag is set. See the design doc §7/§8.
/// </para>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugTopic.cs
-291
View File
@@ -1,291 +0,0 @@
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;
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj
+1
View File
@@ -14,6 +14,7 @@
<ItemGroup>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj"/>
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core\ZB.MOM.WW.OtOpcUa.Core.csproj"/>
</ItemGroup>