diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs
index 47e93a45..8acb9148 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs
@@ -1,5 +1,6 @@
using System.ComponentModel.DataAnnotations;
using System.Text;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
@@ -82,9 +83,40 @@ public sealed record MqttDriverOptions
[Range(1, int.MaxValue)]
public int ReconnectMaxBackoffSeconds { get; init; } = 30;
+ ///
+ /// Ceiling on an inbound message body, in bytes. A larger message is refused before any
+ /// decode or parse and degrades its own tags to BadDecodingError. Default 1 MiB.
+ ///
+ ///
+ /// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their
+ /// cost is paid by every subscription, not just the offending topic — without a bound,
+ /// one publisher shipping a multi-megabyte body stalls the whole driver's delivery. "Unbounded"
+ /// is deliberately not offered; a non-positive value falls back to the driver's own 1 MiB
+ /// default. The literal is duplicated from MqttSubscriptionManager.DefaultMaxPayloadBytes
+ /// because this .Contracts assembly is referenced by the driver assembly and
+ /// cannot reference it back; the manager's constructor is the single enforcement point.
+ ///
+ [Range(1, int.MaxValue)]
+ public int MaxPayloadBytes { get; init; } = 1024 * 1024;
+
/// Selects the ingest shape — Plain topics or Sparkplug B.
public MqttMode Mode { get; init; } = MqttMode.Plain;
+ ///
+ /// The cluster's authored raw MQTT tags, as delivered by the deploy artifact: each carries the
+ /// tag's RawPath (its v3 identity and driver wire reference) plus its TagConfig
+ /// blob. maps each through at
+ /// initialize and re-registers the set wholesale on every reinitialize, so a redeploy
+ /// that drops a tag stops feeding it.
+ ///
+ ///
+ /// This is also the only source of the driver's discoverable node set — plain MQTT has
+ /// no browsable address space, and a tag that is not authored here is never materialized no
+ /// matter how much traffic its topic carries. Mirrors ModbusDriverOptions.RawTags /
+ /// FocasDriverOptions.RawTags.
+ ///
+ public IReadOnlyList RawTags { get; init; } = [];
+
///
/// Sparkplug-only settings. Populated when is
/// ; null in Plain mode.
@@ -127,9 +159,13 @@ public sealed record MqttDriverOptions
builder.Append(", ConnectTimeoutSeconds = ").Append(ConnectTimeoutSeconds);
builder.Append(", ReconnectMinBackoffSeconds = ").Append(ReconnectMinBackoffSeconds);
builder.Append(", ReconnectMaxBackoffSeconds = ").Append(ReconnectMaxBackoffSeconds);
+ builder.Append(", MaxPayloadBytes = ").Append(MaxPayloadBytes);
builder.Append(", Mode = ").Append(Mode);
builder.Append(", Sparkplug = ").Append(Sparkplug);
builder.Append(", Plain = ").Append(Plain);
+ // Count only: a deployment routinely authors thousands of raw tags and each carries a full
+ // TagConfig blob, so printing the list would turn any log of this DTO into a config dump.
+ builder.Append(", RawTags = ").Append(RawTags.Count).Append(" tag(s)");
return true;
}
}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
new file mode 100644
index 00000000..5481f1cd
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
@@ -0,0 +1,732 @@
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
+
+///
+/// The MQTT / Sparkplug B driver shell: composes (broker session +
+/// reconnect supervisor), (authored table, topic routing,
+/// value extraction) and the manager's (the push→poll bridge) into
+/// the driver capability set the Core consumes.
+///
+///
+///
+/// Authored-only discovery. Plain MQTT has no browsable address space — a broker will
+/// happily carry thousands of topics this deployment has nothing to do with.
+/// therefore replays exactly the raw tags the deploy artifact
+/// authored () and nothing else, and
+/// stays so the
+/// universal browser never treats broker traffic as a tag catalogue. The policy is
+/// : the authored set is fully known
+/// synchronously, so re-running discovery on a timer can only produce the same tree.
+///
+///
+/// P1 scope — no Sparkplug. is implemented but
+/// never fires in : the
+/// authored set only changes by redeploy. Sparkplug B flips the policy to
+/// and raises the event on DBIRTH — that
+/// is a later task, and the seam here () exists for it.
+///
+///
+/// Composition order is load-bearing.
+/// must run before any subscribe and before the connect completes — it is what wires
+/// message delivery and, critically, the reconnect re-subscribe. The connection's
+/// Reconnected handler is passed through unwrapped: the manager throws when it can
+/// re-establish nothing, and that throw is precisely what tears the session down and retries.
+/// Swallowing it would leave a driver that reports Connected and receives nothing.
+///
+///
+/// Constructor is connection-free. It maps the authored TagConfig blobs (pure, no I/O)
+/// so discovery and reads work off a configured-but-not-yet-connected driver; every network
+/// operation happens in .
+///
+///
+public sealed class MqttDriver
+ : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable
+{
+ ///
+ /// Rough per-tag driver-attributable cost: the parsed (five
+ /// strings — RawPath, topic, JSONPath and the record header) plus its cached
+ /// and the two dictionary entries that index it. An estimate
+ /// by construction — GetMemoryFootprint exists to drive a cache-budget decision, not
+ /// to be exact.
+ ///
+ private const long ApproxBytesPerAuthoredTag = 512;
+
+ /// How often the host-connectivity poll samples .
+ private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
+
+ private readonly string _driverInstanceId;
+ private readonly ILogger? _logger;
+
+ ///
+ /// Owned by the driver, not by the manager, so a manager rebuilt for changed ingest settings
+ /// inherits the observed values instead of silently regressing every node to
+ /// BadWaitingForInitialData.
+ ///
+ private readonly LastValueCache _values = new();
+
+ private readonly object _hostLock = new();
+
+ /// Guards / / against each other.
+ private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
+
+ private MqttDriverOptions _options;
+ private MqttSubscriptionManager _subscriptions;
+
+ ///
+ /// RawPaths of the currently registered authored tags, in authoring order — the discovery
+ /// enumeration set. The manager resolves a RawPath to its definition but does not enumerate,
+ /// and enumerating from here is what makes "authored only" structural rather than incidental.
+ ///
+ private IReadOnlyList _authoredRawPaths = [];
+
+ private MqttConnection? _connection;
+ private CancellationTokenSource? _hostProbeCts;
+
+ private DriverHealth _health = new(DriverState.Unknown, null, null);
+ private HostState _hostState = HostState.Unknown;
+ private DateTime _hostStateChangedUtc = DateTime.UtcNow;
+ private int _disposed;
+
+ /// Initializes a new driver. Stores + maps configuration only — no network, no client.
+ ///
+ /// Driver configuration, including the authored .
+ /// Task 9's factory deserializes it from the DriverConfig JSON; a later
+ /// may replace it from a fresh config blob.
+ ///
+ /// Stable logical id of this driver instance.
+ /// Optional logger; never receives credentials or payload bodies.
+ public MqttDriver(MqttDriverOptions options, string driverInstanceId, ILogger? logger = null)
+ {
+ ArgumentNullException.ThrowIfNull(options);
+ ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
+
+ _options = options;
+ _driverInstanceId = driverInstanceId;
+ _logger = logger;
+ _subscriptions = CreateSubscriptionManager(options);
+ RegisterAuthoredTags(options);
+ }
+
+ ///
+ public event EventHandler? OnDataChange;
+
+ ///
+ public event EventHandler? OnHostStatusChanged;
+
+ ///
+ public event EventHandler? OnRediscoveryNeeded;
+
+ ///
+ public string DriverInstanceId => _driverInstanceId;
+
+ ///
+ // Literal rather than a constant: DriverTypeNames.Mqtt does not exist yet — Task 9 adds it and
+ // owns that file. The string must stay EXACTLY "Mqtt" and match MqttDriverProbe.DriverType: a
+ // DriverType that drifts from the persisted one silently breaks dispatch (the ModbusTcp/Modbus
+ // incident).
+ public string DriverType => "Mqtt";
+
+ ///
+ /// The ingest path this driver composes. Internal: the P2 Sparkplug handler feeds the same
+ /// OnDataChange + sinks through it, and the shell tests
+ /// drive to simulate broker traffic
+ /// without a broker.
+ ///
+ internal MqttSubscriptionManager Subscriptions => _subscriptions;
+
+ // ---- IDriver: lifecycle ----
+
+ ///
+ /// Adopts (falling back to the constructor's options when
+ /// it is blank or unusable), registers the authored tags, and opens the broker session.
+ ///
+ ///
+ /// Ordering is the contract: register → attach → connect. Attaching before the connect
+ /// completes means the reconnect re-subscribe is wired for the very first drop, and the
+ /// manager's SUBSCRIBE transport is live before the OPC UA server's first subscribe arrives.
+ ///
+ /// The driver configuration as JSON.
+ /// Cancellation for the connect.
+ /// A task that represents the asynchronous operation.
+ public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+ {
+ await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
+ try
+ {
+ await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
+ }
+ finally
+ {
+ _lifecycleGate.Release();
+ }
+ }
+
+ ///
+ /// Applies a config delta in place. A tag-only delta re-registers the authored table without
+ /// touching the broker session; a delta that changes how the driver connects or ingests
+ /// rebuilds the session.
+ ///
+ ///
+ /// A bad delta never faults the driver. Unparseable or unusable config is logged and
+ /// discarded, leaving the running configuration exactly as it was — a redeploy carrying one
+ /// malformed driver blob must not take a healthy driver's whole address space Bad. A genuine
+ /// connect failure against a changed endpoint is a different thing and does fault,
+ /// exactly as would.
+ ///
+ /// The driver configuration as JSON.
+ /// Cancellation for any reconnect the delta forces.
+ /// A task that represents the asynchronous operation.
+ public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+ {
+ MqttDriverOptions? next;
+ try
+ {
+ next = ParseOptions(driverConfigJson);
+ }
+ catch (Exception ex)
+ {
+ // Defence in depth: ParseOptions already swallows the JSON failure modes.
+ _logger?.LogWarning(
+ ex,
+ "MQTT driver '{DriverId}': reinitialize config could not be read; keeping the running configuration.",
+ _driverInstanceId);
+ return;
+ }
+
+ if (next is null)
+ {
+ _logger?.LogWarning(
+ "MQTT driver '{DriverId}': reinitialize config was blank or unusable; keeping the running configuration.",
+ _driverInstanceId);
+ return;
+ }
+
+ await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
+ try
+ {
+ if (!SameSession(next, _options))
+ {
+ // Broker endpoint / credentials / ingest shape changed — the live session cannot be
+ // reused. Full rebuild; a failure here is a real fault, not a bad delta.
+ _logger?.LogInformation(
+ "MQTT driver '{DriverId}': reinitialize changes the broker session; rebuilding.",
+ _driverInstanceId);
+ await TeardownAsync().ConfigureAwait(false);
+ await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
+ return;
+ }
+
+ _options = next;
+ RegisterAuthoredTags(next);
+ _logger?.LogInformation(
+ "MQTT driver '{DriverId}': reinitialize applied in place; {TagCount} authored tag(s).",
+ _driverInstanceId,
+ _authoredRawPaths.Count);
+ }
+ finally
+ {
+ _lifecycleGate.Release();
+ }
+ }
+
+ ///
+ public async Task ShutdownAsync(CancellationToken cancellationToken)
+ {
+ var lastMessage = ReadHealth().LastSuccessfulRead;
+ if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
+ .ConfigureAwait(false))
+ {
+ try
+ {
+ await TeardownAsync().ConfigureAwait(false);
+ }
+ finally
+ {
+ _lifecycleGate.Release();
+ }
+ }
+ else
+ {
+ // A wedged initialize must not turn shutdown into a hang; tear the session down anyway.
+ _logger?.LogWarning(
+ "MQTT driver '{DriverId}': lifecycle gate still held at shutdown; tearing the session down anyway.",
+ _driverInstanceId);
+ await TeardownAsync().ConfigureAwait(false);
+ }
+
+ WriteHealth(new DriverHealth(DriverState.Unknown, lastMessage, null));
+ TransitionHost(HostState.Unknown);
+ }
+
+ ///
+ /// Current health. Derived live from the connection's own state where one exists, so a
+ /// background reconnect is visible without the driver having to mirror every transition.
+ /// LastSuccessfulRead carries the last inbound message instant — MQTT never polls, so
+ /// "last message age" is the only evidence the session is actually delivering.
+ ///
+ /// The driver's current health snapshot.
+ public DriverHealth GetHealth()
+ {
+ var stored = ReadHealth();
+ var connection = _connection;
+ if (connection is null)
+ {
+ return stored;
+ }
+
+ var lastMessage = connection.LastMessageUtc ?? stored.LastSuccessfulRead;
+ return connection.State switch
+ {
+ MqttConnectionState.Connected => new DriverHealth(DriverState.Healthy, lastMessage, stored.LastError),
+ MqttConnectionState.Reconnecting => new DriverHealth(DriverState.Reconnecting, lastMessage, stored.LastError),
+ MqttConnectionState.Faulted => new DriverHealth(DriverState.Faulted, lastMessage, stored.LastError),
+ // Disconnected before the first successful connect, or Disposed after teardown: the
+ // stored snapshot (Initializing / Faulted / Unknown) is the more informative answer.
+ _ => stored with { LastSuccessfulRead = lastMessage },
+ };
+ }
+
+ ///
+ /// Approximate driver-attributable footprint: the authored definition table and the
+ /// last-value cache that indexes it. Both are sized by the authored tag count, which is the
+ /// only thing a deployment can grow.
+ ///
+ /// The approximate memory footprint in bytes.
+ public long GetMemoryFootprint() => _authoredRawPaths.Count * ApproxBytesPerAuthoredTag;
+
+ ///
+ /// No-op, deliberately. This driver holds no optional cache: the authored table is the
+ /// address space itself, and the last-value cache is — MQTT is
+ /// subscribe-first, so a dropped value is not re-fetchable, and flushing it would turn every
+ /// node Bad until its topic next published, which for a slow-changing tag can be never.
+ /// Birth/alias state (Sparkplug, P2) is correctness state for the same reason.
+ ///
+ /// Unused; present for the shape.
+ /// A completed task.
+ public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+ // ---- ITagDiscovery ----
+
+ ///
+ ///
+ /// Plain mode discovers synchronously from the authored set, so one pass is all there is.
+ /// Sparkplug B (P2) fills its metric set in asynchronously from birth certificates and
+ /// switches this to .
+ ///
+ public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
+
+ ///
+ ///
+ /// Always : replays authored tags, it does
+ /// not enumerate a backend. A broker's live topic set is not a tag catalogue — the
+ /// Driver.Mqtt.Browser project is the surface for observing traffic during authoring.
+ ///
+ public bool SupportsOnlineDiscovery => false;
+
+ ///
+ /// Streams the authored tag set — and only that — into .
+ ///
+ ///
+ /// The enumeration source is the driver's own list of registered RawPaths, never anything
+ /// observed on the wire, so no amount of broker traffic can add a node. A RawPath whose
+ /// TagConfig failed to map is absent from the manager and is skipped here too — it surfaces
+ /// as BadNodeIdUnknown on read rather than as a wrongly-typed variable.
+ ///
+ /// The address space builder to stream discovered nodes into.
+ /// Cancellation for the discovery operation.
+ /// A task that represents the asynchronous operation.
+ public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
+ {
+ ArgumentNullException.ThrowIfNull(builder);
+ cancellationToken.ThrowIfCancellationRequested();
+
+ var folder = builder.Folder("Mqtt", "Mqtt");
+ foreach (var rawPath in _authoredRawPaths)
+ {
+ if (!_subscriptions.TryResolve(rawPath, out var def))
+ {
+ continue;
+ }
+
+ folder.Variable(def.Name, def.Name, new DriverAttributeInfo(
+ FullName: def.Name,
+ DriverDataType: def.DataType,
+ IsArray: false,
+ ArrayDim: null,
+ // MQTT ingest is one-way in P1 — this driver implements no IWritable leg, so every
+ // node is ViewOnly rather than an Operate node whose writes would silently vanish.
+ SecurityClass: SecurityClassification.ViewOnly,
+ IsHistorized: false,
+ IsAlarm: false,
+ WriteIdempotent: false));
+ }
+
+ return Task.CompletedTask;
+ }
+
+ // ---- ISubscribable (straight through to the manager) ----
+
+ ///
+ public Task SubscribeAsync(
+ IReadOnlyList fullReferences,
+ TimeSpan publishingInterval,
+ CancellationToken cancellationToken)
+ => _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);
+
+ ///
+ public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
+ => _subscriptions.UnsubscribeAsync(handle, cancellationToken);
+
+ // ---- IReadable ----
+
+ ///
+ /// Serves the batch from the last-value cache. MQTT is push-based: there is nothing to poll,
+ /// so a read is always "what was most recently published".
+ ///
+ ///
+ /// Never throws. A reference that has not been observed — including one that is not an
+ /// authored tag at all — returns BadWaitingForInitialData in its own slot. Failing the
+ /// whole call would let one stale reference blank every other node in the same OPC UA read.
+ ///
+ /// The RawPath references to read.
+ /// Unused; the read touches no network.
+ /// One snapshot per requested reference, in request order.
+ public Task> ReadAsync(
+ IReadOnlyList fullReferences,
+ CancellationToken cancellationToken)
+ {
+ ArgumentNullException.ThrowIfNull(fullReferences);
+
+ var results = new DataValueSnapshot[fullReferences.Count];
+ for (var i = 0; i < fullReferences.Count; i++)
+ {
+ results[i] = _subscriptions.Values.Read(fullReferences[i]);
+ }
+
+ return Task.FromResult>(results);
+ }
+
+ // ---- IHostConnectivityProbe ----
+
+ /// The single broker this driver instance talks to, as host:port.
+ public string HostName => $"{_options.Host}:{_options.Port}";
+
+ ///
+ public IReadOnlyList GetHostStatuses()
+ {
+ lock (_hostLock)
+ {
+ return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
+ }
+ }
+
+ // ---- IRediscoverable ----
+
+ ///
+ /// Raises . Nothing in calls it
+ /// — the authored set only changes by redeploy. It exists for the Sparkplug B path, where a
+ /// DBIRTH can introduce metrics the previous birth certificate did not carry.
+ ///
+ /// Driver-supplied reason for the diagnostic log.
+ /// Optional subtree hint; means the whole tree.
+ internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null)
+ => OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));
+
+ // ---- disposal ----
+
+ ///
+ public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
+
+ ///
+ /// Performs the same teardown as so a caller that uses
+ /// await using without an explicit shutdown does not leak a live broker session.
+ ///
+ /// A task that represents the asynchronous dispose.
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return;
+ }
+
+ // The lifecycle gate is deliberately NOT disposed — same call as MqttConnection's semaphores.
+ // A caller that disposes before its last ShutdownAsync would otherwise get an
+ // ObjectDisposedException naming SemaphoreSlim, which says nothing useful; the gate holds no
+ // timer and no surviving registration, so leaving it undisposed costs nothing.
+ await TeardownAsync().ConfigureAwait(false);
+ }
+
+ // ---- internals ----
+
+ /// The gate-held body of .
+ private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken)
+ {
+ WriteHealth(new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null));
+ try
+ {
+ // A blank / unusable blob keeps the constructor's options rather than degrading the
+ // driver to defaults pointing at localhost.
+ var parsed = ParseOptions(driverConfigJson);
+ if (parsed is not null)
+ {
+ AdoptOptions(parsed);
+ }
+
+ RegisterAuthoredTags(_options);
+
+ var connection = new MqttConnection(_options, _driverInstanceId, _logger);
+
+ // MUST precede the connect: this wires message delivery AND the reconnect re-subscribe.
+ // The manager's Reconnected handler is passed through unwrapped on purpose — its
+ // throw-on-total-failure is what tears a deaf session down and retries it.
+ _subscriptions.AttachTo(connection);
+
+ await connection.ConnectAsync(cancellationToken).ConfigureAwait(false);
+ _connection = connection;
+
+ WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null));
+ TransitionHost(HostState.Running);
+ StartHostProbe();
+ }
+ catch (Exception ex)
+ {
+ WriteHealth(new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, ex.Message));
+ TransitionHost(HostState.Faulted);
+ throw;
+ }
+ }
+
+ ///
+ /// Deserializes a driver-config blob. Returns — never throws — when the
+ /// blob is blank or unusable, so both the initialize fallback and the reinitialize
+ /// keep-running-config rule read off one answer.
+ ///
+ private MqttDriverOptions? ParseOptions(string driverConfigJson)
+ {
+ if (string.IsNullOrWhiteSpace(driverConfigJson))
+ {
+ return null;
+ }
+
+ try
+ {
+ // The probe's shared instance, deliberately: enums must round-trip by NAME. A second,
+ // divergent JsonSerializerOptions is this repo's documented systemic enum bug, where an
+ // AdminUI-authored config with a numeric enum field faults the driver.
+ return JsonSerializer.Deserialize(driverConfigJson, MqttDriverProbe.JsonOpts);
+ }
+ catch (JsonException ex)
+ {
+ _logger?.LogWarning(
+ ex,
+ "MQTT driver '{DriverId}': driver config JSON is invalid.",
+ _driverInstanceId);
+ return null;
+ }
+ catch (NotSupportedException ex)
+ {
+ _logger?.LogWarning(
+ ex,
+ "MQTT driver '{DriverId}': driver config JSON could not be bound to the options shape.",
+ _driverInstanceId);
+ return null;
+ }
+ }
+
+ ///
+ /// Adopts new options, rebuilding the subscription manager only when a setting it captures at
+ /// construction actually changed. The rebuilt manager inherits the driver-owned
+ /// , so adopting new options never blanks observed values.
+ ///
+ private void AdoptOptions(MqttDriverOptions next)
+ {
+ if (!SameIngest(next, _options))
+ {
+ _subscriptions = CreateSubscriptionManager(next);
+ }
+
+ _options = next;
+ }
+
+ /// Builds a manager over the driver-owned cache and bridges its notifications to .
+ private MqttSubscriptionManager CreateSubscriptionManager(MqttDriverOptions options)
+ {
+ var manager = new MqttSubscriptionManager(
+ options,
+ _driverInstanceId,
+ transport: null,
+ logger: _logger,
+ cache: _values,
+ maxPayloadBytes: options.MaxPayloadBytes);
+
+ // The published reference is the RawPath the manager already stamped on the args; the driver
+ // only re-raises with itself as sender.
+ manager.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
+ return manager;
+ }
+
+ /// Registers the authored raw tags wholesale and refreshes the discovery enumeration set.
+ private void RegisterAuthoredTags(MqttDriverOptions options)
+ {
+ var mapped = _subscriptions.Register(options.RawTags);
+
+ // Enumerate in authoring order; the manager owns which of these actually mapped, and
+ // DiscoverAsync skips the rest.
+ _authoredRawPaths = [.. options.RawTags.Select(t => t.RawPath)];
+
+ if (mapped != options.RawTags.Count)
+ {
+ _logger?.LogWarning(
+ "MQTT driver '{DriverId}': {Mapped} of {Authored} authored tag(s) mapped to a definition; "
+ + "the rest will report BadNodeIdUnknown.",
+ _driverInstanceId,
+ mapped,
+ options.RawTags.Count);
+ }
+ }
+
+ ///
+ /// Whether two option sets describe the same broker session. Note the explicit
+ /// : the identity projections are object-typed
+ /// anonymous instances, so != would compare references and report "changed"
+ /// every single time — which would tear down and rebuild a perfectly good session on every
+ /// redeploy, and (via ) swap in a subscription manager that is not
+ /// attached to the live connection.
+ ///
+ private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b)
+ => SessionIdentity(a).Equals(SessionIdentity(b));
+
+ /// Whether two option sets produce an identical .
+ private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b)
+ => IngestIdentity(a).Equals(IngestIdentity(b));
+
+ ///
+ /// The settings whose change invalidates the live broker session. Projected onto an anonymous
+ /// type (structural Equals) rather than comparing the whole options record, whose
+ /// RawTags list would make every comparison unequal by reference.
+ ///
+ private static object SessionIdentity(MqttDriverOptions o) => new
+ {
+ o.Host,
+ o.Port,
+ o.ClientId,
+ o.UseTls,
+ o.AllowUntrustedServerCertificate,
+ o.CaCertificatePath,
+ o.Username,
+ o.Password,
+ o.ProtocolVersion,
+ o.CleanSession,
+ o.KeepAliveSeconds,
+ o.ConnectTimeoutSeconds,
+ o.ReconnectMinBackoffSeconds,
+ o.ReconnectMaxBackoffSeconds,
+ Ingest = IngestIdentity(o),
+ };
+
+ /// The settings captures at construction.
+ private static object IngestIdentity(MqttDriverOptions o) => new
+ {
+ o.Mode,
+ DefaultQos = o.Plain?.DefaultQos,
+ o.MaxPayloadBytes,
+ };
+
+ /// Shared teardown for , and the session rebuild.
+ private async Task TeardownAsync()
+ {
+ var probe = Interlocked.Exchange(ref _hostProbeCts, null);
+ if (probe is not null)
+ {
+ await probe.CancelAsync().ConfigureAwait(false);
+ probe.Dispose();
+ }
+
+ var connection = Interlocked.Exchange(ref _connection, null);
+ if (connection is not null)
+ {
+ await connection.DisposeAsync().ConfigureAwait(false);
+ }
+ }
+
+ ///
+ /// Starts the host-connectivity poll. exposes state but no
+ /// state-changed event, and an whose event never fires is
+ /// a capability that looks wired and is inert — so the transition signal is sampled here.
+ ///
+ private void StartHostProbe()
+ {
+ var cts = new CancellationTokenSource();
+ var previous = Interlocked.Exchange(ref _hostProbeCts, cts);
+ if (previous is not null)
+ {
+ // Cancel before disposing: disposing a live source leaves its loop running against a
+ // token that will never be cancelled, and a second probe would fight the first.
+ previous.Cancel();
+ previous.Dispose();
+ }
+
+ _ = Task.Run(() => HostProbeLoopAsync(cts.Token), cts.Token);
+ }
+
+ private async Task HostProbeLoopAsync(CancellationToken ct)
+ {
+ try
+ {
+ while (!ct.IsCancellationRequested)
+ {
+ await Task.Delay(HostProbeInterval, ct).ConfigureAwait(false);
+
+ var connection = _connection;
+ if (connection is null)
+ {
+ continue;
+ }
+
+ TransitionHost(connection.State switch
+ {
+ MqttConnectionState.Connected => HostState.Running,
+ MqttConnectionState.Reconnecting or MqttConnectionState.Disconnected => HostState.Stopped,
+ MqttConnectionState.Faulted => HostState.Faulted,
+ _ => HostState.Unknown,
+ });
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // Shutdown.
+ }
+ catch (Exception ex)
+ {
+ _logger?.LogWarning(ex, "MQTT driver '{DriverId}': host connectivity probe stopped.", _driverInstanceId);
+ }
+ }
+
+ /// Publishes a host state transition once; repeats of the current state are ignored.
+ private void TransitionHost(HostState next)
+ {
+ HostState previous;
+ lock (_hostLock)
+ {
+ if (_hostState == next)
+ {
+ return;
+ }
+
+ previous = _hostState;
+ _hostState = next;
+ _hostStateChangedUtc = DateTime.UtcNow;
+ }
+
+ OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, previous, next));
+ }
+
+ /// Barrier-protected read of the multi-threaded health field.
+ private DriverHealth ReadHealth() => Volatile.Read(ref _health);
+
+ /// Barrier-protected publish of a new health snapshot.
+ private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
new file mode 100644
index 00000000..70a92223
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
@@ -0,0 +1,305 @@
+using System.Text;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+///
+/// Shell-level tests for : authored-only discovery, the lifecycle
+/// surface (Reinitialize / Shutdown / health / footprint / cache flush), and the
+/// IReadable batch contract. Nothing here dials a broker — every test exercises the
+/// connection-free half of the driver, which is exactly the half a chatty broker must not be
+/// able to influence.
+///
+[Trait("Category", "Unit")]
+public sealed class MqttDriverDiscoveryTests
+{
+ ///
+ /// An authored TagConfig blob. dataType uses the real
+ /// member names — there is no Double; the 64-bit float is Float64.
+ ///
+ private static string TagJson(string topic, string dataType = "String", string payloadFormat = "Raw")
+ => $$"""{"topic":"{{topic}}","payloadFormat":"{{payloadFormat}}","dataType":"{{dataType}}"}""";
+
+ private static RawTagEntry Tag(string rawPath, string topic, string dataType = "String")
+ => new(rawPath, TagJson(topic, dataType), WriteIdempotent: false);
+
+ private static MqttDriver PlainDriver(params RawTagEntry[] tags)
+ => new(new MqttDriverOptions { Mode = MqttMode.Plain, RawTags = tags }, "d", null);
+
+ ///
+ /// Plain mode replays ONLY the authored tag set: one variable per authored raw tag, a
+ /// single discovery pass (), and no online
+ /// discovery — the universal browser must never treat a broker's topic traffic as a
+ /// browsable address space.
+ ///
+ [Fact]
+ public async Task DiscoverAsync_Plain_StreamsAuthoredTagsOnly_PolicyOnce()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+ var b = new CapturingAddressSpaceBuilder();
+
+ await driver.DiscoverAsync(b, CancellationToken.None);
+
+ b.Variables.Count.ShouldBe(1);
+ b.Variables[0].Info.FullName.ShouldBe("Plant/Mqtt/dev1/Temp");
+ driver.RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);
+ ((ITagDiscovery)driver).SupportsOnlineDiscovery.ShouldBeFalse();
+ }
+
+ ///
+ /// The falsifiability pin for authored-only discovery: a message that arrives on a topic no
+ /// authored tag names must not add anything to the discovered set. A driver that
+ /// auto-provisioned from broker traffic would grow the address space on every deploy against
+ /// a chatty broker.
+ ///
+ [Fact]
+ public async Task DiscoverAsync_UnauthoredBrokerTraffic_DoesNotAppear()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+
+ // Traffic this driver never authored — a neighbouring publisher on the same broker.
+ driver.Subscriptions.HandleMessage("some/other/topic", Encoding.UTF8.GetBytes("42"), retained: false);
+ driver.Subscriptions.HandleMessage("f/t/deeper", Encoding.UTF8.GetBytes("42"), retained: false);
+
+ var b = new CapturingAddressSpaceBuilder();
+ await driver.DiscoverAsync(b, CancellationToken.None);
+
+ b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Temp"]);
+ }
+
+ /// A raw tag whose TagConfig does not map is skipped, never thrown, and never discovered.
+ [Fact]
+ public async Task DiscoverAsync_SkipsUnmappableTagConfig()
+ {
+ var driver = PlainDriver(
+ Tag("Plant/Mqtt/dev1/Good", "f/t"),
+ new RawTagEntry("Plant/Mqtt/dev1/Bad", "not-json", WriteIdempotent: false));
+
+ var b = new CapturingAddressSpaceBuilder();
+ await driver.DiscoverAsync(b, CancellationToken.None);
+
+ b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Good"]);
+ }
+
+ /// Discovered MQTT variables are read-only — this driver has no IWritable leg.
+ [Fact]
+ public async Task DiscoverAsync_MarksVariablesViewOnly()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t", "Float64"));
+ var b = new CapturingAddressSpaceBuilder();
+
+ await driver.DiscoverAsync(b, CancellationToken.None);
+
+ b.Variables[0].Info.SecurityClass.ShouldBe(SecurityClassification.ViewOnly);
+ b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Float64);
+ }
+
+ ///
+ /// The falsifiability pin for the batch-read contract: a reference that is not an authored
+ /// tag degrades to BadWaitingForInitialData in its own slot rather than throwing and
+ /// failing every other reference in the same batch.
+ ///
+ [Fact]
+ public async Task ReadAsync_UnknownReference_DoesNotThrow_AndDegradesPerRef()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+ driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);
+
+ var results = await driver.ReadAsync(
+ ["Plant/Mqtt/dev1/Temp", "Plant/Mqtt/dev1/NeverAuthored"],
+ CancellationToken.None);
+
+ results.Count.ShouldBe(2);
+ results[0].StatusCode.ShouldBe(0u);
+ results[0].Value.ShouldBe("hot");
+ results[1].StatusCode.ShouldBe(0x80320000u); // BadWaitingForInitialData
+ }
+
+ /// An empty batch is legal and returns an empty result, not an exception.
+ [Fact]
+ public async Task ReadAsync_EmptyBatch_ReturnsEmpty()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+
+ var results = await driver.ReadAsync([], CancellationToken.None);
+
+ results.ShouldBeEmpty();
+ }
+
+ ///
+ /// The falsifiability pin for the cache-flush contract: the last-value cache backs
+ /// IReadable, so flushing it would silently turn every subscribed node Bad under
+ /// memory pressure. FlushOptionalCachesAsync must leave it untouched.
+ ///
+ [Fact]
+ public async Task FlushOptionalCachesAsync_LeavesLastValueCacheIntact()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+ driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);
+
+ await driver.FlushOptionalCachesAsync(CancellationToken.None);
+
+ var results = await driver.ReadAsync(["Plant/Mqtt/dev1/Temp"], CancellationToken.None);
+ results[0].StatusCode.ShouldBe(0u);
+ results[0].Value.ShouldBe("hot");
+ }
+
+ /// A malformed reinitialize delta must never fault the driver, and must not lose the running config.
+ [Fact]
+ public async Task ReinitializeAsync_MalformedDelta_KeepsRunningConfig_AndDoesNotFault()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+
+ await Should.NotThrowAsync(() => driver.ReinitializeAsync("{ this is not json", CancellationToken.None));
+
+ driver.GetHealth().State.ShouldNotBe(DriverState.Faulted);
+
+ var b = new CapturingAddressSpaceBuilder();
+ await driver.DiscoverAsync(b, CancellationToken.None);
+ b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Temp"]);
+ }
+
+ ///
+ /// A delta that changes only the authored tag set is applied in place — no teardown, no
+ /// reconnect — and is applied WHOLESALE: a tag the redeploy dropped stops being discovered.
+ ///
+ [Fact]
+ public async Task ReinitializeAsync_TagOnlyDelta_AppliedInPlace_Wholesale()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+
+ var pressure = System.Text.Json.JsonSerializer.Serialize(TagJson("f/p"));
+ var flow = System.Text.Json.JsonSerializer.Serialize(TagJson("f/q"));
+ var delta = $$"""
+ {
+ "mode": "Plain",
+ "rawTags": [
+ { "rawPath": "Plant/Mqtt/dev1/Pressure", "tagConfig": {{pressure}}, "writeIdempotent": false },
+ { "rawPath": "Plant/Mqtt/dev1/Flow", "tagConfig": {{flow}}, "writeIdempotent": false }
+ ]
+ }
+ """;
+
+ await driver.ReinitializeAsync(delta, CancellationToken.None);
+
+ var b = new CapturingAddressSpaceBuilder();
+ await driver.DiscoverAsync(b, CancellationToken.None);
+
+ b.Variables.Select(v => v.Info.FullName)
+ .OrderBy(x => x, StringComparer.Ordinal)
+ .ShouldBe(["Plant/Mqtt/dev1/Flow", "Plant/Mqtt/dev1/Pressure"]);
+ }
+
+ /// Identity is fixed at construction and must match the persisted DriverInstance.DriverType.
+ [Fact]
+ public void Identity_IsMqtt()
+ {
+ var driver = PlainDriver();
+
+ driver.DriverType.ShouldBe("Mqtt");
+ driver.DriverInstanceId.ShouldBe("d");
+ }
+
+ /// A driver that has never been initialized reports Unknown, not Healthy.
+ [Fact]
+ public void GetHealth_BeforeInitialize_IsUnknown()
+ {
+ PlainDriver().GetHealth().State.ShouldBe(DriverState.Unknown);
+ }
+
+ /// Plain mode never signals rediscovery — the authored set only changes by redeploy.
+ [Fact]
+ public async Task Plain_NeverRaisesRediscoveryNeeded()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+ var fired = 0;
+ ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
+
+ driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);
+ await driver.DiscoverAsync(new CapturingAddressSpaceBuilder(), CancellationToken.None);
+
+ fired.ShouldBe(0);
+ }
+
+ /// The footprint tracks the authored table; an empty driver reports no driver-attributable bytes.
+ [Fact]
+ public void GetMemoryFootprint_TracksAuthoredTable()
+ {
+ PlainDriver().GetMemoryFootprint().ShouldBe(0);
+ PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).GetMemoryFootprint().ShouldBeGreaterThan(0);
+ }
+
+ /// Shutdown on a never-initialized driver is a no-op, not a null-reference.
+ [Fact]
+ public async Task ShutdownAsync_BeforeInitialize_DoesNotThrow()
+ {
+ var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
+
+ await Should.NotThrowAsync(() => driver.ShutdownAsync(CancellationToken.None));
+ driver.GetHealth().State.ShouldBe(DriverState.Unknown);
+ }
+
+ /// The single-broker connectivity probe names the configured endpoint.
+ [Fact]
+ public void GetHostStatuses_ReportsTheConfiguredBroker()
+ {
+ var driver = new MqttDriver(
+ new MqttDriverOptions { Mode = MqttMode.Plain, Host = "broker.example", Port = 1883 },
+ "d",
+ null);
+
+ var statuses = ((IHostConnectivityProbe)driver).GetHostStatuses();
+
+ statuses.Count.ShouldBe(1);
+ statuses[0].HostName.ShouldBe("broker.example:1883");
+ statuses[0].State.ShouldBe(HostState.Unknown);
+ }
+
+ // ---- test doubles ----
+
+ ///
+ /// Records the streamed discovery tree instead of materializing OPC UA nodes. Local to this
+ /// suite: the driver test project does not reference Commons, where the runtime's own
+ /// capturing builder lives. Mirrors the per-driver RecordingBuilder the AB CIP / FOCAS
+ /// suites use.
+ ///
+ private sealed class CapturingAddressSpaceBuilder : IAddressSpaceBuilder
+ {
+ /// Every folder streamed, in order, across the whole tree.
+ public List<(string BrowseName, string DisplayName)> Folders { get; } = [];
+
+ /// Every variable streamed, in order, across the whole tree.
+ public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];
+
+ ///
+ public IAddressSpaceBuilder Folder(string browseName, string displayName)
+ {
+ Folders.Add((browseName, displayName));
+ return this;
+ }
+
+ ///
+ public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
+ {
+ Variables.Add((browseName, attributeInfo));
+ return new Handle(attributeInfo.FullName);
+ }
+
+ ///
+ public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
+
+ private sealed class Handle(string fullRef) : IVariableHandle
+ {
+ public string FullReference => fullRef;
+
+ public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new NullSink();
+ }
+
+ private sealed class NullSink : IAlarmConditionSink
+ {
+ public void OnTransition(AlarmEventArgs args) { }
+ }
+ }
+}