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

feat(mqtt): MqttDriver shell — lifecycle + authored-only discovery (Once)

Browse Source 
Composes MqttConnection + MqttSubscriptionManager + LastValueCache into the
IDriver / ITagDiscovery / ISubscribable / IReadable / IHostConnectivityProbe /
IRediscoverable capability set.

- Discovery replays ONLY the authored raw tags (RediscoverPolicy = Once,
  SupportsOnlineDiscovery = false) — a chatty broker can never auto-provision.
- Composition order is register -> AttachTo -> ConnectAsync; the manager's
  Reconnected handler is passed through unwrapped so its throw-on-total-failure
  still tears a deaf session down.
- ReinitializeAsync applies a tag-only delta in place and never faults on a bad
  one; a session-changing delta rebuilds.
- ReadAsync serves the last-value cache and degrades per reference
  (BadWaitingForInitialData), never throwing for the batch.
- FlushOptionalCachesAsync is a no-op: the last-value cache IS IReadable.
- Adds MqttDriverOptions.RawTags (the authored-tag delivery mechanism every
  other driver's options DTO already has) and promotes MaxPayloadBytes from a
  manager ctor knob to an operator-facing key.
- Converges on MqttDriverProbe.JsonOpts rather than a second, divergent
  JsonSerializerOptions.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 17:06:39 -04:00
parent 211c6ea6d6
commit 420692b6e5
3 changed files with 1073 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs
+36
View File
@@ -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;
/// <summary>
/// Ceiling on an inbound message body, in bytes. A larger message is refused before any
/// decode or parse and degrades its own tags to <c>BadDecodingError</c>. Default 1 MiB.
/// </summary>
/// <remarks>
/// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their
/// cost is paid by <i>every</i> 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 <c>MqttSubscriptionManager.DefaultMaxPayloadBytes</c>
/// because this <c>.Contracts</c> assembly is <i>referenced by</i> the driver assembly and
/// cannot reference it back; the manager's constructor is the single enforcement point.
/// </remarks>
[Range(1, int.MaxValue)]
public int MaxPayloadBytes { get; init; } = 1024 * 1024;
/// <summary>Selects the ingest shape — Plain topics or Sparkplug B.</summary>
public MqttMode Mode { get; init; } = MqttMode.Plain;
/// <summary>
/// The cluster's authored raw MQTT tags, as delivered by the deploy artifact: each carries the
/// tag's <b>RawPath</b> (its v3 identity and driver wire reference) plus its <c>TagConfig</c>
/// blob. <see cref="MqttDriver"/> maps each through <see cref="MqttTagDefinitionFactory"/> at
/// initialize and re-registers the set <b>wholesale</b> on every reinitialize, so a redeploy
/// that drops a tag stops feeding it.
/// </summary>
/// <remarks>
/// This is also the <b>only</b> 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 <c>ModbusDriverOptions.RawTags</c> /
/// <c>FocasDriverOptions.RawTags</c>.
/// </remarks>
public IReadOnlyList<RawTagEntry> RawTags { get; init; } = [];
/// <summary>
/// Sparkplug-only settings. Populated when <see cref="Mode"/> is
/// <see cref="MqttMode.SparkplugB"/>; <c>null</c> 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;
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+732
View File
@@ -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;
/// <summary>
/// The MQTT / Sparkplug B driver shell: composes <see cref="MqttConnection"/> (broker session +
/// reconnect supervisor), <see cref="MqttSubscriptionManager"/> (authored table, topic routing,
/// value extraction) and the manager's <see cref="LastValueCache"/> (the push→poll bridge) into
/// the driver capability set the Core consumes.
/// </summary>
/// <remarks>
/// <para>
/// <b>Authored-only discovery.</b> Plain MQTT has no browsable address space — a broker will
/// happily carry thousands of topics this deployment has nothing to do with.
/// <see cref="DiscoverAsync"/> therefore replays exactly the raw tags the deploy artifact
/// authored (<see cref="MqttDriverOptions.RawTags"/>) and nothing else, and
/// <see cref="ITagDiscovery.SupportsOnlineDiscovery"/> stays <see langword="false"/> so the
/// universal browser never treats broker traffic as a tag catalogue. The policy is
/// <see cref="DiscoveryRediscoverPolicy.Once"/>: the authored set is fully known
/// synchronously, so re-running discovery on a timer can only produce the same tree.
/// </para>
/// <para>
/// <b>P1 scope — no Sparkplug.</b> <see cref="IRediscoverable"/> is implemented but
/// <see cref="OnRediscoveryNeeded"/> never fires in <see cref="MqttMode.Plain"/>: the
/// authored set only changes by redeploy. Sparkplug B flips the policy to
/// <see cref="DiscoveryRediscoverPolicy.UntilStable"/> and raises the event on DBIRTH — that
/// is a later task, and the seam here (<see cref="RaiseRediscoveryNeeded"/>) exists for it.
/// </para>
/// <para>
/// <b>Composition order is load-bearing.</b> <see cref="MqttSubscriptionManager.AttachTo"/>
/// must run <i>before</i> any subscribe and before the connect completes — it is what wires
/// message delivery and, critically, the reconnect re-subscribe. The connection's
/// <c>Reconnected</c> 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.
/// </para>
/// <para>
/// <b>Constructor is connection-free.</b> 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 <see cref="InitializeAsync"/>.
/// </para>
/// </remarks>
public sealed class MqttDriver
: IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable
{
/// <summary>
/// Rough per-tag driver-attributable cost: the parsed <see cref="MqttTagDefinition"/> (five
/// strings — RawPath, topic, JSONPath and the record header) plus its cached
/// <see cref="DataValueSnapshot"/> and the two dictionary entries that index it. An estimate
/// by construction — <c>GetMemoryFootprint</c> exists to drive a cache-budget decision, not
/// to be exact.
/// </summary>
private const long ApproxBytesPerAuthoredTag = 512;
/// <summary>How often the host-connectivity poll samples <see cref="MqttConnection.State"/>.</summary>
private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
private readonly string _driverInstanceId;
private readonly ILogger? _logger;
/// <summary>
/// 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
/// <c>BadWaitingForInitialData</c>.
/// </summary>
private readonly LastValueCache _values = new();
private readonly object _hostLock = new();
/// <summary>Guards <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> / <see cref="ShutdownAsync"/> against each other.</summary>
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
private MqttDriverOptions _options;
private MqttSubscriptionManager _subscriptions;
/// <summary>
/// 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.
/// </summary>
private IReadOnlyList<string> _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;
/// <summary>Initializes a new driver. Stores + maps configuration only — no network, no client.</summary>
/// <param name="options">
/// Driver configuration, including the authored <see cref="MqttDriverOptions.RawTags"/>.
/// Task 9's factory deserializes it from the <c>DriverConfig</c> JSON; a later
/// <see cref="ReinitializeAsync"/> may replace it from a fresh config blob.
/// </param>
/// <param name="driverInstanceId">Stable logical id of this driver instance.</param>
/// <param name="logger">Optional logger; never receives credentials or payload bodies.</param>
public MqttDriver(MqttDriverOptions options, string driverInstanceId, ILogger<MqttDriver>? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
_options = options;
_driverInstanceId = driverInstanceId;
_logger = logger;
_subscriptions = CreateSubscriptionManager(options);
RegisterAuthoredTags(options);
}
/// <inheritdoc />
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <inheritdoc />
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
/// <inheritdoc />
public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;
/// <inheritdoc />
public string DriverInstanceId => _driverInstanceId;
/// <inheritdoc />
// 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";
/// <summary>
/// The ingest path this driver composes. Internal: the P2 Sparkplug handler feeds the same
/// <c>OnDataChange</c> + <see cref="LastValueCache"/> sinks through it, and the shell tests
/// drive <see cref="MqttSubscriptionManager.HandleMessage"/> to simulate broker traffic
/// without a broker.
/// </summary>
internal MqttSubscriptionManager Subscriptions => _subscriptions;
// ---- IDriver: lifecycle ----
/// <summary>
/// Adopts <paramref name="driverConfigJson"/> (falling back to the constructor's options when
/// it is blank or unusable), registers the authored tags, and opens the broker session.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <param name="driverConfigJson">The driver configuration as JSON.</param>
/// <param name="cancellationToken">Cancellation for the connect.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <b>A bad delta never faults the driver.</b> 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 <i>changed</i> endpoint is a different thing and does fault,
/// exactly as <see cref="InitializeAsync"/> would.
/// </remarks>
/// <param name="driverConfigJson">The driver configuration as JSON.</param>
/// <param name="cancellationToken">Cancellation for any reconnect the delta forces.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
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();
}
}
/// <inheritdoc />
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);
}
/// <summary>
/// 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.
/// <c>LastSuccessfulRead</c> carries the last inbound message instant — MQTT never polls, so
/// "last message age" is the only evidence the session is actually delivering.
/// </summary>
/// <returns>The driver's current health snapshot.</returns>
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 },
};
}
/// <summary>
/// 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.
/// </summary>
/// <returns>The approximate memory footprint in bytes.</returns>
public long GetMemoryFootprint() => _authoredRawPaths.Count * ApproxBytesPerAuthoredTag;
/// <summary>
/// No-op, deliberately. This driver holds no optional cache: the authored table is the
/// address space itself, and the last-value cache <b>is</b> <see cref="IReadable"/> — 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.
/// </summary>
/// <param name="cancellationToken">Unused; present for the <see cref="IDriver"/> shape.</param>
/// <returns>A completed task.</returns>
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
// ---- ITagDiscovery ----
/// <inheritdoc />
/// <remarks>
/// 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 <see cref="DiscoveryRediscoverPolicy.UntilStable"/>.
/// </remarks>
public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
/// <inheritdoc />
/// <remarks>
/// Always <see langword="false"/>: <see cref="DiscoverAsync"/> replays authored tags, it does
/// not enumerate a backend. A broker's live topic set is not a tag catalogue — the
/// <c>Driver.Mqtt.Browser</c> project is the surface for observing traffic during authoring.
/// </remarks>
public bool SupportsOnlineDiscovery => false;
/// <summary>
/// Streams the authored tag set — and only that — into <paramref name="builder"/>.
/// </summary>
/// <remarks>
/// 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 <c>BadNodeIdUnknown</c> on read rather than as a wrongly-typed variable.
/// </remarks>
/// <param name="builder">The address space builder to stream discovered nodes into.</param>
/// <param name="cancellationToken">Cancellation for the discovery operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
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) ----
/// <inheritdoc />
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences,
TimeSpan publishingInterval,
CancellationToken cancellationToken)
=> _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);
/// <inheritdoc />
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
=> _subscriptions.UnsubscribeAsync(handle, cancellationToken);
// ---- IReadable ----
/// <summary>
/// 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".
/// </summary>
/// <remarks>
/// <b>Never throws.</b> A reference that has not been observed — including one that is not an
/// authored tag at all — returns <c>BadWaitingForInitialData</c> in its own slot. Failing the
/// whole call would let one stale reference blank every other node in the same OPC UA read.
/// </remarks>
/// <param name="fullReferences">The RawPath references to read.</param>
/// <param name="cancellationToken">Unused; the read touches no network.</param>
/// <returns>One snapshot per requested reference, in request order.</returns>
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> 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<IReadOnlyList<DataValueSnapshot>>(results);
}
// ---- IHostConnectivityProbe ----
/// <summary>The single broker this driver instance talks to, as <c>host:port</c>.</summary>
public string HostName => $"{_options.Host}:{_options.Port}";
/// <inheritdoc />
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
{
lock (_hostLock)
{
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
}
// ---- IRediscoverable ----
/// <summary>
/// Raises <see cref="OnRediscoveryNeeded"/>. Nothing in <see cref="MqttMode.Plain"/> 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.
/// </summary>
/// <param name="reason">Driver-supplied reason for the diagnostic log.</param>
/// <param name="scopeHint">Optional subtree hint; <see langword="null"/> means the whole tree.</param>
internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null)
=> OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));
// ---- disposal ----
/// <inheritdoc />
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
/// <summary>
/// Performs the same teardown as <see cref="ShutdownAsync"/> so a caller that uses
/// <c>await using</c> without an explicit shutdown does not leak a live broker session.
/// </summary>
/// <returns>A task that represents the asynchronous dispose.</returns>
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 ----
/// <summary>The gate-held body of <see cref="InitializeAsync"/>.</summary>
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;
}
}
/// <summary>
/// Deserializes a driver-config blob. Returns <see langword="null"/> — 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.
/// </summary>
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<MqttDriverOptions>(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;
}
}
/// <summary>
/// Adopts new options, rebuilding the subscription manager only when a setting it captures at
/// construction actually changed. The rebuilt manager inherits the driver-owned
/// <see cref="LastValueCache"/>, so adopting new options never blanks observed values.
/// </summary>
private void AdoptOptions(MqttDriverOptions next)
{
if (!SameIngest(next, _options))
{
_subscriptions = CreateSubscriptionManager(next);
}
_options = next;
}
/// <summary>Builds a manager over the driver-owned cache and bridges its notifications to <see cref="OnDataChange"/>.</summary>
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;
}
/// <summary>Registers the authored raw tags wholesale and refreshes the discovery enumeration set.</summary>
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);
}
}
/// <summary>
/// Whether two option sets describe the same broker session. Note the explicit
/// <see cref="object.Equals(object?)"/>: the identity projections are <c>object</c>-typed
/// anonymous instances, so <c>!=</c> would compare <b>references</b> and report "changed"
/// every single time — which would tear down and rebuild a perfectly good session on every
/// redeploy, and (via <see cref="SameIngest"/>) swap in a subscription manager that is not
/// attached to the live connection.
/// </summary>
private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b)
=> SessionIdentity(a).Equals(SessionIdentity(b));
/// <summary>Whether two option sets produce an identical <see cref="MqttSubscriptionManager"/>.</summary>
private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b)
=> IngestIdentity(a).Equals(IngestIdentity(b));
/// <summary>
/// The settings whose change invalidates the live broker session. Projected onto an anonymous
/// type (structural <c>Equals</c>) rather than comparing the whole options record, whose
/// <c>RawTags</c> list would make every comparison unequal by reference.
/// </summary>
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),
};
/// <summary>The settings <see cref="MqttSubscriptionManager"/> captures at construction.</summary>
private static object IngestIdentity(MqttDriverOptions o) => new
{
o.Mode,
DefaultQos = o.Plain?.DefaultQos,
o.MaxPayloadBytes,
};
/// <summary>Shared teardown for <see cref="ShutdownAsync"/>, <see cref="DisposeAsync"/> and the session rebuild.</summary>
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);
}
}
/// <summary>
/// Starts the host-connectivity poll. <see cref="MqttConnection"/> exposes state but no
/// state-changed event, and an <see cref="IHostConnectivityProbe"/> whose event never fires is
/// a capability that looks wired and is inert — so the transition signal is sampled here.
/// </summary>
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);
}
}
/// <summary>Publishes a host state transition once; repeats of the current state are ignored.</summary>
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));
}
/// <summary>Barrier-protected read of the multi-threaded health field.</summary>
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
/// <summary>Barrier-protected publish of a new health snapshot.</summary>
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
+305
View File
@@ -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;
/// <summary>
/// Shell-level tests for <see cref="MqttDriver"/>: authored-only discovery, the lifecycle
/// surface (<c>Reinitialize</c> / <c>Shutdown</c> / health / footprint / cache flush), and the
/// <c>IReadable</c> 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.
/// </summary>
[Trait("Category", "Unit")]
public sealed class MqttDriverDiscoveryTests
{
/// <summary>
/// An authored TagConfig blob. <c>dataType</c> uses the real <see cref="DriverDataType"/>
/// member names — there is no <c>Double</c>; the 64-bit float is <c>Float64</c>.
/// </summary>
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);
/// <summary>
/// Plain mode replays ONLY the authored tag set: one variable per authored raw tag, a
/// single discovery pass (<see cref="DiscoveryRediscoverPolicy.Once"/>), and no online
/// discovery — the universal browser must never treat a broker's topic traffic as a
/// browsable address space.
/// </summary>
[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();
}
/// <summary>
/// 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.
/// </summary>
[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"]);
}
/// <summary>A raw tag whose TagConfig does not map is skipped, never thrown, and never discovered.</summary>
[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"]);
}
/// <summary>Discovered MQTT variables are read-only — this driver has no <c>IWritable</c> leg.</summary>
[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);
}
/// <summary>
/// The falsifiability pin for the batch-read contract: a reference that is not an authored
/// tag degrades to <c>BadWaitingForInitialData</c> in its own slot rather than throwing and
/// failing every other reference in the same batch.
/// </summary>
[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
}
/// <summary>An empty batch is legal and returns an empty result, not an exception.</summary>
[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();
}
/// <summary>
/// The falsifiability pin for the cache-flush contract: the last-value cache backs
/// <c>IReadable</c>, so flushing it would silently turn every subscribed node Bad under
/// memory pressure. <c>FlushOptionalCachesAsync</c> must leave it untouched.
/// </summary>
[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");
}
/// <summary>A malformed reinitialize delta must never fault the driver, and must not lose the running config.</summary>
[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"]);
}
/// <summary>
/// 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.
/// </summary>
[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"]);
}
/// <summary>Identity is fixed at construction and must match the persisted <c>DriverInstance.DriverType</c>.</summary>
[Fact]
public void Identity_IsMqtt()
{
var driver = PlainDriver();
driver.DriverType.ShouldBe("Mqtt");
driver.DriverInstanceId.ShouldBe("d");
}
/// <summary>A driver that has never been initialized reports Unknown, not Healthy.</summary>
[Fact]
public void GetHealth_BeforeInitialize_IsUnknown()
{
PlainDriver().GetHealth().State.ShouldBe(DriverState.Unknown);
}
/// <summary>Plain mode never signals rediscovery — the authored set only changes by redeploy.</summary>
[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);
}
/// <summary>The footprint tracks the authored table; an empty driver reports no driver-attributable bytes.</summary>
[Fact]
public void GetMemoryFootprint_TracksAuthoredTable()
{
PlainDriver().GetMemoryFootprint().ShouldBe(0);
PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).GetMemoryFootprint().ShouldBeGreaterThan(0);
}
/// <summary>Shutdown on a never-initialized driver is a no-op, not a null-reference.</summary>
[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);
}
/// <summary>The single-broker connectivity probe names the configured endpoint.</summary>
[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 ----
/// <summary>
/// Records the streamed discovery tree instead of materializing OPC UA nodes. Local to this
/// suite: the driver test project does not reference <c>Commons</c>, where the runtime's own
/// capturing builder lives. Mirrors the per-driver <c>RecordingBuilder</c> the AB CIP / FOCAS
/// suites use.
/// </summary>
private sealed class CapturingAddressSpaceBuilder : IAddressSpaceBuilder
{
/// <summary>Every folder streamed, in order, across the whole tree.</summary>
public List<(string BrowseName, string DisplayName)> Folders { get; } = [];
/// <summary>Every variable streamed, in order, across the whole tree.</summary>
public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];
/// <inheritdoc />
public IAddressSpaceBuilder Folder(string browseName, string displayName)
{
Folders.Add((browseName, displayName));
return this;
}
/// <inheritdoc />
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
{
Variables.Add((browseName, attributeInfo));
return new Handle(attributeInfo.FullName);
}
/// <inheritdoc />
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) { }
}
}
}