docs: driver-expansion program — 8 research reports + 7 design docs, parallel-reviewed
Adds the driver-expansion program design (umbrella: universal Discover-backed browser + MTConnect, MQTT/Sparkplug B, BACnet/IP, SQL poll, Omron, Modbus RTU; MELSEC deferred) plus the per-driver research reports. All docs went through a 7-agent parallel review against the codebase before this commit. Highlights fixed in review: - universal browser: FOCAS FixedTree fills post-connect -> UntilStable settle + FixedTree.Enabled patch; MQTT reconciled to bespoke (was contradicting the program doc's SupportsOnlineDiscovery=false verdict) - modbus-rtu: SerialPort.ReadTimeout doesn't bound async BaseStream reads -> linked-CTS per-op deadline (R2-01 class); BCL enum reuse would leak System.IO.Ports into Contracts - bacnet: DiscoveryRediscoverPolicy enum name; UDP 47808 contention; live suite rewritten around unicast Who-Is + BBMD (broadcast doesn't cross VMs) - sql-poll: real tier registration via DriverFactoryRegistry.Register; blackhole gate must not docker-pause the shared central SQL Server - mqtt: Sparkplug v3.0 STATE topic form; first-in-repo proto codegen noted - omron: host hardcodes isIdempotent:false today (retry seam unshipped); v1 scopes UDTs to dotted-leaf access - mtconnect: SecurityClassification.ViewOnly; factory ParseEnum<T> pattern - program doc: both valid enum-serialization patterns; IRediscoverable is change-signal-gated; RTU P2 adds System.IO.Ports; label is host-side
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,292 @@
|
||||
# MQTT / Sparkplug B Driver — Research & Design
|
||||
|
||||
**Status:** Research / design proposal (not yet implemented)
|
||||
**Author:** research sweep, 2026-07-15
|
||||
**Scope:** A standard **Equipment-kind** driver (same shape as Modbus/S7/AbCip/TwinCAT/FOCAS/OpcUaClient) that ingests **plain MQTT** and **Sparkplug B** data into the OtOpcUa OPC UA / UNS address space.
|
||||
|
||||
This is a **subscribe-first** driver: the broker pushes data, we do not poll. Because OtOpcUa is a Unified-Namespace product, Sparkplug B alignment (the de-facto MQTT UNS payload standard) is the headline capability, with plain-MQTT/JSON as the broad-compatibility fallback.
|
||||
|
||||
---
|
||||
|
||||
## 1. Protocol summary + .NET library options
|
||||
|
||||
### 1.1 Plain MQTT
|
||||
|
||||
MQTT is a lightweight pub/sub transport: a client connects to a **broker**, subscribes to **topic filters** (with `+` single-level and `#` multi-level wildcards), and receives **retained** and live messages. Payloads are opaque byte arrays — by convention JSON, a scalar string/number, or a binary blob. There is no schema, no data-type metadata, and no built-in discovery: a consumer only knows a topic exists once a message arrives on it. "Last value" is available only if the publisher set the **retain** flag (the broker keeps the last retained message per topic and replays it to new subscribers).
|
||||
|
||||
Relevant knobs: QoS 0/1/2, `retain`, Last-Will-and-Testament (LWT), clean-session vs. persistent-session, TLS, username/password or client-cert auth, MQTT 3.1.1 vs 5.0.
|
||||
|
||||
### 1.2 Sparkplug B (Eclipse Tahu / Eclipse Sparkplug, spec v3.0)
|
||||
|
||||
Sparkplug B is an **open specification layered on MQTT** that adds the missing pieces for industrial data: a **mandated topic namespace**, a **protobuf payload schema**, **stateful session management**, and **auto-discovery via birth certificates**. It is the dominant "UNS over MQTT" standard.
|
||||
|
||||
**Topic namespace:** `spBv1.0/{group_id}/{message_type}/{edge_node_id}[/{device_id}]`
|
||||
- `group_id` — logical grouping (e.g. a site/area).
|
||||
- `edge_node_id` — an MQTT Edge-of-Network (EoN) node.
|
||||
- `device_id` — an optional physical device attached to the edge node.
|
||||
|
||||
**Message types:**
|
||||
|
||||
| Type | Meaning | Direction |
|
||||
|---|---|---|
|
||||
| `NBIRTH` | Edge-node **birth certificate** — enumerates every node-level metric with name, alias, datatype, and initial value; carries `bdSeq` | edge → host |
|
||||
| `DBIRTH` | Device **birth certificate** — enumerates a device's metrics (name/alias/datatype/value) | edge → host |
|
||||
| `NDATA` / `DDATA` | Node/device **data** — metric changes, usually referenced by **alias** only (no name) | edge → host |
|
||||
| `NDEATH` / `DDEATH` | Node/device **death** — LWT-driven; marks metrics STALE/uncertain | edge → host |
|
||||
| `NCMD` / `DCMD` | **Command / write** to a node or device metric (e.g. setpoint, or `Node Control/Rebirth`) | host → edge |
|
||||
| `STATE` | **Primary-host** application online/offline status (`ONLINE`/`OFFLINE`), retained, LWT-backed | host → edge |
|
||||
|
||||
**Key semantics the driver must honour:**
|
||||
- **Birth-certificate discovery.** DBIRTH/NBIRTH is the *only* place metric **names + datatypes** appear. A consumer must cache the birth to interpret later data. This is exactly what makes Sparkplug **browsable** (§4).
|
||||
- **Aliases.** After birth, NDATA/DDATA typically send only a numeric `alias` + value to save bandwidth. The consumer must maintain an **alias → (name, datatype)** map per edge-node/device, rebuilt on every (re)birth.
|
||||
- **Sequence numbers.** Every Sparkplug payload carries a `seq` (0–255, wraps) for gap detection; NBIRTH resets the sequence. `bdSeq` (birth/death sequence) in NBIRTH/NDEATH ties a death to its birth.
|
||||
- **Rebirth.** If the host sees a gap, an unknown alias, or connects late (missed the birth), it issues an `NCMD` writing boolean `Node Control/Rebirth = true`; the edge node re-publishes NBIRTH + all DBIRTHs. This is the recovery primitive that lets a late-joining consumer recover full metadata.
|
||||
- **Primary host / STATE.** A "primary host application" publishes a retained `STATE` message (with LWT set to `OFFLINE`, QoS 1, retain=true) so edge nodes know a trusted consumer is online. There **MUST be at most one** primary host client per host-id on a broker. OtOpcUa can consume as a non-primary application, or opt in as primary host (needed for guaranteed store-and-forward semantics on some edge devices).
|
||||
- **Metric datatypes** (protobuf `DataType` enum): `Int8/16/32/64`, `UInt8/16/32/64`, `Float`, `Double`, `Boolean`, `String`, `DateTime`, `Text`, `UUID`, `Bytes`, `File`, `DataSet`, `Template`, plus array variants (`Int8Array`, …). Each metric = `{name, alias, timestamp, datatype, value, is_historical, is_transient, is_null}`.
|
||||
|
||||
Sources: [Eclipse Sparkplug spec (PDF, v2.2 — namespace/STATE rules carry into v3.0)](https://sparkplug.eclipse.org/specification/version/2.2/documents/sparkplug-specification-2.2.pdf) · [Eclipse Sparkplug normative statements](https://github.com/eclipse-sparkplug/sparkplug/blob/master/docs/normative_statements.md) · [Tahu `sparkplug_b.proto`](https://raw.githubusercontent.com/eclipse/tahu/master/sparkplug_b/sparkplug_b.proto) · [HiveMQ: Sparkplug session state](https://www.hivemq.com/blog/understanding-mqtt-topic-namespace-iiot/) · [Steve's Internet Guide: Sparkplug payloads/messages](http://www.steves-internet-guide.com/sparkplug-payloads-and-messages/) · [EMQX Sparkplug docs](https://docs.emqx.com/en/emqx/latest/data-integration/sparkplug.html)
|
||||
|
||||
### 1.3 .NET library options
|
||||
|
||||
| Library | License | Maturity | .NET 10 | Notes |
|
||||
|---|---|---|---|---|
|
||||
| **MQTTnet** | **MIT** | **High.** Now a **.NET Foundation** project hosted under **`dotnet/MQTTnet`**; the standard .NET MQTT client+server. Millions of NuGet downloads, active. v5.x current. | **Yes** — v5 explicitly added `dotnet10` target. | Client + broker, MQTT 3.1.1 & 5.0, TLS, WebSocket. This is the transport layer for **both** modes. |
|
||||
| **SparkplugNet** | **MIT** | **Moderate.** Single-maintainer (SeppPenner), regular releases; latest **1.3.10 (2024-07-02)**. Supports Sparkplug **v3.0 / spBv1.0** (spAv1.0 obsolete). Provides `Application`, `Node`, `Device` base classes; the **Application** role receives N/DBIRTH + N/DDATA and can publish NCMD/DCMD. Lower adoption than MQTTnet. | **Targets net8.0/net9.0** today (plus "latest/LTS Core & Framework"). **No explicit net10 TFM yet** — will run on .NET 10 via net9.0 compat, but confirm at build time; may need an upstream TFM bump or a fork pin. | Wraps MQTTnet + `protobuf-net` internally. Depends on `MQTTnet >= 4.3.x` — watch for a **version clash** with a directly-referenced MQTTnet 5.x (see risk §6). |
|
||||
| **MQTTnet + hand-decoded Tahu protobuf** | MIT (+ Tahu, EPL/Apache) | Build-it-yourself | Yes | Reference the Tahu `sparkplug_b.proto`, generate C# with `protobuf-net` or `Google.Protobuf`, decode payloads ourselves on top of an MQTTnet subscription. Maximum control (aliases, seq, rebirth logic exactly as we want), no third-party Sparkplug abstraction, but we own all the state-machine code. |
|
||||
|
||||
**Recommendation:** Use **MQTTnet (v5, MIT, .NET Foundation, net10-ready)** as the transport for both modes. For Sparkplug decoding, **start by evaluating SparkplugNet** to save the birth/alias/rebirth plumbing, but treat the **"MQTTnet + Tahu-proto hand-decode"** path as the fallback if SparkplugNet's net10 support or its internal MQTTnet-4.x pin proves awkward. The Sparkplug protobuf schema is small and stable, so hand-decoding is a bounded, well-understood effort.
|
||||
|
||||
Sources: [MQTTnet on NuGet (v5.2.0)](https://www.nuget.org/packages/MQTTnet/) · [dotnet/MQTTnet (GitHub)](https://github.com/dotnet/MQTTnet) · [MQTTnet — .NET Foundation](https://old.dotnetfoundation.org/projects/mqttnet) · [SparkplugNet on NuGet](https://www.nuget.org/packages/SparkplugNet) · [SparkplugNet (GitHub, SeppPenner)](https://github.com/SeppPenner/SparkplugNet)
|
||||
|
||||
---
|
||||
|
||||
## 2. Capability mapping
|
||||
|
||||
The driver implements the composable `IDriver` capability interfaces from `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/`. Recommended set: `IDriver` + `ITagDiscovery` + `ISubscribable` + `IReadable` + `IHostConnectivityProbe` + `IRediscoverable`, and **optionally** `IWritable` (phase 2). Notably **not** a poller — unlike Modbus/S7 it does not use `PollGroupEngine`; it keeps a live broker connection and raises `OnDataChange` from the MQTT receive callback (closest existing analog: the native-push side of the OpcUaClient driver).
|
||||
|
||||
### 2.1 Subscribe (primary — `ISubscribable`)
|
||||
|
||||
The core of the driver. On `InitializeAsync`, connect to the broker (MQTTnet), set LWT/clean-session, and subscribe to the configured topic filters:
|
||||
- **Plain mode:** subscribe to each authored tag's topic (or a shared wildcard) and to nothing else.
|
||||
- **Sparkplug mode:** subscribe to `spBv1.0/{group_id}/#` (and `STATE/#` if primary host). Maintain per-edge-node/device **alias→metric** tables from N/DBIRTH; on N/DDATA, resolve each metric and raise `OnDataChange` with the authored tag's `FullReference`. On NDEATH/DDEATH, publish STALE/Bad-quality snapshots for that node's metrics. On a missed birth / unknown alias / seq gap, issue a **rebirth NCMD**.
|
||||
|
||||
`SubscribeAsync(fullReferences, publishingInterval, …)` maps requested `FullReference`s onto the live receive stream. MQTT/Sparkplug are inherently event-driven, so `publishingInterval` is advisory (used only for optional server-side coalescing/deadband, mirroring Modbus's `ShouldPublish`). The driver holds the broker connection for its whole lifetime; `OnDataChange` fires from the MQTT message handler. Emit an initial value from the retained message / last birth value on subscribe (OPC UA initial-data convention).
|
||||
|
||||
### 2.2 Discover (`ITagDiscovery`)
|
||||
|
||||
- **Sparkplug mode (rich):** DBIRTH/NBIRTH **enumerates** every metric with name + datatype → map directly onto Equipment + Tags. Natural UNS shape: `group_id` → Area/Line, `edge_node_id`/`device_id` → **Equipment**, each **metric → Tag**. Because births arrive asynchronously after connect, use `RediscoverPolicy = UntilStable` (like FOCAS) — keep discovering until the observed birth set stops growing, and implement `IRediscoverable` so a *new* DBIRTH (new device joins, or a rebirth with changed metrics) triggers an address-space rebuild.
|
||||
- **Plain mode (passive):** no schema. Two options: (a) **authored-only** — the operator declares tags (topic + JSON path) up front, `DiscoverAsync` returns exactly those (like Modbus's pre-declared `Tags`, `RediscoverPolicy = Once`); or (b) **observe-and-suggest** — subscribe `#` for a bounded window, build the observed topic tree, surface it in the browser (§4) for the operator to pick. Runtime discovery stays authored-only; wildcard observation is a **browse-time** concern, not an auto-provisioning one (avoids unbounded address spaces from chatty brokers).
|
||||
|
||||
### 2.3 Read (`IReadable`)
|
||||
|
||||
MQTT has no request/response read. `ReadAsync` returns the **last-value cache**: the last message seen on the topic (plain) or the last metric value from data/birth (Sparkplug). Seed it from the broker's **retained** message at connect (plain) and from birth values (Sparkplug). If no value has been observed yet, return `GoodNoData`/uncertain rather than an error. This satisfies OPC UA reads and HistoryRead-at-time without a live round-trip.
|
||||
|
||||
### 2.4 Write (`IWritable`) — **verdict: defer to phase 2, not in v1**
|
||||
|
||||
Writes are possible but semantically heavier and lower-priority for a UNS-ingest product:
|
||||
- **Sparkplug:** publish an **NCMD/DCMD** to `spBv1.0/{group}/NCMD/{node}[/{device}]` with the target metric (by name/alias) + new value. Fire-and-forget at the MQTT layer — there is **no synchronous write ack**; success is only observable when the edge node echoes the change back in the next N/DDATA. This mirrors the Galaxy gateway's fire-and-forget write (Galaxy "can never surface a write failure"), so a write returns `Good` optimistically and the value self-corrects on the next data message. Respect `WriteIdempotent` — Sparkplug commands (rebirth, pulse) are frequently non-idempotent.
|
||||
- **Plain:** publish to a configured command/write topic (often distinct from the read topic) with a formatted payload.
|
||||
|
||||
**Recommendation:** ship v1 **read/subscribe/discover only** (a genuinely useful UNS-ingest driver on its own), add `IWritable` (NCMD/DCMD + plain publish) in phase 2. Rationale: the write path has no ack model, needs careful idempotency/rebirth handling, and most UNS-ingest deployments are read-only consumers.
|
||||
|
||||
### 2.5 Data-type mapping (`DriverDataType`)
|
||||
|
||||
Map Sparkplug metric datatypes / inferred JSON types to `DriverDataType` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverDataType.cs`):
|
||||
|
||||
| Sparkplug `DataType` | `DriverDataType` | OPC UA |
|
||||
|---|---|---|
|
||||
| Int8, Int16 | Int16 | Int16 |
|
||||
| Int32, UInt16 | Int32 / UInt16 | Int32 / UInt16 |
|
||||
| Int64, UInt32 | Int64 / UInt32 | Int64 / UInt32 |
|
||||
| UInt64 | UInt64 | UInt64 |
|
||||
| Float | Float32 | Float |
|
||||
| Double | Double64 → **Float64** | Double |
|
||||
| Boolean | Boolean | Boolean |
|
||||
| String, Text, UUID | String | String |
|
||||
| DateTime | DateTime | DateTime |
|
||||
| Bytes, File | (String/ByteString) | ByteString — phase 2 |
|
||||
| DataSet, Template | **unsupported v1** | skip / raw-JSON string |
|
||||
| `*Array` variants | `IsArray=true` + element type | ValueRank=1 |
|
||||
|
||||
Plain-MQTT/JSON: infer from the JSON token at the configured JSON path (number→Double/Int64, bool→Boolean, string→String), or honour an explicit per-tag `dataType` override (preferred — inference is brittle). `Int8`/`UInt8` widen to `Int16`/`UInt16` (no OPC UA byte-signed distinction needed). Complex `DataSet`/`Template` metrics are out of scope for v1 (flag + skip, or expose the raw JSON as a String tag).
|
||||
|
||||
---
|
||||
|
||||
## 3. TagConfig JSON shape
|
||||
|
||||
Two layers, mirroring every other driver: **driver options** (broker connection, one per driver instance) parsed by the driver factory, and **per-tag `TagConfig.FullName` + config** authored on the `/uns` Equipment → Tags tab.
|
||||
|
||||
### 3.1 Driver options (`MqttDriverOptions`, in `.Contracts`)
|
||||
|
||||
```jsonc
|
||||
{
|
||||
"host": "10.100.0.35",
|
||||
"port": 8883,
|
||||
"clientId": "otopcua-uns-1",
|
||||
"useTls": true,
|
||||
"allowUntrustedServerCertificate": false,
|
||||
"caCertificatePath": null,
|
||||
"username": "otopcua",
|
||||
"password": "", // supply via env, never commit
|
||||
"protocolVersion": "V500", // MQTT 3.1.1 | 5.0
|
||||
"cleanSession": true,
|
||||
"keepAliveSeconds": 30,
|
||||
"reconnectMinBackoffSeconds": 1,
|
||||
"reconnectMaxBackoffSeconds": 30,
|
||||
|
||||
"mode": "SparkplugB", // "Plain" | "SparkplugB"
|
||||
|
||||
// ---- Sparkplug-only ----
|
||||
"sparkplug": {
|
||||
"groupId": "Plant1", // subscribe spBv1.0/Plant1/#
|
||||
"hostId": "otopcua-host-1", // STATE/{hostId} identity
|
||||
"actAsPrimaryHost": false,
|
||||
"requestRebirthOnGap": true,
|
||||
"birthObservationWindowSeconds": 15 // discovery: how long to collect births
|
||||
},
|
||||
|
||||
// ---- Plain-only ----
|
||||
"plain": {
|
||||
"topicPrefix": "factory/",
|
||||
"defaultQos": 1
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3.2 Per-tag config — Sparkplug
|
||||
|
||||
`TagConfig.FullName` is the driver-side full reference the router resolves. For Sparkplug the natural key is the fully-qualified metric path; store the full descriptor in the tag config so the driver can resolve by **name or alias** and survive alias reassignment across rebirths.
|
||||
|
||||
```jsonc
|
||||
// TagConfig for one Sparkplug metric tag
|
||||
{
|
||||
"groupId": "Plant1",
|
||||
"edgeNodeId": "Line3EdgeNode",
|
||||
"deviceId": "Filler1", // omit/null for node-level metrics
|
||||
"metricName": "Temperature/degC", // stable identity across rebirths
|
||||
"dataType": "Float", // from DBIRTH; explicit for safety
|
||||
"isHistorized": false
|
||||
}
|
||||
// FullName convention: "Plant1/Line3EdgeNode/Filler1:Temperature/degC"
|
||||
```
|
||||
|
||||
### 3.3 Per-tag config — Plain MQTT
|
||||
|
||||
```jsonc
|
||||
// TagConfig for one plain-MQTT tag
|
||||
{
|
||||
"topic": "factory/line3/oven/temp",
|
||||
"payloadFormat": "Json", // "Json" | "Raw" | "Scalar"
|
||||
"jsonPath": "$.value", // JSONPath into the payload (Json only)
|
||||
"dataType": "Double", // explicit — inference is a fallback
|
||||
"qos": 1,
|
||||
"retainSeed": true // seed last-value from retained msg
|
||||
}
|
||||
// FullName convention: "factory/line3/oven/temp#$.value"
|
||||
```
|
||||
|
||||
Resolution mirrors Modbus's `EquipmentTagRefResolver<TDef>`: the router hands the driver a `FullReference` (the raw `TagConfig` JSON or a stable string), the driver parses+caches it once, and matches inbound messages/metrics against it.
|
||||
|
||||
---
|
||||
|
||||
## 4. Browseability verdict + browse design
|
||||
|
||||
### Verdict: **BROWSEABLE — both modes, but discovery is passive and time-bounded.**
|
||||
|
||||
Unlike an OPC UA server (synchronous request/response `Browse`) or a Galaxy repository (queryable DB), MQTT/Sparkplug discovery is **observational**: you learn what exists only by *listening* for a while. The browse session therefore **connects to the broker and observes for a bounded window**, accumulating a tree, then answers `ExpandAsync` from that accumulated snapshot. This is the central design wrinkle.
|
||||
|
||||
- **Sparkplug (rich, structured):** subscribe `spBv1.0/{group}/#`, optionally publish a **rebirth NCMD** to force every edge node to re-announce immediately (turns a passive wait into a near-synchronous enumeration), and collect NBIRTH/DBIRTH. Births yield a clean, typed tree: **Group → EdgeNode → Device → Metric**, with datatypes — directly pickable, no guessing.
|
||||
- **Plain (best-effort):** subscribe `#` (or a configured prefix) for the observation window and build the **observed topic tree**, splitting each topic on `/`. Leaves are topics that have carried a payload; the picker shows the last payload + inferred type as `AttributesAsync`. Coverage is only as good as broker traffic during the window (a topic silent during the window is invisible) — clearly a "suggestions" browse, not an authoritative enumeration.
|
||||
|
||||
### `IDriverBrowser` / `IBrowseSession` design
|
||||
|
||||
New project **`ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser`** (parallels `.OpcUaClient.Browser` / `.Galaxy.Browser`), implementing `Commons/Browsing/IDriverBrowser` + `IBrowseSession`, registered in `EndpointRouteBuilderExtensions` alongside the others:
|
||||
|
||||
```csharp
|
||||
// EndpointRouteBuilderExtensions.cs (~line 49)
|
||||
services.AddSingleton<IDriverBrowser, OpcUaClientDriverBrowser>();
|
||||
services.AddSingleton<IDriverBrowser, GalaxyDriverBrowser>();
|
||||
services.AddSingleton<IDriverBrowser, MqttDriverBrowser>(); // NEW
|
||||
```
|
||||
|
||||
**`MqttDriverBrowser.OpenAsync(configJson, ct)`** — connect an MQTTnet client from the same `MqttDriverOptions` JSON the runtime driver consumes (separate client-id suffix so it never collides with the running driver's session), subscribe to the discovery filter, and **kick off the observation window immediately**. Returns an `MqttBrowseSession` whose internal tree fills in over the window (and keeps filling as long as the session lives). Because `IBrowseSession` methods are async and the session is registered in the `BrowseSessionRegistry` (TTL-reaped, like the OPC UA one), the "wait for observation" cost is naturally amortised across the user's clicks in the picker.
|
||||
|
||||
**`MqttBrowseSession`:**
|
||||
- `RootAsync` — Sparkplug: the set of observed **groups** (or edge nodes under the configured group). Plain: top-level topic segments. Returns `BrowseNode { Kind = Folder, HasChildrenHint = true }`. If the window hasn't yielded anything yet, return what's accumulated so far (possibly empty) and let the user re-expand — or briefly await the first births. Optionally trigger a rebirth NCMD on first `RootAsync` in Sparkplug mode to populate fast.
|
||||
- `ExpandAsync(nodeId)` — walk one level down the accumulated tree (EdgeNode→Device→Metric, or next topic segment). Metrics / leaf-topics are `Kind = Leaf`.
|
||||
- `AttributesAsync(nodeId)` — Sparkplug: return the metric's `AttributeInfo { Name, DriverDataType, IsArray, SecurityClass }` from the cached birth. Plain: return the leaf topic's inferred type + last-seen payload snippet as a single synthetic attribute. (Sparkplug leaves are self-describing, so unlike OPC UA the side-panel is populated.)
|
||||
- `DisposeAsync` — disconnect the MQTTnet client; best-effort (registry reaper may race a disconnect), same pattern as `OpcUaClientBrowseSession`.
|
||||
|
||||
**Picked node → TagConfig:** on commit, the picker projects the selected `BrowseNode`/`AttributeInfo` into the §3 TagConfig JSON. Sparkplug: `NodeId` encodes `group/node/device:metric`, split into the descriptor fields. Plain: `NodeId` is the topic path → `topic` + a default `jsonPath` (`$` or operator-edited) + inferred `dataType`.
|
||||
|
||||
**Addressing the passive/time-bounded wrinkle explicitly:**
|
||||
1. Show a live "listening… N nodes/topics discovered" affordance in the picker so the operator understands coverage is accumulating, not instantaneous.
|
||||
2. In Sparkplug mode, **actively force a rebirth** to convert the passive wait into a fast, near-complete enumeration.
|
||||
3. Keep the browse session alive (registry TTL) so re-expanding later reflects newly observed topics without reconnecting.
|
||||
4. Always allow **manual entry** of a topic/metric (the typed tag editor, §7) as an escape hatch when a tag is silent during the window.
|
||||
|
||||
---
|
||||
|
||||
## 5. Test-fixture strategy
|
||||
|
||||
Follow the existing driver-fixture pattern: a `docker-compose.yml` under `tests/Drivers/.../Docker/` with the `project: lmxopcua` label, deployed to the shared Docker host `10.100.0.35` via `lmxopcua-fix sync/up` (see `CLAUDE.md` Docker Workflow).
|
||||
|
||||
- **Broker:** **Eclipse Mosquitto** (`eclipse-mosquitto`, tiny, MIT/EPL) as the default fixture — carries both plain MQTT and Sparkplug B unchanged. Alternative **EMQX** (`emqx/emqx`, Apache-2.0, open-source CE up to 1000 conns) when a management dashboard or built-in Sparkplug tooling is wanted. HiveMQ CE is a third option.
|
||||
- **Sparkplug edge-node simulator:** options, in rough order of convenience —
|
||||
- **Bevywise MQTT/IoT Simulator** — purpose-built Sparkplug B realtime data simulation.
|
||||
- **Eclipse Tahu** reference implementations (Java/Python) — author a small edge-node script that publishes NBIRTH/DBIRTH then periodic NDATA/DDATA; also exercises rebirth (subscribe NCMD `Node Control/Rebirth`). Most faithful to the spec.
|
||||
- A **project-owned C# simulator** using the same MQTTnet + Tahu-proto path the driver uses (kills two birds: validates our encode/decode symmetrically).
|
||||
- **Plain-MQTT fixture:** a tiny publisher container (`mosquitto_pub` loop, or a Python `paho-mqtt` script) emitting JSON on a handful of topics with `retain=true` so the last-value/read path and `#`-observation browse are testable.
|
||||
- **Public test brokers** (for ad-hoc manual smoke only, never CI — unauthenticated, rate-limited, no privacy): `test.mosquitto.org`, `broker.hivemq.com`, `broker.emqx.io`. Do **not** put Sparkplug production-shaped data on public brokers.
|
||||
- **Live-gate pattern:** an env-gated `*.IntegrationTests` suite (à la the HistorianGateway `LiveIntegration` category) that skips cleanly when `MQTT_FIXTURE_ENDPOINT` is unset, so the suite is macOS-offline-safe.
|
||||
|
||||
Sources: [Cedalo: Sparkplug B on Mosquitto](https://www.cedalo.com/blog/mqtt-sparkplug-mosquitto) · [EMQ: MQTT Sparkplug step-by-step](https://www.emqx.com/en/blog/mqtt-sparkplug-in-action-a-step-by-step-tutorial) · [Bevywise Sparkplug simulation](https://www.bevywise.com/blog/sparkplug-b-mqtt-simulation/) · [Ian Craggs: getting started with MQTT + Sparkplug](https://modelbasedtesting.co.uk/2022/01/22/getting-started-with-mqtt-and-sparkplug/)
|
||||
|
||||
---
|
||||
|
||||
## 6. Effort / risk / phasing
|
||||
|
||||
### Recommended order: **plain MQTT first, then Sparkplug B.**
|
||||
|
||||
Even though Sparkplug is the headline UNS capability, plain MQTT is the smaller, dependency-light slice that stands up the whole driver skeleton (project layout, `MqttDriverOptions`, factory/probe registration, `ISubscribable` receive loop, last-value `IReadable`, browser project, typed tag editor, fixtures). Sparkplug then layers the birth/alias/rebirth state machine + protobuf decode onto a proven connection+subscribe substrate. This also front-loads the reusable plumbing and de-risks the library decision (you can validate MQTTnet v5/net10 before committing to SparkplugNet).
|
||||
|
||||
| Phase | Scope | Rel. effort |
|
||||
|---|---|---|
|
||||
| **1 — Plain MQTT** | Projects (`Driver.Mqtt`, `.Contracts`, `.Browser`) + MQTTnet connect/TLS/auth/reconnect; `ISubscribable` (topic subscribe → `OnDataChange`); `IReadable` last-value (retained seed); authored-tag `ITagDiscovery`; `IHostConnectivityProbe`; `#`-observation browser; typed tag editor + validator; Mosquitto + JSON-publisher fixtures. | **M** |
|
||||
| **2 — Sparkplug B ingest** | `spBv1.0/{group}/#` subscribe; Tahu-proto decode (SparkplugNet **or** hand-rolled); N/DBIRTH → discovery (`UntilStable` + `IRediscoverable`); alias tables; seq-gap detect + **rebirth NCMD**; N/DDEATH → STALE; STATE/primary-host option; birth-driven browser + `AttributesAsync`; Sparkplug datatype map; Tahu/Bevywise simulator fixture. | **L** |
|
||||
| **3 — Write-through (optional)** | `IWritable`: Sparkplug NCMD/DCMD + plain publish; optimistic-Good + self-correct on echo; `WriteIdempotent` respect; write-topic config. | **M** |
|
||||
|
||||
### Top risks
|
||||
|
||||
1. **Sparkplug state-machine correctness (aliases + rebirth + seq).** The alias→metric mapping, sequence-gap detection, late-join rebirth, and death→STALE handling are the load-bearing logic and easy to get subtly wrong (e.g. an alias reused across a rebirth with a *different* metric silently mis-routes data to the wrong tag). Mitigation: bind tags by **stable metric name**, treat alias as a per-birth cache only; rebuild the alias table on every (re)birth; test explicitly against rebirth + missed-birth + gap scenarios with a controllable simulator. This is the single biggest correctness risk.
|
||||
|
||||
2. **Library / dependency friction (SparkplugNet ↔ MQTTnet version + net10).** SparkplugNet pins **MQTTnet 4.3.x** and has **no explicit net10 TFM** yet; a direct MQTTnet-5 reference (needed for the plain-mode transport + net10 support) can clash with SparkplugNet's transitive 4.x, and central-package-version pinning in this repo is already known to be fragile (see the Roslyn transitive-pinning memory). Mitigation: prototype the version matrix early; be ready to drop SparkplugNet and hand-decode the Tahu proto over a single MQTTnet-5 client (the proto is small/stable), which also removes the net10 TFM concern entirely.
|
||||
|
||||
Secondary risks: **passive-browse coverage gaps** (silent topics invisible in the observation window — mitigated by rebirth-forcing + manual entry, §4); **unbounded address space** from wildcard auto-provisioning (mitigated by authored-only runtime discovery); **write has no ack** (accepted, modelled as optimistic-Good self-correcting like Galaxy, §2.4); **broker security** (enforce TLS + real auth, never ship public-broker defaults).
|
||||
|
||||
---
|
||||
|
||||
## Appendix — project layout (mirrors existing drivers)
|
||||
|
||||
```
|
||||
src/Drivers/
|
||||
ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ # MqttDriver : IDriver, ITagDiscovery,
|
||||
# ISubscribable, IReadable,
|
||||
# IHostConnectivityProbe, IRediscoverable
|
||||
# (+ IWritable in phase 3)
|
||||
MqttDriver.cs, MqttDriverProbe.cs, MqttDriverFactoryExtensions.cs
|
||||
Sparkplug/ (SparkplugDecoder, AliasTable, BirthCache, RebirthRequester)
|
||||
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ # MqttDriverOptions, MqttTagDefinition,
|
||||
# MqttEquipmentTagParser
|
||||
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ # MqttDriverBrowser, MqttBrowseSession
|
||||
|
||||
src/Server/.../AdminUI/
|
||||
Uns/TagEditors/TagConfigEditorMap.cs # + ["Mqtt"] = MqttTagConfigEditor
|
||||
Uns/TagEditors/TagConfigValidator.cs # + Mqtt validation
|
||||
Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor
|
||||
EndpointRouteBuilderExtensions.cs # + AddSingleton<IDriverBrowser, MqttDriverBrowser>
|
||||
|
||||
tests/Drivers/.../Docker/docker-compose.yml # mosquitto + sparkplug-sim + json-publisher
|
||||
```
|
||||
|
||||
DriverType string: **`"Mqtt"`** (single canonical string across AdminUI page, probe, factory, editor map, browser — heed the Modbus `"ModbusTcp"` vs `"Modbus"` mismatch lesson; pick one and use it everywhere).
|
||||
Reference in New Issue
Block a user