diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx index e07dc176..02a8700d 100644 --- a/ZB.MOM.WW.OtOpcUa.slnx +++ b/ZB.MOM.WW.OtOpcUa.slnx @@ -44,6 +44,7 @@ + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs new file mode 100644 index 00000000..eca93d61 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs @@ -0,0 +1,365 @@ +using System.Collections.Concurrent; +using System.Globalization; +using System.Text; +using Microsoft.Extensions.Logging; +using MQTTnet; +using ZB.MOM.WW.OtOpcUa.Commons.Browsing; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; + +/// +/// A passive MQTT observation window, served through the AdminUI's standard browse-session +/// interface. +/// +/// MQTT has no discovery protocol — a broker cannot be asked "what topics exist". The only way +/// to learn the topic space is to subscribe to a wildcard and watch what arrives. So this +/// session accumulates every topic it observes into a segment tree (split on /) and +/// serves that tree one level at a time. It is therefore inherently incomplete: a topic +/// that never publishes during the window is invisible. That is a property of MQTT, not a +/// defect — the picker always keeps manual entry as the escape hatch. +/// +/// +/// The load-bearing safety property is that browsing publishes nothing. A picker opened +/// by an operator runs against a live production broker; if browse published anything, +/// opening an address picker could inject messages into a running plant. Every outbound +/// message must route through the single seam, which counts its +/// calls into so the guarantee is assertable. In P1 nothing +/// calls it; P2's operator-triggered Sparkplug rebirth (Task 23) is the one intended exception. +/// +/// +/// Observations arrive on MQTTnet's dispatcher thread while the picker calls +/// from the UI, so the tree is built from concurrent collections and +/// every mutation is lock-free. +/// +/// +internal sealed class MqttBrowseSession : IBrowseSession +{ + /// Default cap on recorded nodes; a busy broker's # firehose is unbounded. + internal const int DefaultNodeCap = 50_000; + + /// Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet. + internal const int PayloadSnippetMaxChars = 64; + + /// Sentinel node id of the "results truncated" marker, mirroring CapturedTreeBrowseSession. + internal const string TruncatedNodeId = "__truncated__"; + + /// Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker. + private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5); + + private readonly IMqttClient? _client; + private readonly ILogger? _logger; + private readonly int _nodeCap; + + /// Flat path → node index, so is O(1) rather than a tree walk. + private readonly ConcurrentDictionary _byPath = new(StringComparer.Ordinal); + + /// First-segment nodes, keyed by segment. + private readonly ConcurrentDictionary _roots = new(StringComparer.Ordinal); + + private int _disposed; + private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks; + private int _nodeCount; + private int _publishCount; + private int _truncated; + + /// + /// Test seam: an unconnected session whose tree is fed through + /// . The accumulating tree is the whole of the P1 logic, so it + /// is exercised without a live broker. + /// + /// The ingest shape the session was opened for. + /// Cap on recorded nodes; defaults to . + internal MqttBrowseSession(MqttMode mode, int nodeCap = DefaultNodeCap) + : this(mode, client: null, logger: null, nodeCap) { } + + /// Production constructor: takes ownership of an already-connected client. + /// The ingest shape the session was opened for. + /// The connected MQTT client; disconnected and disposed by . + /// Optional logger; never receives credentials or payload bodies. + /// Cap on recorded nodes; defaults to . + internal MqttBrowseSession(MqttMode mode, IMqttClient? client, ILogger? logger, int nodeCap = DefaultNodeCap) + { + Mode = mode; + _client = client; + _logger = logger; + _nodeCap = nodeCap; + } + + /// + /// Number of messages this session has published. Always 0 in P1 — the assertion that + /// browsing a live plant broker is strictly read-only. + /// + internal int PublishCountForTest => Volatile.Read(ref _publishCount); + + /// True once the node cap stopped recording; surfaced as a marker node from . + internal bool Truncated => Volatile.Read(ref _truncated) != 0; + + private bool Disposed => Volatile.Read(ref _disposed) != 0; + + /// + /// The ingest shape this session was opened for. P1 serves only; + /// the P2 Sparkplug tasks branch the tree shape (Group → EdgeNode → Device → Metric) on it. + /// + internal MqttMode Mode { get; } + + /// + public Guid Token { get; } = Guid.NewGuid(); + + /// + public DateTime LastUsedUtc => new(Interlocked.Read(ref _lastUsedTicksUtc), DateTimeKind.Utc); + + /// + public Task> RootAsync(CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(Disposed, this); + var nodes = Project(_roots); + if (Truncated) + { + nodes.Add(new BrowseNode( + TruncatedNodeId, + "⚠ Observation truncated — narrow the topic prefix or use manual entry", + BrowseNodeKind.Folder, + HasChildrenHint: false)); + } + + Touch(); + return Task.FromResult>(nodes); + } + + /// + /// + /// is the full slash-joined topic path of the node + /// (e.g. factory/line3/oven) — the same value carried on the + /// the picker was handed. An unknown path returns empty rather than throwing: the tree grows + /// while the picker browses it, so a stale id is an ordinary race, not an error. + /// + public Task> ExpandAsync(string nodeId, CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(Disposed, this); + var children = _byPath.TryGetValue(nodeId ?? "", out var node) + ? Project(node.Children) + : []; + Touch(); + return Task.FromResult>(children); + } + + /// + /// + /// A topic has no attribute model — MQTT carries an opaque payload. The single synthetic + /// attribute returned here exists to help the operator confirm they picked the right topic: + /// carries the type inferred from the last + /// observed payload (the authored tag's own dataType remains the authority — inference + /// is the brittle fallback, per design §4), and + /// carries that payload's snippet. The SecurityClass slot is reused deliberately: it is + /// the picker's free-text descriptor line (rendered as + /// {DriverDataType} · {SecurityClass}) and MQTT has no security classes of its own. + /// Returns empty for a synthesised path segment nothing was ever published to. + /// + public Task> AttributesAsync(string nodeId, CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(Disposed, this); + IReadOnlyList attrs = + _byPath.TryGetValue(nodeId ?? "", out var node) && node.Observed is { } observed + ? [new AttributeInfo(node.Segment, observed.DataType.ToString(), IsArray: false, observed.Snippet)] + : []; + Touch(); + return Task.FromResult(attrs); + } + + /// + /// Best-effort teardown. Safe on a session that was never opened (test seam, or an + /// OpenAsync that failed between constructing the client and connecting it) and + /// idempotent under the reaper racing the picker's own close. + /// + /// A task that represents the asynchronous teardown. + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; + + if (_client is { } client) + { + try + { + using var cts = new CancellationTokenSource(DisconnectTimeout); + await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + // The broker may already be gone, or never have answered the CONNECT at all. + _logger?.LogDebug(ex, "MQTT browse session disconnect failed; client abandoned."); + } + + try { client.Dispose(); } + catch (Exception ex) { _logger?.LogDebug(ex, "MQTT browse client dispose failed."); } + } + + _byPath.Clear(); + _roots.Clear(); + } + + /// + /// Records one observed topic (and its payload) into the accumulating tree. Called from + /// MQTTnet's dispatcher thread, so it does no I/O, takes no lock, and never throws — a + /// throwing handler would stall the client's pump. + /// + /// The topic the message arrived on. + /// The message body; used only for type inference and the display snippet. + internal void Observe(string topic, ReadOnlySpan payload) + { + if (Disposed || string.IsNullOrWhiteSpace(topic)) return; + + var segments = topic.Split('/'); + var level = _roots; + TopicNode? node = null; + var path = string.Empty; + + foreach (var segment in segments) + { + path = path.Length == 0 ? segment : path + "/" + segment; + + if (!level.TryGetValue(segment, out var child)) + { + if (Volatile.Read(ref _nodeCount) >= _nodeCap) + { + Volatile.Write(ref _truncated, 1); + return; // stop recording; already-recorded topics keep serving + } + + var created = new TopicNode(path, segment); + child = level.GetOrAdd(segment, created); + if (ReferenceEquals(child, created)) + { + Interlocked.Increment(ref _nodeCount); + _byPath[path] = child; + } + } + + node = child; + level = child.Children; + } + + node?.Record(InferPayload(payload)); + } + + /// Test seam over that takes the payload as text. + /// The observed topic. + /// The message body as text, or for an empty payload. + internal void ObserveTopicForTest(string topic, string? payload = null) => + Observe(topic, payload is null ? ReadOnlySpan.Empty : Encoding.UTF8.GetBytes(payload)); + + /// + /// The one and only outbound-message seam. Nothing in P1 calls it — browse against a + /// live plant broker is strictly read-only, and is the + /// assertion of that. P2's Sparkplug RequestRebirthAsync (Task 23) is the single + /// explicitly-operator-triggered exception and must route through here rather than + /// touching directly, or the guarantee stops being checkable. + /// + /// The message to publish. + /// Cancellation token for the publish. + /// A task that represents the asynchronous publish. + private async Task PublishAsync(MqttApplicationMessage message, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _publishCount); + if (_client is { } client) + { + await client.PublishAsync(message, cancellationToken).ConfigureAwait(false); + } + } + + /// + /// Best-effort type inference from a payload body. Explicit authoring stays the authority + /// (design §4) — this only decorates the picker. Non-UTF-8 bodies are reported as opaque. + /// + /// The raw message body. + /// The inferred type and a bounded display snippet. + private static ObservedValue InferPayload(ReadOnlySpan payload) + { + if (payload.IsEmpty) return new ObservedValue(DriverDataType.String, "(empty)"); + + string text; + try + { + // Throw-on-invalid so genuinely binary payloads are reported as such rather than + // silently rendered as a field of replacement characters. + text = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true) + .GetString(payload); + } + catch (DecoderFallbackException) + { + return new ObservedValue(DriverDataType.String, $"({payload.Length} bytes, binary)"); + } + + var trimmed = text.Trim(); + var snippet = trimmed.Length > PayloadSnippetMaxChars + ? string.Concat(trimmed.AsSpan(0, PayloadSnippetMaxChars), "…") + : trimmed; + + // Ordered narrowest-first. "1" infers Int64 rather than Boolean: an integer reading is the + // commoner meaning and, unlike the reverse, does not silently collapse a range to two states. + var dataType = trimmed switch + { + _ when bool.TryParse(trimmed, out _) => DriverDataType.Boolean, + _ when long.TryParse(trimmed, NumberStyles.Integer, CultureInfo.InvariantCulture, out _) + => DriverDataType.Int64, + _ when double.TryParse(trimmed, NumberStyles.Float, CultureInfo.InvariantCulture, out _) + => DriverDataType.Float64, + _ => DriverDataType.String, + }; + + return new ObservedValue(dataType, snippet); + } + + /// Projects one level of the tree into browse nodes, ordered so the picker is stable. + /// The child map to project. + /// The projected nodes, sorted by segment. + private static List Project(ConcurrentDictionary level) => + level.Values + .OrderBy(n => n.Segment, StringComparer.Ordinal) + .Select(n => + { + // A childless node is always the terminal segment of a topic that actually published, + // so it is the committable address. A node that both published and has children below + // it renders as a folder — MQTT allows that shape and the picker must stay expandable. + var hasChildren = !n.Children.IsEmpty; + return new BrowseNode( + n.Path, + n.Segment, + hasChildren ? BrowseNodeKind.Folder : BrowseNodeKind.Leaf, + hasChildren); + }) + .ToList(); + + private void Touch() => Interlocked.Exchange(ref _lastUsedTicksUtc, DateTime.UtcNow.Ticks); + + /// The most recent payload observed on a topic, published as one atomic reference. + /// The type inferred from that payload. + /// A bounded, display-safe rendering of that payload. + private sealed record ObservedValue(DriverDataType DataType, string Snippet); + + /// One segment of the observed topic space. + private sealed class TopicNode(string path, string segment) + { + /// Full slash-joined topic path — the node id the picker round-trips. + public string Path { get; } = path; + + /// This node's own topic segment — the display label. + public string Segment { get; } = segment; + + /// Child segments, keyed by segment name. + public ConcurrentDictionary Children { get; } = new(StringComparer.Ordinal); + + /// + /// The last payload observed on exactly this topic, or when this is + /// a synthesised intermediate segment nothing published to. Written by the dispatcher + /// thread and read by the UI thread, so the whole record is swapped as one reference. + /// + public ObservedValue? Observed => Volatile.Read(ref _observed); + + private ObservedValue? _observed; + + /// Publishes the latest observation for this topic. + /// The inferred type + snippet to record. + public void Record(ObservedValue value) => Volatile.Write(ref _observed, value); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs new file mode 100644 index 00000000..76194cab --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs @@ -0,0 +1,172 @@ +using System.Buffers; +using System.Text.Json; +using System.Text.Json.Serialization; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using MQTTnet; +using MQTTnet.Protocol; +using ZB.MOM.WW.OtOpcUa.Commons.Browsing; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; + +/// +/// Bespoke MQTT address-picker browser: opens a short-lived, strictly passive observation +/// window against the broker named by the form's JSON. +/// +/// MQTT has no discovery protocol, so there is nothing to enumerate — the browser subscribes +/// to the wildcard (#, or {topicPrefix}/#) and lets +/// accumulate whatever arrives. Nothing on any browse path +/// publishes: an operator opening a picker must not be able to inject a message into a +/// running plant. +/// +/// +public sealed class MqttDriverBrowser : IDriverBrowser +{ + /// Floor on the open budget — a 1 s ConnectTimeoutSeconds would make browse unusable. + internal const int MinOpenBudgetSeconds = 5; + + /// Ceiling on the open budget — the picker must never hang on an unreachable broker. + internal const int MaxOpenBudgetSeconds = 30; + + /// Marks the transient browse identity in the broker's client-id/session logs. + internal const string BrowseClientIdPrefix = "-browse-"; + + /// + /// Matches the runtime driver factory's parsing so a given DriverConfig JSON means the same + /// thing to the browser as it does to the deployed driver. JsonStringEnumConverter lets + /// mode / protocolVersion be authored as their string names — the natural form + /// for AdminUI-emitted JSON — while still accepting numeric ordinals. + /// + private static readonly JsonSerializerOptions JsonOpts = new() + { + UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip, + PropertyNameCaseInsensitive = true, + Converters = { new JsonStringEnumConverter() }, + }; + + private readonly ILogger _logger; + + /// + /// Creates a browser. Connection-free by contract — the universal browser's + /// CanBrowse constructs a throwaway instance purely to ask which driver type it + /// handles, so the constructor must never touch the network. + /// + /// Optional logger; defaults to . Never receives credentials. + public MqttDriverBrowser(ILogger? logger = null) => + _logger = logger ?? NullLogger.Instance; + + /// + // 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": a DriverType string that drifts from the + // persisted one silently breaks driver dispatch (the repo's ModbusTcp/Modbus incident). + public string DriverType => "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 + /// ; nothing here publishes. + /// + public async Task OpenAsync(string configJson, CancellationToken cancellationToken) + { + var opts = JsonSerializer.Deserialize(configJson, JsonOpts) + ?? throw new InvalidOperationException("Mqtt options deserialized to null."); + + if (opts.Mode == MqttMode.SparkplugB) + { + // P2 seam (Task 23): Sparkplug browse is a Group → EdgeNode → Device → Metric tree built + // from observed NBIRTH/DBIRTH payloads, with an operator-triggered rebirth request as the + // one sanctioned publish. Failing fast beats serving a raw spBv1.0 topic tree that looks + // like a metric tree but is not one. + throw new NotSupportedException( + "Sparkplug B browse (Group → EdgeNode → Device → Metric) is not available yet; " + + "the MQTT browser currently observes plain topics only. Author Sparkplug tags manually."); + } + + var filter = BuildRootFilter(opts); + var suffix = BuildBrowseClientIdSuffix(); + var clientOptions = MqttConnection.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 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} observing '{Filter}' (read-only).", + opts.Host, opts.Port, filter); + + return session; + } + catch + { + await session.DisposeAsync().ConfigureAwait(false); // owns the client — disconnects + disposes + throw; + } + } + + /// + /// The wildcard the observation window subscribes to: the whole broker (#) unless the + /// config scopes it with a Plain-mode topic prefix. + /// + /// The browse configuration. + /// The MQTT topic filter to subscribe. + internal static string BuildRootFilter(MqttDriverOptions options) + { + var prefix = (options.Plain?.TopicPrefix ?? string.Empty).Trim(); + if (prefix.Length == 0) return "#"; + if (!prefix.EndsWith('/')) prefix += "/"; + return prefix + "#"; + } + + /// + /// A unique per-session client-id suffix. A broker disconnects an existing client when a new + /// one CONNECTs with the same client id, so sharing the driver's id would knock the + /// live driver offline every time an operator opened the picker. + /// + /// The suffix appended to the configured client id. + internal static string BuildBrowseClientIdSuffix() => + BrowseClientIdPrefix + Guid.NewGuid().ToString("N")[..8]; + + /// + /// 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. + /// + /// The authored configuration. + /// The configuration the browse CONNECT is built from. + internal static MqttDriverOptions ToBrowseOptions(MqttDriverOptions options) => + options with { CleanSession = true }; + + /// + /// Whole-open bound (connect + subscribe), clamped to + /// – around the config's + /// own connect deadline. + /// + /// The browse configuration. + /// The clamped open budget. + internal static TimeSpan OpenBudget(MqttDriverOptions options) => + TimeSpan.FromSeconds(Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds)); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj new file mode 100644 index 00000000..bdc8e8c6 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj @@ -0,0 +1,36 @@ + + + + net10.0 + enable + enable + latest + true + ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser + ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser + + + + + + + + + + + + + + + + + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs new file mode 100644 index 00000000..b2a309c9 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs @@ -0,0 +1,321 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Browsing; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Covers the passive #-observation browse session and the browser that opens it. The +/// session is exercised through its test seam (ObserveTopicForTest) rather than a live +/// broker: the accumulating tree is the whole of the P1 logic, and the MQTTnet plumbing that +/// feeds it is a one-line handler. +/// +public sealed class MqttBrowseSessionTests +{ + // ---------------------------------------------------------------- tree shape + + [Fact] + public async Task ExpandAsync_BuildsTopicSegmentTree_FromObservedTopics() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/line3/oven/temp"); + + var root = await s.RootAsync(TestContext.Current.CancellationToken); + root.ShouldContain(n => n.DisplayName == "factory"); + + var lvl2 = await s.ExpandAsync("factory", TestContext.Current.CancellationToken); + lvl2.ShouldContain(n => n.DisplayName == "line3"); + + var lvl3 = await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken); + lvl3.ShouldContain(n => n.DisplayName == "oven"); + + var lvl4 = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken); + var leaf = lvl4.ShouldHaveSingleItem(); + leaf.DisplayName.ShouldBe("temp"); + leaf.NodeId.ShouldBe("factory/line3/oven/temp"); + leaf.Kind.ShouldBe(BrowseNodeKind.Leaf); + leaf.HasChildrenHint.ShouldBeFalse(); + } + + [Fact] + public async Task ObservedTopics_AccumulateAcrossMessages_AndDoNotDuplicate() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/line3/oven/temp"); + s.ObserveTopicForTest("factory/line3/oven/temp"); // repeat — must not duplicate + s.ObserveTopicForTest("factory/line3/oven/pressure"); + s.ObserveTopicForTest("factory/line4/mixer/rpm"); + s.ObserveTopicForTest("plant/power"); + + var root = await s.RootAsync(TestContext.Current.CancellationToken); + root.Select(n => n.DisplayName).ShouldBe(["factory", "plant"]); // sorted, deduped + + var lines = await s.ExpandAsync("factory", TestContext.Current.CancellationToken); + lines.Select(n => n.DisplayName).ShouldBe(["line3", "line4"]); + + var oven = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken); + oven.Select(n => n.DisplayName).ShouldBe(["pressure", "temp"]); + } + + [Fact] + public async Task IntermediateNode_IsFolder_WithChildrenHint() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/line3/oven/temp"); + + var root = await s.RootAsync(TestContext.Current.CancellationToken); + var factory = root.ShouldHaveSingleItem(); + factory.Kind.ShouldBe(BrowseNodeKind.Folder); + factory.HasChildrenHint.ShouldBeTrue(); + } + + [Fact] + public async Task ExpandAsync_UnknownNode_ReturnsEmpty() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/line3"); + + var children = await s.ExpandAsync("nope/not/here", TestContext.Current.CancellationToken); + children.ShouldBeEmpty(); + } + + [Fact] + public async Task RootAsync_BeforeAnyTraffic_IsEmpty() + { + // MQTT has no discovery protocol: an observation window that has seen nothing shows nothing. + await using var s = new MqttBrowseSession(MqttMode.Plain); + (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty(); + } + + // ---------------------------------------------------------------- read-only guarantee + + [Fact] + public async Task BrowseCalls_PublishNothing() // browse is read-only + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/line3/oven/temp", "23.5"); + + await s.RootAsync(TestContext.Current.CancellationToken); + await s.ExpandAsync("factory", TestContext.Current.CancellationToken); + await s.AttributesAsync("factory/line3/oven/temp", TestContext.Current.CancellationToken); + + s.PublishCountForTest.ShouldBe(0); + } + + // ---------------------------------------------------------------- attributes + + [Theory] + [InlineData("23.5", nameof(Core.Abstractions.DriverDataType.Float64))] + [InlineData("42", nameof(Core.Abstractions.DriverDataType.Int64))] + [InlineData("true", nameof(Core.Abstractions.DriverDataType.Boolean))] + [InlineData("{\"v\":1}", nameof(Core.Abstractions.DriverDataType.String))] + public async Task AttributesAsync_InfersTypeFromLastPayload(string payload, string expectedType) + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/oven/temp", payload); + + var attrs = await s.AttributesAsync("factory/oven/temp", TestContext.Current.CancellationToken); + var a = attrs.ShouldHaveSingleItem(); + a.Name.ShouldBe("temp"); + a.DriverDataType.ShouldBe(expectedType); + a.IsArray.ShouldBeFalse(); + a.IsAlarm.ShouldBeFalse(); + a.SecurityClass.ShouldContain(payload); // the last-payload snippet + } + + [Fact] + public async Task AttributesAsync_LongPayload_IsTruncatedToASnippet() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("t/long", new string('x', 500)); + + var a = (await s.AttributesAsync("t/long", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1); + } + + [Fact] + public async Task AttributesAsync_UnobservedFolder_ReturnsEmpty() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("factory/oven/temp", "1"); + + // "factory" is a synthesised path segment — nothing was ever published to it. + (await s.AttributesAsync("factory", TestContext.Current.CancellationToken)).ShouldBeEmpty(); + (await s.AttributesAsync("no/such/node", TestContext.Current.CancellationToken)).ShouldBeEmpty(); + } + + // ---------------------------------------------------------------- concurrency + + [Fact] + public async Task ObserveDuringBrowse_IsThreadSafe() + { + // Messages arrive on MQTTnet's dispatcher thread while the picker calls ExpandAsync. + await using var s = new MqttBrowseSession(MqttMode.Plain); + using var stop = new CancellationTokenSource(); + + var observed = 0; + var writer = Task.Run(() => + { + var i = 0; + while (!stop.Token.IsCancellationRequested) + { + s.ObserveTopicForTest($"factory/line{i % 8}/dev{i % 32}/m{i}", i.ToString()); + Interlocked.Increment(ref observed); + i++; + } + }, CancellationToken.None); + + // The read loop is pure in-memory work and would otherwise finish before the writer thread + // is even scheduled, leaving the two never actually overlapping. + while (Volatile.Read(ref observed) == 0) await Task.Yield(); + + for (var i = 0; i < 400; i++) + { + await s.RootAsync(TestContext.Current.CancellationToken); + await s.ExpandAsync("factory", TestContext.Current.CancellationToken); + await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken); + } + + await stop.CancelAsync(); + await writer; // rethrows anything the observer thread hit + + (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldNotBeEmpty(); + } + + // ---------------------------------------------------------------- lifetime + + [Fact] + public async Task LastUsedUtc_AdvancesOnEveryCall() + { + var s = new MqttBrowseSession(MqttMode.Plain); + var before = s.LastUsedUtc; + await Task.Delay(15, TestContext.Current.CancellationToken); + + await s.RootAsync(TestContext.Current.CancellationToken); + s.LastUsedUtc.ShouldBeGreaterThan(before); + + var afterRoot = s.LastUsedUtc; + await Task.Delay(15, TestContext.Current.CancellationToken); + await s.ExpandAsync("x", TestContext.Current.CancellationToken); + s.LastUsedUtc.ShouldBeGreaterThan(afterRoot); + + await s.DisposeAsync(); + } + + [Fact] + public async Task DisposeAsync_OnNeverOpenedSession_IsIdempotentAndDoesNotThrow() + { + var s = new MqttBrowseSession(MqttMode.Plain); + await Should.NotThrowAsync(async () => await s.DisposeAsync()); + await Should.NotThrowAsync(async () => await s.DisposeAsync()); + } + + [Fact] + public async Task AfterDispose_BrowseCallsThrowObjectDisposed() + { + var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("a/b"); + await s.DisposeAsync(); + + await Should.ThrowAsync( + async () => await s.RootAsync(TestContext.Current.CancellationToken)); + await Should.ThrowAsync( + async () => await s.ExpandAsync("a", TestContext.Current.CancellationToken)); + await Should.ThrowAsync( + async () => await s.AttributesAsync("a/b", TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task ObserveAfterDispose_IsIgnored() + { + // A message already in MQTTnet's dispatcher when the reaper disposes the session. + var s = new MqttBrowseSession(MqttMode.Plain); + await s.DisposeAsync(); + Should.NotThrow(() => s.ObserveTopicForTest("late/arrival", "1")); + } + + // ---------------------------------------------------------------- node cap + + [Fact] + public async Task NodeCap_StopsRecording_AndSurfacesATruncationMarker() + { + await using var s = new MqttBrowseSession(MqttMode.Plain, nodeCap: 4); + s.ObserveTopicForTest("a/b/c"); // 3 nodes + s.ObserveTopicForTest("d/e/f"); // would take it past 4 + + var root = await s.RootAsync(TestContext.Current.CancellationToken); + root.ShouldContain(n => n.NodeId == MqttBrowseSession.TruncatedNodeId); + } + + // ---------------------------------------------------------------- the browser + + [Fact] + public void Ctor_IsConnectionFree_AndReportsTheMqttDriverType() + { + // The universal-browser CanBrowse pattern constructs a throwaway instance purely to ask + // what driver type it handles — the ctor must never touch the network. + var browser = new MqttDriverBrowser(); + browser.DriverType.ShouldBe("Mqtt"); + } + + [Fact] + public async Task OpenAsync_SparkplugMode_FailsFastWithoutConnecting() + { + var browser = new MqttDriverBrowser(); + const string cfg = """{"host":"broker.invalid","port":8883,"mode":"SparkplugB"}"""; + + await Should.ThrowAsync( + async () => await browser.OpenAsync(cfg, TestContext.Current.CancellationToken)); + } + + [Theory] + [InlineData(null, "#")] + [InlineData("", "#")] + [InlineData(" ", "#")] + [InlineData("factory", "factory/#")] + [InlineData("factory/", "factory/#")] + [InlineData("factory/line3", "factory/line3/#")] + public void BuildRootFilter_ScopesTheWildcardToTheConfiguredPrefix(string? prefix, string expected) + { + var opts = new MqttDriverOptions + { + Mode = MqttMode.Plain, + Plain = prefix is null ? null : new MqttPlainOptions { TopicPrefix = prefix }, + }; + MqttDriverBrowser.BuildRootFilter(opts).ShouldBe(expected); + } + + [Fact] + public void BrowseClientIdSuffix_IsUnique_AndIsAppliedToTheConnectClientId() + { + // A broker disconnects an existing client when a new one CONNECTs with the same id — + // browsing must never knock the running driver offline. + var a = MqttDriverBrowser.BuildBrowseClientIdSuffix(); + var b = MqttDriverBrowser.BuildBrowseClientIdSuffix(); + a.ShouldNotBe(b); + a.ShouldStartWith("-browse-"); + a.Length.ShouldBe("-browse-".Length + 8); + + var opts = new MqttDriverOptions { ClientId = "otopcua-node1" }; + MqttConnection.BuildClientOptions(opts, a).ClientId.ShouldBe("otopcua-node1" + a); + } + + [Fact] + public void BrowseConnect_ForcesACleanSession() + { + // A transient browse identity must not leave queued-message state behind on the broker. + var opts = new MqttDriverOptions { CleanSession = false }; + MqttDriverBrowser.ToBrowseOptions(opts).CleanSession.ShouldBeTrue(); + } + + [Theory] + [InlineData(1, 5)] // clamped up + [InlineData(15, 15)] + [InlineData(600, 30)] // clamped down + public void OpenBudget_IsClampedTo5To30Seconds(int configured, int expected) + { + var opts = new MqttDriverOptions { ConnectTimeoutSeconds = configured }; + MqttDriverBrowser.OpenBudget(opts).ShouldBe(TimeSpan.FromSeconds(expected)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj index 2ad93142..f1a88765 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj @@ -22,6 +22,7 @@ +