Every concern the 7-agent review parked is now decided and integrated: - universal browser: in-flight capture coalescing keyed (driverType, config-hash) + global cap of 4 concurrent captures; CanBrowse catches a throwing TryCreate and defensively shuts down the throwaway instance; cleanup ShutdownAsync bounded at 10s (R2-01); picker Refresh = close + re-capture. Coalesced sessions share an immutable tree, so DisposeAsync drops the session's reference, not the tree. - mtconnect: UNAVAILABLE pinned to BadNoCommunication (fleet's BadCommunicationError stays reserved for the driver's own transport failures); TIME_SERIES sampleCount flows into ArrayDim; Subscribe timeout bounds only the stream-start handshake; library version/TFM folded into the license checklist. - mqtt: browse rebirth is an explicit DriverOperator-gated "Request rebirth" button (RequestRebirthAsync(scope)) - never fired by open/root/expand; hand-rolled reconnect loop committed for P1 (MQTTnet v5 dropped ManagedMqttClient); Wave-2-start library re-verification checkbox; implement from the Sparkplug v3.0 spec text. - bacnet: P1 opens with a first-spike checklist (package TFM, BBMD/ segmented-RPM/COV API smoke, unicast-I-Am fixture behavior); AdminUI-node browse registers a second foreign device - BBMD FD-table sizing note. - sql-poll: split-node browse needs env parity for Sql__ConnectionStrings__ refs on admin nodes (actionable error + session-only pasted literal); one-row-per-key query contract (last-wins + rate-limited warning); absent key -> BadNoData; operationTimeout > commandTimeout authoring rule; Browser->Driver.Sql SqlClient transitive on AdminUI accepted on record. - omron: FINS framer gated on W227 + live golden vectors; CIP string layout is the first P1 live-gate item; AbCip harness static-init anti-pattern note; writable-defaults-true operator warning. - modbus-rtu: first P1 step confirms pymodbus exposes framer=rtu on a TCP server (custom-script fallback otherwise). - program doc: new cross-cutting rule - driver ctors must be connection-free (the universal browser's CanBrowse throwaway instances depend on it).
44 KiB
MQTT / Sparkplug B driver — executable implementation design
Status: design, build-ready. 2026-07-15.
Research input: docs/research/drivers/mqtt-sparkplug.md (verdicts + protocol/library survey — this doc turns that into a build plan; it does not re-litigate the research).
Related: docs/plans/2026-07-15-universal-discovery-browser-design.md (why MQTT needs a bespoke browser despite the universal one — §4), docs/plans/2026-06-12-galaxy-standard-driver-design.md (standard Equipment-kind driver shape + fire-and-forget write precedent), docs/plans/2026-05-28-driver-browsers-design.md (IDriverBrowser/IBrowseSession pattern reused by the observation browser).
1. Motivation + scope
Build a standard Equipment-kind driver — same composable-IDriver shape as Modbus / S7 / AbCip / TwinCAT / FOCAS / OpcUaClient — that ingests plain MQTT and Sparkplug B into the OtOpcUa OPC UA / UNS address space.
- OtOpcUa is a Unified-Namespace product, and Sparkplug B is the de-facto UNS-over-MQTT payload standard (mandated topic namespace + protobuf schema + birth-certificate discovery + stateful sessions). Sparkplug alignment is therefore the strategic headline; plain MQTT/JSON is the broad-compatibility fallback for brokers that carry raw JSON topics.
- This is a subscribe-first driver. Unlike the polled drivers (Modbus/S7/AbCip/FOCAS with
PollGroupEngine), it holds a live broker connection for its whole lifetime and raisesOnDataChangefrom the MQTT receive callback. The closest existing analog is the native-push side ofOpcUaClientDriver. - In scope (this design): connect/TLS/auth/reconnect; topic + Sparkplug subscribe →
OnDataChange; last-valueIReadable; authored-onlyITagDiscovery;IRediscoverable;IHostConnectivityProbe; a bespoke observation browser; typed AdminUI tag editor + validator; Docker fixtures + env-gated live suite. - Out of scope for v1: write-through (
IWritable— Sparkplug NCMD/DCMD + plain publish) is deferred to phase 3 (§3.4, §10);DataSet/TemplateSparkplug metrics;Bytes/Filemetrics beyond a raw-String fallback.
Canonical DriverType string: "Mqtt" — one string used identically across the AdminUI driver page, probe, factory, editor map, validator, and browser. (Heed the Modbus "ModbusTcp" vs "Modbus" mismatch lesson: a divergent string silently drops tags to the raw-JSON editor / faults resolution. Pick "Mqtt" and grep to enforce it.)
2. Project layout
Three new projects under src/Drivers/, mirroring the OpcUaClient triad (.Driver + .Contracts + .Browser). All net10.0, Nullable+ImplicitUsings enabled, TreatWarningsAsErrors=true opted in per-csproj (mandatory for new projects — Directory.Build.props deliberately does not set it globally because legacy test projects would fail the build).
src/Drivers/
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ # options + tag DTOs + parser — zero transport deps
MqttDriverOptions.cs # driver-config DTO (broker conn + mode + sparkplug/plain sub-objects)
MqttMode.cs, MqttPayloadFormat.cs # enums (Plain|SparkplugB ; Json|Raw|Scalar)
MqttTagDefinition.cs # parsed per-tag descriptor (plain OR sparkplug variant)
MqttEquipmentTagParser.cs # TagConfig JSON → MqttTagDefinition (mirrors ModbusEquipmentTagParser)
SparkplugDataType.cs # Sparkplug metric-datatype enum + → DriverDataType map
ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ # runtime driver — references .Contracts + Core.Abstractions + Core
MqttDriver.cs # IDriver, ITagDiscovery, ISubscribable, IReadable,
# IHostConnectivityProbe, IRediscoverable (+ IWritable phase 3)
MqttDriverProbe.cs # IDriverProbe — broker CONNECT handshake
MqttDriverFactoryExtensions.cs # DriverTypeName="Mqtt"; Register(registry, loggerFactory)
MqttConnection.cs # MQTTnet client wrapper: connect/TLS/auth/reconnect/backoff
LastValueCache.cs # FullReference → last DataValueSnapshot (seeds IReadable)
Sparkplug/
SparkplugCodec.cs # Tahu-proto decode (Payload/Metric); encode NCMD (phase 3)
AliasTable.cs # per edge-node/device alias→(name,datatype); rebuilt each birth
BirthCache.cs # NBIRTH/DBIRTH metric catalog (name/alias/datatype/last value)
SequenceTracker.cs # seq (0-255 wrap) gap detection; bdSeq death→birth tie
RebirthRequester.cs # publishes NCMD Node Control/Rebirth=true
SparkplugTopic.cs # parse/format spBv1.0/{group}/{type}/{node}[/{device}]
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ # bespoke observation browser — references .Contracts + Commons(.Browsing) + MQTTnet
MqttDriverBrowser.cs # IDriverBrowser (DriverType="Mqtt"); OpenAsync → MqttBrowseSession
MqttBrowseSession.cs # IBrowseSession over an accumulating observed tree
src/Server/.../AdminUI/
Uns/TagEditors/MqttTagConfigModel.cs # typed working model (FromJson/ToJson/Validate; preserves unknown keys)
Uns/TagEditors/TagConfigEditorMap.cs # + ["Mqtt"] = MqttTagConfigEditor
Uns/TagEditors/TagConfigValidator.cs # + ["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate()
Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor
EndpointRouteBuilderExtensions.cs # + AddSingleton<IDriverBrowser, MqttDriverBrowser>()
src/Server/.../Host/Drivers/DriverFactoryBootstrap.cs
# + MqttDriverFactoryExtensions.Register(registry, loggerFactory)
# + TryAddEnumerable(IDriverProbe, MqttDriverProbe)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ # unit — codec/alias/seq/parser (net10, macOS-safe)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ # env-gated live suite (MQTT_FIXTURE_ENDPOINT)
tests/Drivers/.../Docker/docker-compose.yml # mosquitto + sparkplug-sim + json-publisher (project=lmxopcua)
Why the split .Contracts project: it carries MqttDriverOptions + MqttTagDefinition + MqttEquipmentTagParser + the enums with zero transport (MQTTnet) dependency, exactly like Driver.Modbus.Contracts depends only on Modbus.Addressing + Core.Abstractions. This lets the AdminUI typed editor (and any tooling) reference the DTO shape/enums without dragging MQTTnet into AdminUI. The runtime .Driver and .Browser both reference .Contracts and add MQTTnet.
2.1 Libraries + the pinning caveat
| Concern | Choice | Package |
|---|---|---|
| MQTT transport (both modes) | MQTTnet v5 — MIT, .NET Foundation (dotnet/MQTTnet), the standard .NET MQTT client. v5 explicitly ships a net10.0 TFM. Client + TLS + MQTT 3.1.1/5.0. Note: v5 dropped v4's ManagedMqttClient — reconnect is hand-rolled in P1 (§8). |
MQTTnet (add PackageVersion Include="MQTTnet" Version="5.x" to Directory.Packages.props) |
| Sparkplug decode | Hand-rolled Tahu protobuf (recommended) — generate C# from Tahu's sparkplug_b.proto with Google.Protobuf (Grpc.Tools/protoc), decode over a single MQTTnet-5 client. Small, stable schema; we own the alias/seq/rebirth state machine exactly as we want it. |
Google.Protobuf + Grpc.Tools (build-time) |
| Alternative Sparkplug decode | SparkplugNet — MIT, provides Application/Node/Device base classes, saves the birth/alias plumbing. Rejected as primary for two repo-specific reasons (below). |
SparkplugNet (only if the risk clears) |
Pinning caveat (load-bearing — this is why hand-roll is recommended). This repo uses central package management (Directory.Packages.props, ManagePackageVersionsCentrally=true) and deliberately leaves CentralPackageTransitivePinningEnabled OFF because turning it on breaks the Roslyn 5.0.0/4.12.0 split (see the "Transitive pinning breaks Roslyn build" memory + Directory.Packages.props:103 comment). SparkplugNet pins MQTTnet 4.3.x transitively and has no explicit net10.0 TFM yet. A direct MQTTnet-5 reference (needed for the plain-mode transport + net10) will coexist in the graph with SparkplugNet's transitive 4.x, and because transitive pinning is off we cannot cleanly force-align the two. Net: SparkplugNet risks a hard-to-resolve MQTTnet 4/5 diamond in exactly the pinning configuration this repo can't change.
Decision: hand-roll Tahu-proto decode over one MQTTnet-5 client. It removes the diamond and the net10-TFM concern entirely, and the Sparkplug payload schema (Payload { seq, timestamp, Metric[], body }, Metric { name, alias, timestamp, datatype, value, is_null, ... }) is ~90 lines of .proto — a bounded, well-understood effort. Phase 1 (plain MQTT) references only MQTTnet-5, so we validate MQTTnet-v5-on-net10 before ever touching Sparkplug. (If a future need for SparkplugNet arises, the phase-2 codec seam SparkplugCodec is the single swap point.)
- Re-verify the external-library claims at Wave-2 start (before P1 coding): confirm the exact MQTTnet v5 version +
net10.0TFM on NuGet, and re-check the SparkplugNet MQTTnet-4.x transitive-pin claim — it is directionally right from the research but was unverifiable offline. The hand-roll decision does not hinge on that pin being exact (it stands on the net10-TFM concern + owning the alias/seq/rebirth state machine regardless), so this is a confirm-the-record check, not a decision gate.
Vendor the proto: commit Tahu sparkplug_b.proto into Driver.Mqtt.Contracts/Protos/ (pinned to a known Tahu commit, with a provenance comment) and generate at build via Grpc.Tools. Don't fetch at build time.
Existing protobuf tooling, honestly: Google.Protobuf is already centrally pinned (Directory.Packages.props:31, 3.34.1) and consumed by Driver.Galaxy — but the Galaxy proto types ship pre-generated inside the ZB.MOM.WW.MxGateway.* NuGets; there is no .proto file or Grpc.Tools protoc codegen anywhere in this repo today. So Grpc.Tools is a new (build-time-only, PrivateAssets=all) package and this is the repo's first in-repo proto codegen step. If that step proves objectionable, the fallback is checking in the generated C# (the schema is small and stable) with the same provenance comment.
3. Capability mapping
MqttDriver : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable (+ IWritable in phase 3). Not IHistoryProvider (historization is the server-side isHistorized path, unchanged). Not a poller — no PollGroupEngine.
3.0 Lifecycle (IDriver)
InitializeAsync(configJson, ct)— deserializeMqttDriverOptions(sharedJsonSerializerOptionswithJsonStringEnumConverter, §6); buildMqttConnection; connect to broker under a bounded connect deadline (§8); set LWT/clean-session; subscribe to the mode-appropriate filter (§3.1). PublishSTATEbirth ifactAsPrimaryHost(Sparkplug).ReinitializeAsync— apply a config delta in place: if only tag membership changed, re-subscribe; if broker connection changed, tear down + reconnect. Never crash the process (marks Faulted → nodes go Bad, per driver-stability Tier A/B).ShutdownAsync— publishSTATE OFFLINE(if primary host), disconnect the MQTTnet client, dispose.GetHealth— Connected/Reconnecting/Faulted + last-message-age; drives ServiceLevel + status dashboard.GetMemoryFootprint/FlushOptionalCachesAsync— footprint = last-value cache + birth caches + alias tables. The runtime driver has no optional caches to flush: birth/alias tables are required-for-correctness (flushing them mis-routes aliased DDATA —IDriver.FlushOptionalCachesAsyncforbids flushing correctness state) and the last-value cache is whatIReadableserves. Flush is a no-op; all three structures are bounded by the authored tag set + observed births. (The observation browser's accumulating tree lives in the AdminUI-sideMqttBrowseSession, not in the runtime driver.)
3.1 Subscribe (ISubscribable — primary)
-
Plain mode: subscribe to each authored tag's
topicat itsqos(dedupe identical topics; coalesce shared wildcards where the operator authored one). On message: match topic → authored tag(s), extract value atjsonPath(or raw/scalar), raiseOnDataChange(handle, FullReference, snapshot). Seed initial value from the broker's retained message replayed on subscribe (OPC UA initial-data convention). -
Sparkplug mode: subscribe
spBv1.0/{groupId}/#(+ the host-state topic when acting as primary host —spBv1.0/STATE/{hostId}with a JSON{online, timestamp}payload in spec v3.0; pre-3.0 peers used top-levelSTATE/{hostId}withONLINE/OFFLINEtext — target the v3.0 form, tolerate the legacy one on receive). Maintain aBirthCache+AliasTableper (edgeNode, device) from NBIRTH/DBIRTH. On NDATA/DDATA: resolve each metric by alias → (name, datatype) via the alias table, map to the authored tag'sFullReferenceby stable metric name, raiseOnDataChange. On NDEATH/DDEATH: emit STALE / Bad-quality snapshots for that node's/device's metrics. On missed birth / unknown alias / seq gap: issue a rebirth NCMD (§3.6). Spec version: implement from the Sparkplug v3.0 spec text — the research report cites v2.2, which is superseded on exactly the STATE topic/payload point above (this doc already carries the correct v3.0 form). -
publishingIntervalis advisory (MQTT/Sparkplug are event-driven). Use it only for optional server-side coalescing/deadband, mirroring Modbus'sShouldPublish.SubscribeAsyncreturns anISubscriptionHandlewhoseDiagnosticIdnames the mode+filter;OnDataChangefires from the MQTT receive handler for the whole driver lifetime.
SubscribeAsync maps the requested FullReferences onto the live receive stream: it registers them in a FullReference → MqttTagDefinition resolver (parsed once, cached — reuse the shared EquipmentTagRefResolver<TDef> from Core.Abstractions, exactly as Modbus does), so an inbound message/metric routes to the right node.
3.2 Read (IReadable)
MQTT has no request/response read. ReadAsync(fullReferences) returns the last-value cache (LastValueCache): last message seen on the topic (plain) or last metric value from data/birth (Sparkplug). Seed from the retained message at connect (plain) and from birth values (Sparkplug). If nothing has been observed yet, return a GoodNoData/uncertain snapshot per reference (never throw the batch — per IReadable contract, per-reference status is the snapshot's StatusCode). This satisfies OPC UA reads + HistoryRead-AtTime without a live round-trip.
3.3 Discover (ITagDiscovery) — authored-only
DiscoverAsync replays authored tags only, both modes — it does not auto-provision from wildcard/birth observation. Rationale (from research §2.2): a chatty broker or a large Sparkplug plant would produce an unbounded auto-provisioned address space. Wildcard/birth observation is a browse-time concern (§4), not a runtime-provisioning one.
- The operator authors tags (topic+jsonPath, or group/node/device/metric) via the typed editor / observation browser.
DiscoverAsyncstreams exactly those authored tags into theIAddressSpaceBuilder, each as aVariable(browseName, displayName, DriverAttributeInfo{ FullName, DriverDataType, ... }). RediscoverPolicy:- Plain:
Once(authored set is static per published config — like Modbus). - Sparkplug:
UntilStable(metric datatypes fill in asynchronously as births arrive after connect; keep re-streaming until the authored tags' resolved datatypes stop changing — like FOCAS's FixedTree). This lets a tag authored before its first birth pick up its true datatype once the birth lands.
- Plain:
SupportsOnlineDiscovery => false(the default-member from the universal-browser design). This is deliberate and decoupled from browseability (§4): runtime discovery is authored-only, so the universal Discover-backed browser would only replay authored tags — useless for discovering new topics/metrics. Leaving the flag false keeps MQTT off the universal fallback; the bespoke observation browser (§4) is what backs the picker.
3.4 IRediscoverable
Sparkplug fires OnRediscoveryNeeded when a new DBIRTH (a device joins) or a rebirth with a changed metric set is observed — the Core re-runs DiscoverAsync + diffs. ScopeHint = the edge-node/device folder path so the rebuild is surgical. Plain mode never fires it (no backend change signal — a config republish is the only tag-set change), so plain-only drivers effectively no-op this interface.
3.5 Data-type mapping
Sparkplug metric datatype (or inferred/authored plain type) → DriverDataType (Boolean, Int16, Int32, Int64, UInt16, UInt32, UInt64, Float32, Float64, String, DateTime, Reference):
Sparkplug DataType |
DriverDataType |
OPC UA |
|---|---|---|
| Int8, Int16 | Int16 (Int8 widens) |
Int16 |
| Int32 | Int32 |
Int32 |
| UInt8, UInt16 | UInt16 (UInt8 widens) |
UInt16 |
| UInt32 | UInt32 |
UInt32 |
| Int64 | Int64 |
Int64 |
| UInt64 | UInt64 |
UInt64 |
| Float | Float32 |
Float |
| Double | Float64 |
Double |
| Boolean | Boolean |
Boolean |
| String, Text, UUID | String |
String |
| DateTime | DateTime |
DateTime |
| Bytes, File | String (base64/raw fallback, v1) |
String |
| DataSet, Template | unsupported v1 — skip + warn, or expose raw-JSON as String |
— |
*Array variants |
element type + IsArray=true, ArrayDim from birth |
ValueRank=1 |
Live map in SparkplugDataType.ToDriverDataType() (.Contracts). Plain mode: honour an explicit per-tag dataType (preferred — inference is brittle); fall back to inferring from the JSON token at jsonPath (number→Float64/Int64, bool→Boolean, string→String). Int8/UInt8 widen (no OPC UA signed-byte distinction needed).
3.6 Sparkplug state-machine correctness (the #1 risk)
The alias→metric map, seq-gap detection, late-join rebirth, and death→STALE are load-bearing and easy to get subtly wrong. Invariants the implementation must hold:
- Bind tags by stable metric NAME, never by alias. The authored
TagConfigstoresmetricName; the alias is a per-birth cache only. An alias can be reused across a rebirth for a different metric — binding by alias silently mis-routes data to the wrong tag. - Rebuild the alias table on every (re)birth. NBIRTH/DBIRTH replaces the
(edgeNode,device)AliasTablewholesale; never merge. - Track
seq(0–255, wraps) per edge-node. A gap (or an unknown alias, or a data message before any birth) ⇒ request rebirth (RebirthRequesterpublishesNCMDwritingNode Control/Rebirth = true), gated onrequestRebirthOnGap.bdSeqties an NDEATH to its NBIRTH. - Death → STALE. NDEATH/DDEATH (LWT-driven) emits Bad/uncertain-quality snapshots for all affected metrics; the next birth restores Good.
- Late join. On connect the driver may have missed births — proactively request rebirth (or wait for the retained STATE + natural rebirth) so metadata is recovered. (This runtime-driver late-join rebirth is automatic and correct — the driver owns its subscription; it is the same NCMD primitive the browser exposes only as the explicit operator "Request rebirth" action, §4.1.)
Unit-test these explicitly against a controllable simulator: rebirth, missed-birth, seq-gap, alias-reuse-across-rebirth, death→stale→rebirth.
3.7 Write (IWritable) — verdict: NOT in v1; phase 3, fire-and-forget optimistic-Good
Writes are possible but deferred (research §2.4). When shipped in phase 3:
- Sparkplug: publish
NCMD/DCMDtospBv1.0/{group}/NCMD/{node}[/{device}]with the target metric (by name+alias) + value. No synchronous write ack — success is only observable when the edge echoes the change in the next N/DDATA. This mirrors the Galaxy gateway's fire-and-forget write ("Galaxy can never surface a write failure"): returnGoodoptimistically; the value self-corrects on the next data message. RespectWriteIdempotent— Sparkplug commands (rebirth, pulse) are frequently non-idempotent, so default retry off. - Plain: publish to a configured command/write topic with a formatted payload.
v1 ships read/subscribe/discover only — a genuinely useful UNS-ingest driver. MqttDriver does not implement IWritable in v1 (the interface is composable; its absence means nodes materialize read-only).
4. Bespoke observation browser
Decision — reconciling with the universal Discover-backed browser
The universal browser (docs/plans/2026-07-15-universal-discovery-browser-design.md) captures whatever DiscoverAsync streams and re-presents it, gated on SupportsOnlineDiscovery. MQTT/Sparkplug is the one roadmap driver that needs a bespoke browser despite the universal one, because:
- Runtime
DiscoverAsyncis authored-only (§3.3). A universal Discover-backed browser would replay only tags the operator already authored — it would not discover new topics/metrics, which is the entire point of a picker. - MQTT/Sparkplug discovery is observational: you learn what exists only by listening for a bounded window. That's a fundamentally different session shape than "connect → one-shot enumerate → disconnect" — the tree fills in over time, and Sparkplug can be accelerated on demand by an operator-requested rebirth NCMD (§4.1 — explicitly not automatic).
Considered + rejected: feeding the Sparkplug BirthCache into DiscoverAsync device-enumeration to plug into universal. Rejected because it would make runtime discovery auto-provision from births (the unbounded-address-space failure §3.3 explicitly avoids) and still wouldn't cover plain MQTT. The birth cache stays a browse-time + subscribe-time structure, not a discovery source.
Decision: keep SupportsOnlineDiscovery=false (authored-only runtime discovery) and build a bespoke IDriverBrowser (MqttDriverBrowser) with an observation-window IBrowseSession (MqttBrowseSession). Registering the bespoke browser for DriverType="Mqtt" overrides the universal fallback for MQTT (the universal browser resolves bespoke-first — BrowserSessionService §3.1 of that doc).
4.1 MqttDriverBrowser : IDriverBrowser
DriverType => "Mqtt". OpenAsync(configJson, ct):
- Parse the same
MqttDriverOptionsJSON the runtime driver consumes (sharedJsonSerializerOptions), but connect with a distinct client-id suffix ({clientId}-browse-{guid8}) so the browse session never collides with the running driver's MQTT session / clean-session state. - Connect an MQTTnet client under a bounded connect budget (clamp like OpcUaClient's
PerEndpointConnectTimeout, 5–30 s). - Subscribe to the discovery filter:
spBv1.0/{groupId}/#(Sparkplug) or#/{topicPrefix}#(plain). - Kick off the observation window immediately — passively. The session only listens; opening it publishes nothing. This keeps browse read-only by default, matching the house browse pattern (driver-browsers design §6: "browse is read-only and doesn't mutate config or driver state"), and avoids rebirth-storming every edge node in the group whenever an operator merely opens a picker — on a large plant an auto-rebirth on session open would stampede the whole group's edge nodes into re-publishing full births.
- Return an
MqttBrowseSessionwhose internal tree fills in over the window and keeps filling while the session lives. BecauseIBrowseSessionmethods are async and the session lands in the AdminUIBrowseSessionRegistry(TTL-reaped, per-call 20 s timeout), the observation cost is amortised across the user's clicks in the picker.
Explicit "Request rebirth" — the one operator-initiated write (Sparkplug mode only). The picker carries a "Request rebirth" button, enabled in Sparkplug mode and scoped to the currently selected group or edge node. Clicking it invokes MqttBrowseSession.RequestRebirthAsync(scope) — an MQTT-specific session method beyond the base IBrowseSession surface, dispatched by the picker through BrowserSessionService for MQTT sessions — which publishes the rebirth NCMD (Node Control/Rebirth = true, via the same RebirthRequester encode path) to the selected edge node, or to each observed edge node under the selected group. That converts the passive wait into a near-synchronous enumeration precisely when the operator asks for it. This is the design's single deliberate deviation from browse-is-read-only: it is operator-initiated, scoped, gated by the same DriverOperator policy that gates the picker, and logged (Info, with the target scope) — it is never fired automatically by OpenAsync, RootAsync, or expansion.
4.2 MqttBrowseSession : IBrowseSession
Serves from an accumulating observed tree (thread-safe; the MQTT receive handler mutates it, browse calls read it). Refreshes LastUsedUtc on each call.
RootAsync— Sparkplug: observed groups (or edge nodes under the configured group). Plain: top-level topic segments.BrowseNode { Kind=Folder, HasChildrenHint=true }. If the window hasn't yielded yet, return what's accumulated (possibly empty) and let the user re-expand — or, in Sparkplug mode, click the explicit "Request rebirth" button (§4.1) to populate fast.RootAsyncitself never publishes; browse calls are pure reads of the accumulated tree.ExpandAsync(nodeId)— one level down the accumulated tree. Sparkplug: Group → EdgeNode → Device → Metric. Plain: next topic segment (split on/). Metrics / leaf-topics areKind=Leaf.AttributesAsync(nodeId)— Sparkplug: the metric'sAttributeInfo { Name, DriverDataType (from birth), IsArray, SecurityClass }— self-describing, so the picker side-panel is populated (unlike the OPC UA browser which returns empty). Plain: a single synthetic attribute carrying the leaf topic's inferred type + last-seen payload snippet.RequestRebirthAsync(scope)— Sparkplug only, beyond the baseIBrowseSessionsurface (§4.1): the single session member that publishes, and only ever on explicit operator click.DisposeAsync— disconnect the MQTTnet client, best-effort (registry reaper may race), same pattern asOpcUaClientBrowseSession.
4.3 Picked node → TagConfig
On commit the picker projects the selected BrowseNode/AttributeInfo into the §5 TagConfig JSON:
- Sparkplug: the
NodeIdencodesgroup/node/device:metric→ split intogroupId/edgeNodeId/deviceId/metricName;dataTypefromAttributeInfo. - Plain: the
NodeIdis the topic path →topic+ a defaultjsonPath($, operator-editable) + inferreddataType.
4.4 Addressing the passive/time-bounded wrinkle explicitly
- Picker shows a live "listening… N nodes/topics discovered" affordance so the operator understands coverage is accumulating, not instantaneous.
- Sparkplug mode offers the explicit "Request rebirth" button (§4.1) — operator-initiated, scoped to the selected group/node — to convert the passive wait into a fast near-complete enumeration on demand. The window itself stays passive; if the operator doesn't click, coverage accumulates from natural traffic only.
- The browse session stays alive (registry TTL) so re-expanding later reflects newly observed topics without reconnecting.
- Manual entry of a topic/metric via the typed editor (§6) is always available as an escape hatch for a tag that's silent during the window.
4.5 Registration
// EndpointRouteBuilderExtensions.cs — alongside the existing two (near line 49-50)
services.AddSingleton<IDriverBrowser, OpcUaClientDriverBrowser>();
services.AddSingleton<IDriverBrowser, GalaxyDriverBrowser>();
services.AddSingleton<IDriverBrowser, MqttDriverBrowser>(); // NEW — overrides universal fallback for "Mqtt"
5. TagConfig + driver-config JSON
Two layers, mirroring every other driver: driver options (broker connection, one per driver instance, parsed by the factory) and per-tag config (TagConfig JSON authored on the /uns Equipment → Tags tab; the driver's FullName is the driver-side reference the router resolves).
5.1 Driver options — MqttDriverOptions (.Contracts)
{
"host": "10.100.0.35",
"port": 8883,
"clientId": "otopcua-uns-1",
"useTls": true,
"allowUntrustedServerCertificate": false, // dev/on-prem only
"caCertificatePath": null, // PEM CA pin; null ⇒ OS trust store
"username": "otopcua",
"password": "", // supply via env, NEVER commit
"protocolVersion": "V500", // "V311" | "V500"
"cleanSession": true,
"keepAliveSeconds": 30,
"connectTimeoutSeconds": 15, // bounded connect deadline (§8)
"reconnectMinBackoffSeconds": 1,
"reconnectMaxBackoffSeconds": 30,
"mode": "SparkplugB", // "Plain" | "SparkplugB"
// ---- Sparkplug-only (null in Plain mode) ----
"sparkplug": {
"groupId": "Plant1", // subscribe spBv1.0/Plant1/#
"hostId": "otopcua-host-1", // spBv1.0/STATE/{hostId} identity (v3.0)
"actAsPrimaryHost": false, // at most ONE primary host per host-id per broker
"requestRebirthOnGap": true,
"birthObservationWindowSeconds": 15 // browse/discovery: how long to collect births
},
// ---- Plain-only (null in SparkplugB mode) ----
"plain": {
"topicPrefix": "factory/",
"defaultQos": 1
}
}
5.2 Per-tag — Sparkplug
{
"FullName": "Plant1/Line3EdgeNode/Filler1:Temperature/degC", // PascalCase key (composer/walker contract)
"groupId": "Plant1",
"edgeNodeId": "Line3EdgeNode",
"deviceId": "Filler1", // omit/null for node-level metrics
"metricName": "Temperature/degC", // STABLE identity across rebirths — bind by this, not alias
"dataType": "Float" // from DBIRTH; explicit for safety
}
FullName convention: "{group}/{node}[/{device}]:{metric}". The driver parses FullName (or the descriptor fields) once, caches it, and resolves inbound metrics by (group,node,device,metricName).
5.3 Per-tag — Plain MQTT
{
"FullName": "factory/line3/oven/temp#$.value", // topic#jsonPath — stable per-tag key
"topic": "factory/line3/oven/temp",
"payloadFormat": "Json", // "Json" | "Raw" | "Scalar"
"jsonPath": "$.value", // JSONPath into payload (Json only)
"dataType": "Double", // explicit — inference is the fallback
"qos": 1,
"retainSeed": true // seed last-value from retained msg on subscribe
}
FullName convention: "{topic}#{jsonPath}" (Json) or just "{topic}" (Raw/Scalar). MqttEquipmentTagParser.TryParse(reference, out def) follows ModbusEquipmentTagParser: a leading { marks a TagConfig blob; parse it, use strict enum reads (TagConfigJson.TryReadEnumStrict) so a typo'd payloadFormat/dataType rejects the tag (→ BadNodeIdUnknown) rather than silently defaulting to a wrong type; the def's Name = the reference string so published values key the forward router. Add an Inspect(reference) deploy-time warning pass (like Modbus) for present-but-invalid enums / unparseable blobs.
Historization: the server-side isHistorized / historianTagname keys are owned by the TagModal-merge seam (TagHistorizeConfig) and survive load→save of the typed model as preserved unknown keys — the driver does not model them (identical to OpcUaClientTagConfigModel).
6. Typed editor + validator
6.1 MqttTagConfigModel (.AdminUI/Uns/TagEditors/)
Thin typed working model over a preserved JsonObject key bag, per the OpcUaClientTagConfigModel template: FromJson(json) / ToJson() / Validate(), preserving unknown keys across a load→save (so history keys + fields this editor doesn't expose survive). It carries both mode shapes — a Mode field (Plain|SparkplugB, defaulted from which fields are present) selects which sub-shape the editor renders + validates.
FromJsonreads the descriptor fields for the active mode plus the always-presentFullName(PascalCase, case-sensitive — the composer/artifact-decoder/walker read it via case-sensitiveTryGetProperty("FullName", …)).ToJsonwritesFullNamePascalCase and re-derives it from the descriptor fields ({group}/{node}[/{device}]:{metric}or{topic}#{jsonPath}) so a hand-edited descriptor stays consistent with the routed key.Validate():- Sparkplug:
groupId,edgeNodeId,metricNamerequired;deviceIdoptional;dataTypemust be a known Sparkplug type. - Plain:
topicrequired and non-wildcard (a subscription topic for a tag must be concrete — reject+/#);payloadFormat=Json⇒jsonPathrequired;dataTypea knownDriverDataType.
- Sparkplug:
6.2 MqttTagConfigEditor.razor
Copy the Modbus editor template (Components/Shared/Uns/TagEditors/): a razor shell over the pure model, mode-switch at top, per-mode field group below, client-side validation via Validate(). Register:
// TagConfigEditorMap.cs
["Mqtt"] = typeof(Components.Shared.Uns.TagEditors.MqttTagConfigEditor),
// TagConfigValidator.cs
["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate(),
6.3 The JsonStringEnumConverter enum-serialization trap
Every JSON seam that (de)serializes MqttDriverOptions / tag configs — factory, probe, browser, and the AdminUI driver-config page + probe DTO — MUST use a JsonSerializerOptions that includes new JsonStringEnumConverter() (plus PropertyNameCaseInsensitive=true, UnmappedMemberHandling=Skip). The repo's SYSTEMIC bug (memory: "Driver enum-serialization bug"): AdminUI pages that serialize enums numerically while factory DTOs are string-typed fault the driver on any enum field. MqttMode, MqttPayloadFormat, protocolVersion are all enums — the AdminUI Mqtt driver page and its probe must serialize them as strings (mirror the OpcUaClient page). JsonStringEnumConverter also accepts numeric ordinals, so existing numeric configs still load. Keep one shared JsonSerializerOptions definition referenced by factory + probe + browser (as OpcUaClient does).
7. Factory + registration
MqttDriverFactoryExtensions (mirror OpcUaClientDriverFactoryExtensions):
public static class MqttDriverFactoryExtensions
{
public const string DriverTypeName = "Mqtt";
private static readonly JsonSerializerOptions JsonOptions = new()
{
PropertyNameCaseInsensitive = true,
UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip,
Converters = { new JsonStringEnumConverter() },
};
public static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory = null)
{
ArgumentNullException.ThrowIfNull(registry);
registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory));
}
public static MqttDriver CreateInstance(string driverInstanceId, string driverConfigJson, ILoggerFactory? lf = null)
{
var options = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, JsonOptions)
?? throw new InvalidOperationException($"Mqtt driver config for '{driverInstanceId}' deserialised to null");
return new MqttDriver(options, driverInstanceId, lf?.CreateLogger<MqttDriver>());
}
}
Wire in the two host bootstrap sites:
// Host/Drivers/DriverFactoryBootstrap.cs — factory registration (near line 133)
Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory);
// … and probe registration in AddOtOpcUaDriverProbes (near line 114)
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, Driver.Mqtt.MqttDriverProbe>());
MqttDriverProbe : IDriverProbe (DriverType => "Mqtt"): parse the config with the shared JsonSerializerOptions, open an MQTTnet CONNECT to host:port under the timeout, return green + latency on CONNACK-accepted, or a targeted error (connection refused / TLS failure / auth failure / timeout). Analogous to ModbusDriverProbe's TCP+FC03 handshake — CONNACK is the MQTT "device is answering" proof. Probes are admin-pinned (the admin-operations singleton dispatches by DriverType) — the TryAddEnumerable line must be on admin nodes (it is, via AddOtOpcUaDriverProbes in Program.cs's hasAdmin block).
8. Resilience / timeout / security
- Bounded connect.
InitializeAsyncconnects underconnectTimeoutSeconds(default 15) via a linked CTS — no unbounded wait on a dead broker (heed the S7/FOCAS read-timeout lessons: an async network call that ignores its deadline wedges the whole driver). - Reconnect with backoff — hand-rolled, committed for P1. MQTTnet v5 dropped the v4
ManagedMqttClient, so there is no library-managed reconnect to lean on:MqttConnectionowns a hand-rolled reconnect loop. Requirements: exponential backoffreconnectMinBackoffSeconds → reconnectMaxBackoffSeconds; re-subscribe on every reconnect (never assume broker-held subscription state — withcleanSession=truesubscriptions don't survive the session, and even with a persistent session treat re-subscribe as idempotent insurance); and session-state handling (honourcleanSession/session-expiry semantics; re-establish the Sparkplug STATE birth when acting as primary host). On disconnect:GetHealth→ Reconnecting; on reconnect: re-subscribe, and in Sparkplug mode request rebirth to recover metadata (late-join, §3.6). Nodes go Bad-quality while disconnected via the standard fan-out. - Subscribe deadline.
SubscribeAsynccompletes under a bounded deadline; SUBACK failure surfaces as a per-reference Bad, not a hang. - No poll loop — the receive handler must be non-blocking (offload heavy decode off the MQTT callback thread; never block the MQTTnet dispatcher).
- Broker security — enforce, never ship public-broker defaults. Default
useTls=true; real auth (username/password or client-cert).allowUntrustedServerCertificate+caCertificatePathmirror the ServerHistorian TLS knobs (dev/on-prem escape hatch, off by default). Never put Sparkplug production-shaped data on public brokers (test.mosquitto.orgetc. — manual smoke only). Credentials come from env (passwordblank in committed JSON), same posture asServerHistorian__ApiKey. - Primary-host guard. At most one primary-host client per
hostIdper broker (Sparkplug MUST).actAsPrimaryHost=trueon two OtOpcUa instances sharing ahostId(e.g. a redundancy pair) is a misconfiguration — the redundancy pair should use distinct host-ids (mirrors the distinctMxAccess.ClientNameredundancy rule).
9. Test fixtures
Follow the driver-fixture pattern: tests/Drivers/.../Docker/docker-compose.yml with the project: lmxopcua label, deployed to 10.100.0.35 via lmxopcua-fix sync/up (CLAUDE.md Docker Workflow). Repo files are source-of-truth; /opt/otopcua-mqtt/ is the mirrored deployment.
- Broker: Eclipse Mosquitto (
eclipse-mosquitto, tiny, carries both plain + Sparkplug unchanged) as the default fixture; EMQX (emqx/emqx, Apache-2.0 CE) as the alternative when a dashboard / built-in Sparkplug tooling helps. Configure with auth + TLS so the fixtures exercise the real security path (not anonymous). - Sparkplug edge-node simulator: prefer a project-owned C# simulator using the same MQTTnet + Tahu-proto path the driver uses — validates encode/decode symmetrically and gives full control of rebirth/seq-gap/death scenarios (the §3.6 test matrix). Alternatives: Eclipse Tahu reference (Java/Python) edge-node script; Bevywise MQTT/IoT Simulator (purpose-built Sparkplug B). The simulator must publish NBIRTH/DBIRTH → periodic NDATA/DDATA, honour a rebirth NCMD, and be able to inject a seq gap / alias reuse / N-D-DEATH on demand.
- Plain-MQTT fixture: a tiny publisher container (
mosquitto_publoop or apaho-mqttscript) emitting JSON on a handful of topics withretain=trueso the last-value/read path and the#-observation browse are testable. - Env-gated live suite:
Driver.Mqtt.IntegrationTests, categoryLiveIntegration(à la HistorianGateway), skips cleanly whenMQTT_FIXTURE_ENDPOINTis unset → macOS-offline-safe. Covers: connect/TLS/auth; plain subscribe+retained-read; Sparkplug birth→data→alias-resolve→OnDataChange; rebirth recovery; death→STALE; the browser observation window (passive birth-driven tree accumulation — assert no NCMD is published by open/browse calls — plus the explicitRequestRebirthAsyncenumeration, scoped to node vs. group). - Live-verify (docker-dev): author an Mqtt driver + tag via the
/unspicker on:9200, deploy, confirm the node carries live broker values and the typed editor renders both modes (the usual live-/rundiscipline — Razor binding bugs pass unit tests).
Add the fixture endpoint to CLAUDE.md's endpoint list once landed (e.g. 10.100.0.35:1883/8883, MQTT_FIXTURE_ENDPOINT).
10. Phasing + effort
Order: plain MQTT first, then Sparkplug B, then write. Plain MQTT is the dependency-light slice that stands up the whole skeleton and validates MQTTnet-v5-on-net10 before any Sparkplug/proto work — front-loading the reusable plumbing and de-risking the library decision.
| Phase | Scope | Rel. effort |
|---|---|---|
| P1 — Plain MQTT | 3 projects; MQTTnet-5 connect/TLS/auth + the hand-rolled reconnect loop (backoff + resubscribe-on-reconnect + session-state handling, §8 — v5 has no ManagedMqttClient) in MqttConnection; ISubscribable (topic subscribe → OnDataChange); IReadable last-value (retained seed); authored-tag ITagDiscovery (Once); IHostConnectivityProbe; MqttDriverProbe; factory + host wiring; #-observation browser (MqttDriverBrowser/MqttBrowseSession); typed editor + validator + AdminUI driver page; Mosquitto + JSON-publisher fixtures + env-gated suite. Validate MQTTnet-5/net10 here. |
M |
| P2 — Sparkplug B ingest | Vendored Tahu proto + Google.Protobuf codegen; SparkplugCodec decode; spBv1.0/{group}/# subscribe; N/DBIRTH → BirthCache/AliasTable; bind-by-name + seq-gap detect + rebirth NCMD; N/DDEATH → STALE; STATE/primary-host option; ITagDiscovery UntilStable + IRediscoverable; passive birth-driven browser tree + AttributesAsync + the operator "Request rebirth" picker action (§4.1); Sparkplug datatype map; C# edge-node simulator fixture + §3.6 test matrix. |
L |
| P3 — Write-through (optional) | IWritable: Sparkplug NCMD/DCMD + plain publish; optimistic-Good + self-correct on echo; WriteIdempotent respect (default off for non-idempotent Sparkplug commands); write-topic config. |
M |
Top risks: (1) Sparkplug state-machine correctness (aliases + rebirth + seq — §3.6; bind-by-name is the mitigation, test against a controllable simulator); (2) library/dependency friction — SparkplugNet's MQTTnet-4.x transitive pin ✕ this repo's fragile central pinning (transitive pinning off), mitigated by hand-rolling the Tahu proto over one MQTTnet-5 client (§2.1). Secondary: passive-browse coverage gaps (the explicit "Request rebirth" action + manual entry — the window itself never publishes), unbounded address space (authored-only runtime discovery), write-has-no-ack (accepted, optimistic-Good like Galaxy), broker security (TLS + real auth enforced).