Joseph Doherty
826312befe
Clears CommentChecker findings in src/ (3,280 -> 209, the remainder being false positives like PLC-5/IEEE-754/M0.0) by adding missing summaries/params/returns, normalising inheritdoc usage, and removing Gitea/review tracking IDs while keeping the rationale they annotated. Comment-only: comment-stripped sources are identical to the parent commit.
191 lines
10 KiB
C#
191 lines
10 KiB
C#
using System.Buffers;
|
||
using System.Text.Json;
|
||
using Microsoft.Extensions.Logging;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using MQTTnet;
|
||
using MQTTnet.Protocol;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
|
||
|
||
/// <summary>
|
||
/// Bespoke MQTT address-picker browser: opens a short-lived, <b>strictly passive</b> observation
|
||
/// window against the broker named by the form's JSON.
|
||
/// <para>
|
||
/// MQTT has no discovery protocol, so there is nothing to enumerate — the browser subscribes
|
||
/// to a wildcard (<c>#</c> / <c>{topicPrefix}/#</c> in plain mode, <c>spBv1.0/{groupId}/#</c>
|
||
/// in Sparkplug mode) and lets <see cref="MqttBrowseSession"/> accumulate whatever arrives:
|
||
/// a topic segment tree in plain mode, a Group → EdgeNode → Device → Metric tree decoded from
|
||
/// observed birth certificates in Sparkplug mode. Nothing on any browse path publishes — in
|
||
/// either mode: an operator opening a picker must not be able to inject a message into a
|
||
/// running plant. The session's operator-triggered <c>RequestRebirthAsync</c> is the sole
|
||
/// sanctioned publish, and it is never reached by <c>OpenAsync</c> or by any browse call.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Layering note.</b> This project depends on the shared leaves only — <c>.Contracts</c> for
|
||
/// the Sparkplug wire schema and codec, <c>.Transport</c> for
|
||
/// <see cref="MqttClientOptionsFactory"/> — and NOT on the runtime <c>.Driver</c> project, which
|
||
/// it used to reference under a documented exception. Reuse of the TLS / CA-pin path
|
||
/// is what that exception bought, and it is preserved: there is still exactly one copy, now in
|
||
/// <c>.Transport</c>. Do not fork it. Pinned by <c>MqttBrowserLayeringTests</c>.
|
||
/// </para>
|
||
/// </summary>
|
||
public sealed class MqttDriverBrowser : IDriverBrowser
|
||
{
|
||
/// <summary>Floor on the open budget — a 1 s <c>ConnectTimeoutSeconds</c> would make browse unusable.</summary>
|
||
internal const int MinOpenBudgetSeconds = 5;
|
||
|
||
/// <summary>Ceiling on the open budget — the picker must never hang on an unreachable broker.</summary>
|
||
internal const int MaxOpenBudgetSeconds = 30;
|
||
|
||
/// <summary>Marks the transient browse identity in the broker's client-id/session logs.</summary>
|
||
internal const string BrowseClientIdPrefix = "-browse-";
|
||
|
||
private readonly ILogger<MqttDriverBrowser> _logger;
|
||
|
||
/// <summary>
|
||
/// Creates a browser. <b>Connection-free by contract</b> — the universal browser's
|
||
/// <c>CanBrowse</c> constructs a throwaway instance purely to ask which driver type it
|
||
/// handles, so the constructor must never touch the network.
|
||
/// </summary>
|
||
/// <param name="logger">Optional logger; defaults to <see cref="NullLogger{T}"/>. Never receives credentials.</param>
|
||
public MqttDriverBrowser(ILogger<MqttDriverBrowser>? logger = null) =>
|
||
_logger = logger ?? NullLogger<MqttDriverBrowser>.Instance;
|
||
|
||
/// <inheritdoc />
|
||
public string DriverType => DriverTypeNames.Mqtt;
|
||
|
||
// Connects with a browse-only client id, subscribes the wildcard at QoS 0, and hands back a
|
||
// session that serves whatever the subscription observes. The whole open is bounded by
|
||
// OpenBudget; nothing here publishes.
|
||
/// <inheritdoc />
|
||
public async Task<IBrowseSession> OpenAsync(string configJson, CancellationToken cancellationToken)
|
||
{
|
||
// MqttJson.Options — the ONE shared instance across factory / probe / driver / browser
|
||
// (see its remarks). Parsing the same DriverConfig blob through a second, divergent
|
||
// JsonSerializerOptions is this repo's documented systemic enum bug: the picker would accept
|
||
// a `mode` / `protocolVersion` spelling the deployed driver rejects, or vice versa.
|
||
var opts = JsonSerializer.Deserialize<MqttDriverOptions>(configJson, MqttJson.Options)
|
||
?? throw new InvalidOperationException("Mqtt options deserialized to null.");
|
||
|
||
var filter = BuildRootFilter(opts);
|
||
var suffix = BuildBrowseClientIdSuffix();
|
||
var clientOptions = MqttClientOptionsFactory.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger);
|
||
|
||
using var openCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||
openCts.CancelAfter(OpenBudget(opts));
|
||
|
||
var client = new MqttClientFactory().CreateMqttClient();
|
||
var session = new MqttBrowseSession(opts.Mode, client, _logger);
|
||
try
|
||
{
|
||
// Runs on MQTTnet's dispatcher: record and return, nothing else.
|
||
client.ApplicationMessageReceivedAsync += args =>
|
||
{
|
||
var payload = args.ApplicationMessage.Payload;
|
||
ReadOnlySpan<byte> body = payload.IsSingleSegment ? payload.FirstSpan : payload.ToArray();
|
||
session.Observe(args.ApplicationMessage.Topic, body);
|
||
return Task.CompletedTask;
|
||
};
|
||
|
||
await client.ConnectAsync(clientOptions, openCts.Token).ConfigureAwait(false);
|
||
|
||
// QoS 0: a transient observation window has no delivery guarantee to offer and should
|
||
// cost the broker as little as possible.
|
||
var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
|
||
.WithTopicFilter(f => f
|
||
.WithTopic(filter)
|
||
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce))
|
||
.Build();
|
||
await client.SubscribeAsync(subscribeOptions, openCts.Token).ConfigureAwait(false);
|
||
|
||
_logger.LogInformation(
|
||
"AdminUI MQTT browse session opened against {Host}:{Port} in {Mode} mode observing "
|
||
+ "'{Filter}' (read-only).",
|
||
opts.Host, opts.Port, opts.Mode, filter);
|
||
|
||
return session;
|
||
}
|
||
catch
|
||
{
|
||
await session.DisposeAsync().ConfigureAwait(false); // owns the client — disconnects + disposes
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// The wildcard the observation window subscribes to.
|
||
/// <para>
|
||
/// <b>Plain mode:</b> the whole broker (<c>#</c>) unless the config scopes it with a
|
||
/// topic prefix.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Sparkplug mode:</b> the Sparkplug namespace only —
|
||
/// <c>spBv1.0/{groupId}/#</c>, matching the group the deployed driver itself subscribes
|
||
/// (design §3.1), or <c>spBv1.0/#</c> when no group has been authored yet, so the picker
|
||
/// can discover which groups exist. The Plain-mode topic prefix is deliberately ignored:
|
||
/// it is a different mode's key, and honouring it would silently narrow the window to a
|
||
/// prefix that cannot match a Sparkplug topic at all.
|
||
/// </para>
|
||
/// </summary>
|
||
/// <param name="options">The browse configuration.</param>
|
||
/// <returns>The MQTT topic filter to subscribe.</returns>
|
||
internal static string BuildRootFilter(MqttDriverOptions options)
|
||
{
|
||
if (options.Mode == MqttMode.SparkplugB)
|
||
{
|
||
var groupId = (options.Sparkplug?.GroupId ?? string.Empty).Trim();
|
||
return groupId.Length == 0 ? "spBv1.0/#" : $"spBv1.0/{groupId}/#";
|
||
}
|
||
|
||
var prefix = (options.Plain?.TopicPrefix ?? string.Empty).Trim();
|
||
if (prefix.Length == 0) return "#";
|
||
if (!prefix.EndsWith('/')) prefix += "/";
|
||
return prefix + "#";
|
||
}
|
||
|
||
/// <summary>
|
||
/// A unique per-session client-id suffix. A broker disconnects an existing client when a new
|
||
/// one CONNECTs with the <i>same</i> client id, so sharing the driver's id would knock the
|
||
/// live driver offline every time an operator opened the picker.
|
||
/// </summary>
|
||
/// <returns>The suffix appended to the configured client id.</returns>
|
||
internal static string BuildBrowseClientIdSuffix() =>
|
||
BrowseClientIdPrefix + Guid.NewGuid().ToString("N")[..8];
|
||
|
||
/// <summary>
|
||
/// Projects the authored options into the ones the browse CONNECT uses. Clean session is
|
||
/// forced: a transient browse identity must not leave queued-message state behind on the
|
||
/// broker after the picker closes.
|
||
/// <para>
|
||
/// ⚠ <b>THE landmine — a Last Will would break the read-only guarantee from outside it.</b>
|
||
/// <c>MqttDriverOptions</c> (and <c>MqttSparkplugOptions</c>) carry no will, <b>re-verified
|
||
/// since Sparkplug browse was unsealed</b>, so there is nothing to strip and the
|
||
/// projection stays a single <c>CleanSession</c> override. If a will is ever added — NDEATH
|
||
/// <i>is</i> a will message — it MUST be cleared here: a will is published by the
|
||
/// <i>broker</i>, not by us, so it is invisible to
|
||
/// <see cref="MqttBrowseSession.PublishCountForTest"/>, and any ungraceful end to a browse
|
||
/// session (crash, dropped link, the disconnect deadline expiring) would then fire an
|
||
/// NDEATH under the plant's own edge-node identity — killing a live Sparkplug node because
|
||
/// an operator opened an address picker. <c>MqttBrowseSessionTests</c>'
|
||
/// <c>MqttDriverOptions_CarryNothingWillShaped_…</c> is the reflection guard that turns that
|
||
/// "there is nothing to strip" into a checked fact rather than a claim.
|
||
/// </para>
|
||
/// </summary>
|
||
/// <param name="options">The authored configuration.</param>
|
||
/// <returns>The configuration the browse CONNECT is built from.</returns>
|
||
internal static MqttDriverOptions ToBrowseOptions(MqttDriverOptions options) =>
|
||
options with { CleanSession = true };
|
||
|
||
/// <summary>
|
||
/// Whole-open bound (connect + subscribe), clamped to
|
||
/// <see cref="MinOpenBudgetSeconds"/>–<see cref="MaxOpenBudgetSeconds"/> around the config's
|
||
/// own connect deadline.
|
||
/// </summary>
|
||
/// <param name="options">The browse configuration.</param>
|
||
/// <returns>The clamped open budget.</returns>
|
||
internal static TimeSpan OpenBudget(MqttDriverOptions options) =>
|
||
TimeSpan.FromSeconds(Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds));
|
||
}
|