diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index 02f81e7d..1ed55064 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -1,11 +1,67 @@ +using System.Buffers; using System.Net.Security; using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; using Microsoft.Extensions.Logging; using MQTTnet; +using MQTTnet.Protocol; namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; +/// +/// Receives one inbound MQTT application message. Invoked on MQTTnet's own dispatcher +/// thread: an implementation must not block, must not do I/O and must not throw, or it stalls +/// the client's pump for every other subscription. +/// +/// The concrete topic the message arrived on (never a filter). +/// +/// The message body. Only valid for the duration of the call — the underlying buffer belongs to +/// MQTTnet, so anything an implementation wants to keep must be copied out. +/// +/// +/// The message's MQTT retain flag as delivered to this client, i.e. true for +/// the broker's stored last-known value replayed at subscribe time, and false for an +/// ordinary live publish. This is the retained seed signal. +/// +public delegate void MqttMessageObserver(string topic, ReadOnlySpan payload, bool retained); + +/// One topic filter to SUBSCRIBE, as the subscription manager wants it established. +/// The topic filter (concrete, or carrying MQTT wildcards). +/// Requested QoS, 0–2. +/// +/// Whether the broker should replay its retained message for this filter at subscribe time. +/// Honoured natively on MQTT 5.0 (retain handling); on 3.1.1 the broker always replays and the +/// manager drops the seed client-side instead. +/// +public sealed record MqttTopicSubscription(string Topic, int Qos, bool SeedRetained); + +/// The broker's SUBACK verdict for one requested filter. +/// The filter this outcome answers. +/// Whether the broker granted the subscription. +/// The broker's reason code / string, for logs and diagnostics. +public sealed record MqttSubscribeOutcome(string Topic, bool Granted, string Reason); + +/// +/// The one seam through which the subscription manager establishes MQTT subscriptions. +/// is the production implementation; the interface exists so the +/// manager's SUBACK handling (per-filter grant / rejection, total failure) is exercisable without +/// a broker. +/// +public interface IMqttSubscribeTransport +{ + /// + /// Issues one SUBSCRIBE carrying every filter and returns the broker's per-filter verdict. + /// Implementations must be bounded — a broker that accepts SUBSCRIBE and never SUBACKs must + /// fail at a deadline, not hang. + /// + /// The filters to establish; never empty. + /// Caller cancellation, linked with the implementation's deadline. + /// One outcome per requested filter. + Task> SubscribeAsync( + IReadOnlyList filters, + CancellationToken cancellationToken); +} + /// Lifecycle state of an . public enum MqttConnectionState { @@ -144,10 +200,10 @@ public enum MqttConnectionState /// socket no later dispose could ever reach. Now the losing side of that race disposes the /// client it created and reports . /// -/// Subscription (Task 6) and Sparkplug rebirth-on-reconnect (Task 21, which hangs off -/// ) are deliberately not implemented here. +/// Sparkplug rebirth-on-reconnect (Task 21, which hangs off ) is +/// deliberately not implemented here. /// -public sealed class MqttConnection : IAsyncDisposable +public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport { private readonly string _driverId; @@ -217,6 +273,19 @@ public sealed class MqttConnection : IAsyncDisposable /// public event Func? Reconnected; + /// + /// Raised for every inbound application message, on MQTTnet's own dispatcher thread. This + /// connection does no routing, parsing or typing of its own — that is the subscription + /// manager's job; here the message is only stamped onto and + /// handed on. + /// + /// + /// Handlers must not block, do I/O or throw. A throwing handler is caught and logged here + /// rather than being allowed to escape into the library's pump — one bad tag must not stop + /// delivery for every other subscriber. + /// + public event MqttMessageObserver? MessageReceived; + /// Whether the underlying client currently holds an established MQTT session. public bool IsConnected => _client?.IsConnected ?? false; @@ -572,20 +641,28 @@ public sealed class MqttConnection : IAsyncDisposable } /// - /// Maps a connect failure onto this type's documented contract. MQTTnet wraps a cancelled - /// connect in MqttConnectingFailedException rather than letting the + /// Maps a bounded-operation failure onto this type's documented contract. MQTTnet wraps a + /// cancelled connect in MqttConnectingFailedException rather than letting the /// surface, so the legs are told apart by which /// token fired, not by exception type. Precedence: teardown, then caller intent, then the /// deadline. /// - private Exception Classify(Exception ex, CancellationToken cancellationToken, CancellationTokenSource deadline) + /// The failure to classify. + /// The caller's token — cancelled means caller intent. + /// The operation's own deadline source. + /// The operation name for the message ("connect", "subscribe"). + private Exception Classify( + Exception ex, + CancellationToken cancellationToken, + CancellationTokenSource deadline, + string operation = "connect") { if (Disposed) { return new ObjectDisposedException( nameof(MqttConnection), new InvalidOperationException( - $"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} was aborted because the " + $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was aborted because the " + "connection was disposed while the attempt was in flight.", ex)); } @@ -593,7 +670,7 @@ public sealed class MqttConnection : IAsyncDisposable if (cancellationToken.IsCancellationRequested) { return new OperationCanceledException( - $"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} was cancelled.", + $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was cancelled.", ex, cancellationToken); } @@ -601,7 +678,7 @@ public sealed class MqttConnection : IAsyncDisposable if (deadline.IsCancellationRequested) { return new TimeoutException( - $"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} did not complete within " + $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} did not complete within " + $"{_options.ConnectTimeoutSeconds}s.", ex); } @@ -640,9 +717,139 @@ public sealed class MqttConnection : IAsyncDisposable private Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs args) { Interlocked.Exchange(ref _lastMessageTicksUtc, DateTime.UtcNow.Ticks); + + var observers = MessageReceived; + if (observers is null) + { + return Task.CompletedTask; + } + + var message = args.ApplicationMessage; + var payload = message.Payload; + + // Same idiom the browse session uses: the common single-segment case is served straight off + // the library's buffer; a fragmented sequence is flattened once. Either way the span is only + // valid for this call, which is exactly what MqttMessageObserver documents. + ReadOnlySpan body = payload.IsSingleSegment ? payload.FirstSpan : payload.ToArray(); + + try + { + observers(message.Topic, body, message.Retain); + } + catch (Exception ex) + { + // An exception escaping here surfaces inside MQTTnet's own pump. Contain it: a broken + // observer must degrade its own tags, never stop delivery for every other subscription. + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': an inbound-message observer threw; the message was dropped.", + _driverId); + } + return Task.CompletedTask; } + /// + /// Issues one SUBSCRIBE carrying every requested filter, bounded by + /// so a broker that accepts SUBSCRIBE and + /// never answers SUBACK — the frozen-peer shape again — fails at the deadline instead of + /// parking the caller (and, on the reconnect path, the supervisor) forever. + /// + /// + /// The deadline is this method's own linked source, deliberately not + /// : that knob already governs every MQTTnet operation + /// and repurposing it would couple the subscribe budget to the connect budget in a way neither + /// side could change independently. + /// + /// A rejected filter is not an exception: the returned outcomes carry the broker's + /// per-filter verdict so the caller can degrade exactly the affected references. Only a + /// failure of the SUBSCRIBE itself (transport error, deadline, teardown) throws. + /// + /// + /// The filters to establish. + /// Caller cancellation; linked with the subscribe deadline. + /// One outcome per requested filter, in request order. + /// The subscribe deadline elapsed. + /// was cancelled. + /// The connection was disposed. + /// There is no established session to subscribe on. + public async Task> SubscribeAsync( + IReadOnlyList filters, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(filters); + ObjectDisposedException.ThrowIf(Disposed, this); + + if (filters.Count == 0) + { + return []; + } + + var client = _client + ?? throw new InvalidOperationException( + $"MQTT driver '{_driverId}': cannot subscribe before a session to {_options.Host}:{_options.Port} " + + "has been established."); + + var builder = new MqttClientSubscribeOptionsBuilder(); + foreach (var filter in filters) + { + builder = builder.WithTopicFilter(f => f + .WithTopic(filter.Topic) + .WithQualityOfServiceLevel(MapQos(filter.Qos)) + // Only meaningful on MQTT 5.0. On 3.1.1 the broker always replays its retained + // message, so the manager also drops unwanted seeds client-side off the retain flag. + .WithRetainHandling(filter.SeedRetained + ? MqttRetainHandling.SendAtSubscribe + : MqttRetainHandling.DoNotSendOnSubscribe)); + } + + using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds)); + using var linked = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + deadline.Token, + _lifetimeCts.Token); + + MqttClientSubscribeResult result; + try + { + result = await client.SubscribeAsync(builder.Build(), linked.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + throw Classify(ex, cancellationToken, deadline, operation: "subscribe"); + } + + // Pair each requested filter with its SUBACK item positionally: MQTT guarantees SUBACK reason + // codes arrive in the order of the SUBSCRIBE's filters. A short/absent SUBACK (a + // specification-violating broker) is reported as ungranted rather than silently assumed good. + var items = result.Items as IList ?? [.. result.Items]; + var outcomes = new List(filters.Count); + for (var i = 0; i < filters.Count; i++) + { + if (i >= items.Count) + { + outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, Granted: false, "NoSubAckReasonCode")); + continue; + } + + var code = items[i].ResultCode; + var granted = code is MqttClientSubscribeResultCode.GrantedQoS0 + or MqttClientSubscribeResultCode.GrantedQoS1 + or MqttClientSubscribeResultCode.GrantedQoS2; + outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, granted, code.ToString())); + } + + return outcomes; + } + + /// Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value. + private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch + { + <= 0 => MqttQualityOfServiceLevel.AtMostOnce, + 1 => MqttQualityOfServiceLevel.AtLeastOnce, + _ => MqttQualityOfServiceLevel.ExactlyOnce, + }; + /// /// The reconnect loop MQTTnet v5 no longer provides. Started on the first successful connect /// and stopped by cancelling the lifetime token; it owns every diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs new file mode 100644 index 00000000..8e8aa1e9 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -0,0 +1,1038 @@ +using System.Collections.Concurrent; +using System.Collections.Frozen; +using System.Globalization; +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using MQTTnet; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// The plain-MQTT ingest path: holds the driver's authored RawPath → +/// table, establishes one broker subscription per distinct authored +/// topic, and turns each inbound message into notifications plus a +/// update. +/// +/// +/// +/// The published reference is the RawPath. Every this +/// type raises carries — the tag's RawPath — as its +/// , never the topic and never the authored +/// TagConfig blob. DriverHostActor's dual-namespace fan-out is keyed by RawPath; +/// publishing under any other key produces a driver that passes every unit test and delivers +/// nothing at all in production. +/// +/// +/// A message fans out to EVERY matching tag. Several tags routinely share one topic +/// (one JSON document, one JSONPath per signal), so the topic index maps a topic to a +/// list of definitions and delivery walks all of them. Stopping at the first match is +/// the silent-starvation bug this shape exists to prevent, and it is pinned by a test. +/// +/// +/// Nothing here may throw on MQTTnet's dispatcher thread. +/// runs on it: extraction failures degrade the offending +/// tag to a Bad status code, a throwing downstream subscriber +/// is caught (and the tags behind it in the fan-out are still delivered), and no path does I/O +/// or takes a lock. +/// +/// +/// Unauthored topics are ignored. A wildcard-authored tag, or simply a chatty broker, +/// delivers messages this driver never asked for. They are dropped: MQTT ingest never +/// auto-provisions a tag the operator did not author. +/// +/// +/// Wildcards are matched for real. An authored topic carrying + / # is +/// accepted by the parser (warned at deploy, see MqttTagDefinitionFactory.Inspect), so +/// matching uses MQTTnet's own rather than string +/// equality — the same comparer a broker applies. Concrete topics stay a single dictionary +/// lookup; the linear wildcard scan runs only when a wildcard tag actually exists. +/// +/// +/// Sparkplug seam. P1 ingests plain topics only. Sparkplug B (Task 21) decodes a +/// protobuf payload and resolves metrics by the stable +/// (group, node, device, metricName) tuple rather than by topic; it belongs in its own +/// handler alongside this one, feeding the same + +/// sinks. Nothing Sparkplug-shaped is built here. +/// +/// +public sealed class MqttSubscriptionManager +{ + // OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled). + private const uint StatusGood = 0x00000000u; + + /// The payload could not be decoded, or the JSONPath selected nothing. + private const uint StatusBadDecodingError = 0x80070000u; + + /// The payload decoded but does not fit the tag's declared dataType. + private const uint StatusBadTypeMismatch = 0x80740000u; + + /// The subscribed reference is not an authored MQTT tag. + private const uint StatusBadNodeIdUnknown = 0x80340000u; + + /// The broker refused, or never answered, the SUBSCRIBE for this reference's topic. + private const uint StatusBadCommunicationError = 0x80050000u; + + private readonly int _defaultQos; + private readonly string _driverId; + private readonly ILogger? _logger; + private readonly MqttMode _mode; + + /// RawPath → the handle of the subscription currently covering it. + private readonly ConcurrentDictionary _handleByRawPath = new(StringComparer.Ordinal); + + /// Handle → the references it was created for, so unsubscribe can undo exactly its own. + private readonly ConcurrentDictionary _refsByHandle = new(); + + /// Topics already SUBSCRIBEd on the live session — the dedupe set and the re-subscribe set. + private readonly ConcurrentDictionary _subscribedTopics = new(StringComparer.Ordinal); + + /// RawPaths whose extraction failure has already been logged loudly once. + private readonly ConcurrentDictionary _warnedRefs = new(StringComparer.Ordinal); + + private readonly EquipmentTagRefResolver _resolver; + + private long _handleSeq; + + /// The authored table + its topic index, swapped atomically by . + private volatile AuthoredTable _authored = AuthoredTable.Empty; + + private IMqttSubscribeTransport? _transport; + + /// Initializes a new instance of the class. + /// Driver options — supplies the ingest mode and the default QoS. + /// Driver instance id, used only for log correlation. + /// + /// The SUBSCRIBE seam. records subscriptions without dialling a broker + /// (the shape tests use, and the shape a not-yet-connected driver is in); + /// supplies the live connection. + /// + /// Optional logger; never receives credentials or payload bodies. + /// + /// The last-value store backing IReadable. Defaults to a fresh one owned by this + /// manager and exposed as . + /// + public MqttSubscriptionManager( + MqttDriverOptions options, + string driverId, + IMqttSubscribeTransport? transport = null, + ILogger? logger = null, + LastValueCache? cache = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentException.ThrowIfNullOrWhiteSpace(driverId); + + _mode = options.Mode; + _defaultQos = options.Plain?.DefaultQos ?? 1; + _driverId = driverId; + _transport = transport; + _logger = logger; + Values = cache ?? new LastValueCache(); + _resolver = new EquipmentTagRefResolver( + rawPath => _authored.ByRawPath.TryGetValue(rawPath, out var def) ? def : null); + } + + /// + public event EventHandler? OnDataChange; + + /// + /// The last observed value per RawPath — the bridge from MQTT's push model to the OPC UA + /// server's polled IReadable.ReadAsync batches. + /// + public LastValueCache Values { get; } + + /// + /// Wires this manager onto a live connection: inbound messages route to + /// and every reconnect re-establishes the subscriptions through + /// . The connection also becomes this manager's SUBSCRIBE + /// transport. + /// + /// + /// The manager does not own the connection's lifetime and never detaches — the driver disposes + /// both together. Call once per connection, and before the first + /// : a subscribe issued with no transport attached records its + /// intent (and warns) but sends nothing to a broker, and would only be established if the + /// connection later dropped and recovered. + /// + /// The connection to observe. + public void AttachTo(MqttConnection connection) + { + ArgumentNullException.ThrowIfNull(connection); + + _transport = connection; + connection.MessageReceived += HandleMessage; + connection.Reconnected += OnReconnectedAsync; + } + + /// + /// Replaces the authored tag table with the deploy's raw tags, rebuilding the topic index. + /// Wholesale, not additive: a redeploy that drops a tag must stop feeding it. + /// + /// + /// A whose TagConfig does not map is skipped with a warning, + /// never thrown — one bad tag must not fail the whole driver's initialise. Such a reference + /// then resolves to nothing and surfaces BadNodeIdUnknown, which is the stated contract + /// of MqttTagDefinitionFactory.FromTagConfig. + /// + /// The authored raw tags delivered by the deploy artifact. + /// How many entries mapped to a usable definition. + public int Register(IEnumerable rawTags) + { + ArgumentNullException.ThrowIfNull(rawTags); + + var byRawPath = new Dictionary(StringComparer.Ordinal); + foreach (var entry in rawTags) + { + if (MqttTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def)) + { + byRawPath[entry.RawPath] = def; + } + else + { + _logger?.LogWarning( + "MQTT driver '{DriverId}': tag config did not map to a definition; skipping. RawPath={RawPath}", + _driverId, + entry.RawPath); + } + } + + _authored = AuthoredTable.Build(byRawPath); + _warnedRefs.Clear(); + return byRawPath.Count; + } + + /// Resolves a RawPath to its authored definition through the shared v3 resolver. + /// The driver wire reference. + /// The definition, when this returns . + /// when the reference is an authored MQTT tag. + public bool TryResolve(string rawPath, out MqttTagDefinition def) => _resolver.TryResolve(rawPath, out def!); + + /// + /// Subscribes a batch of RawPath references: records them against a new handle and issues one + /// SUBSCRIBE for every distinct authored topic not already established. + /// + /// + /// + /// This is the first subscribe. deliberately + /// does not fire after the initial ConnectAsync, so the manager must issue the + /// opening SUBSCRIBE itself; only re-establishes it later. + /// + /// + /// Never throws, never hangs. A reference that is not an authored tag is recorded as + /// BadNodeIdUnknown; a SUBACK rejection, transport failure or subscribe deadline + /// degrades exactly the affected references to BadCommunicationError. Either way a + /// handle is returned — the OPC UA server's subscribe call must not fail because one topic + /// is ACL-denied. Recovery for a transient failure comes from the reconnect path; + /// a persistent broker-side refusal is visible as a Bad status on the affected nodes. + /// + /// + /// The RawPath references to subscribe. + /// + /// Ignored. MQTT is push-based — the broker's publish rate is the notification rate, and this + /// driver neither polls nor throttles. + /// + /// Cancellation for the SUBSCRIBE round-trip. + /// The handle every resulting is attributed to. + public async Task SubscribeAsync( + IReadOnlyList fullReferences, + TimeSpan publishingInterval, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(fullReferences); + + var defs = new List(fullReferences.Count); + var now = DateTime.UtcNow; + foreach (var reference in fullReferences) + { + if (_resolver.TryResolve(reference, out var def)) + { + defs.Add(def); + continue; + } + + // Not an authored MQTT tag. Say so, rather than leaving it in "waiting for initial data" + // forever — a reference that will never be fed is a different fault from one that has + // simply not been fed yet. + _logger?.LogWarning( + "MQTT driver '{DriverId}': subscribe reference '{RawPath}' is not an authored MQTT tag.", + _driverId, + reference); + Values.Update(reference, new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now)); + } + + var handle = new MqttSubscriptionHandle(BuildDiagnosticId(defs)); + _refsByHandle[handle] = [.. defs.Select(d => d.Name)]; + foreach (var def in defs) + { + _handleByRawPath[def.Name] = handle; + } + + // One SUBSCRIBE per distinct topic, and only for topics not already live: re-SUBSCRIBEing an + // established filter is legal but pointless, and it re-triggers the retained seed. + var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic)).ToList(); + await EstablishAsync(BuildFilters(pending, _defaultQos), cancellationToken).ConfigureAwait(false); + + return handle; + } + + /// + /// Stops attributing notifications to . References another live + /// subscription also covers keep publishing. + /// + /// + /// The broker-side subscription is deliberately left in place: MQTT UNSUBSCRIBE is per-filter, + /// several handles can share one filter, and an idle filter costs only the traffic a topic + /// already produces. The session teardown drops them all. + /// + /// The handle returned by . + /// Unused; present for the ISubscribable shape. + /// A completed task. + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + { + if (handle is not null && _refsByHandle.TryRemove(handle, out var refs)) + { + foreach (var rawPath in refs) + { + // Only clear the mapping if it is still OURS — a later overlapping subscription may + // have taken the reference over, and it must keep publishing. + _handleByRawPath.TryRemove(new KeyValuePair(rawPath, handle)); + } + } + + return Task.CompletedTask; + } + + /// + /// Re-establishes every live subscription after a reconnect. Wired to + /// by . + /// + /// + /// + /// Idempotent — re-SUBSCRIBEing an already-subscribed filter is expected on a + /// persistent session and is harmless; missing one is a healthy-looking client that + /// receives nothing forever. + /// + /// + /// Total failure THROWS, partial rejection does not. This is a deliberate split. + /// 's contract is that a throwing Reconnected subscriber + /// tears the fresh session down and retries under backoff — the correct response to "no + /// subscription was re-established", because the alternative is serving a connected, + /// permanently deaf driver. But if the broker granted some filters and refused others + /// (an ACL denying one topic), tearing the session down would retry forever and take + /// every tag dark to punish one; those references degrade to Bad instead and the + /// rest keep flowing. That is the plan's "SUBACK failure → per-ref Bad, not hang", scoped + /// to the case where it does not cost the whole driver. + /// + /// + /// The connection's lifetime token; cancelled by its dispose. + /// A task that completes when the re-subscribe has been answered. + /// + /// The subscribe failed outright, or every filter was refused — the caller is expected to tear + /// the session down and retry. + /// + public async Task OnReconnectedAsync(CancellationToken cancellationToken) + { + var authored = _authored; + var live = _subscribedTopics.Keys + .Select(topic => authored.ByTopic.TryGetValue(topic, out var defs) ? defs : []) + .SelectMany(defs => defs) + .ToList(); + + var filters = BuildFilters(live, _defaultQos); + if (filters.Count == 0) + { + return; + } + + var granted = await EstablishAsync(filters, cancellationToken).ConfigureAwait(false); + if (granted == 0) + { + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': no subscription could be re-established after reconnect " + + $"({filters.Count} filter(s) requested); tearing the session down rather than serving a " + + "connection that receives nothing."); + } + } + + /// + /// Routes one inbound message to every authored tag whose topic matches, extracts the value, + /// updates , and raises for the tags that are + /// currently subscribed. + /// + /// + /// Runs on MQTTnet's dispatcher thread: no I/O, no locks, no allocation beyond the extracted + /// value, and no path throws. A tag whose payload is malformed degrades to a Bad status + /// code and the fan-out continues to the next tag. + /// + /// The concrete topic the message arrived on. + /// The message body; valid only for the duration of this call. + /// The message's retain flag — true is the broker's subscribe-time seed. + public void HandleMessage(string topic, ReadOnlySpan payload, bool retained) + { + if (string.IsNullOrEmpty(topic)) + { + return; + } + + var authored = _authored; + var now = DateTime.UtcNow; + + // Concrete topics — the overwhelmingly common case — cost one lookup, so a chatty broker + // publishing thousands of topics this driver does not care about stays cheap. + if (authored.ByTopic.TryGetValue(topic, out var exact)) + { + foreach (var def in exact) + { + Deliver(def, payload, retained, now); + } + } + + // A published topic name may not itself contain a wildcard, so a wildcard-authored definition + // can never also be an exact-index hit: no tag is delivered twice. + foreach (var def in authored.Wildcards) + { + if (MqttTopicFilterComparer.Compare(topic, def.Topic) == MqttTopicFilterCompareResult.IsMatch) + { + Deliver(def, payload, retained, now); + } + } + } + + /// + /// Collapses a set of definitions into the filters to SUBSCRIBE: one per distinct topic, + /// carrying the strongest QoS any tag on that topic asked for (so no tag silently gets a + /// weaker delivery guarantee than it was authored with), and seeding from the retained message + /// if any tag on the topic wants it (tags that do not are filtered on the retain flag at + /// delivery). + /// + /// The definitions to collapse. + /// The driver-level QoS applied to tags that authored none. + /// One filter per distinct topic. + internal static IReadOnlyList BuildFilters( + IEnumerable defs, + int defaultQos) + { + var byTopic = new Dictionary(StringComparer.Ordinal); + foreach (var def in defs) + { + var qos = def.Qos ?? defaultQos; + byTopic[def.Topic] = byTopic.TryGetValue(def.Topic, out var existing) + ? existing with + { + Qos = Math.Max(existing.Qos, qos), + SeedRetained = existing.SeedRetained || def.RetainSeed, + } + : new MqttTopicSubscription(def.Topic, qos, def.RetainSeed); + } + + return [.. byTopic.Values]; + } + + /// + /// Issues the SUBSCRIBE and applies its verdict: granted topics join the live set, refused ones + /// degrade their references to BadCommunicationError. + /// + /// The filters to establish; an empty list is a no-op. + /// Cancellation for the round-trip. + /// How many filters the broker granted. + private async Task EstablishAsync( + IReadOnlyList filters, + CancellationToken cancellationToken) + { + if (filters.Count == 0) + { + return 0; + } + + var transport = _transport; + if (transport is null) + { + // No transport: the intent is recorded so the reconnect path establishes it, but NOTHING + // reaches a broker now. Loud, because outside tests this means the composition order is + // wrong — AttachTo must run before the OPC UA server's first subscribe, or these topics + // stay unestablished until the connection happens to drop and recover. + foreach (var filter in filters) + { + _subscribedTopics[filter.Topic] = 0; + } + + _logger?.LogWarning( + "MQTT driver '{DriverId}': {FilterCount} subscription(s) recorded with no connection attached; " + + "nothing was sent to a broker. Call AttachTo before subscribing.", + _driverId, + filters.Count); + + return filters.Count; + } + + IReadOnlyList outcomes; + try + { + outcomes = await transport.SubscribeAsync(filters, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': SUBSCRIBE for {FilterCount} filter(s) failed; the affected tags degrade to Bad.", + _driverId, + filters.Count); + foreach (var filter in filters) + { + DegradeTopic(filter.Topic, ex.GetType().Name); + } + + return 0; + } + + var granted = 0; + foreach (var outcome in outcomes) + { + if (outcome.Granted) + { + _subscribedTopics[outcome.Topic] = 0; + granted++; + } + else + { + _logger?.LogError( + "MQTT driver '{DriverId}': broker refused SUBSCRIBE '{Topic}' ({Reason}); its tags degrade to Bad.", + _driverId, + outcome.Topic, + outcome.Reason); + DegradeTopic(outcome.Topic, outcome.Reason); + } + } + + return granted; + } + + /// Marks every authored reference on as unreachable. + /// The topic filter whose subscription could not be established. + /// The failure reason, for logs. + private void DegradeTopic(string topic, string reason) + { + _subscribedTopics.TryRemove(topic, out _); + + if (!_authored.ByTopic.TryGetValue(topic, out var defs)) + { + return; + } + + var now = DateTime.UtcNow; + var snapshot = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); + foreach (var def in defs) + { + Values.Update(def.Name, snapshot); + Raise(def.Name, snapshot); + } + + _logger?.LogDebug( + "MQTT driver '{DriverId}': degraded {Count} tag(s) on '{Topic}' ({Reason}).", + _driverId, + defs.Length, + topic, + reason); + } + + /// + /// Delivers one message to one tag: honours the retained-seed opt-out, extracts the value, + /// caches it and raises the notification. Never throws. + /// + /// The authored tag this message matched. + /// The message body. + /// The message's retain flag. + /// One clock read shared by the whole fan-out, so siblings agree. + private void Deliver(MqttTagDefinition def, ReadOnlySpan payload, bool retained, DateTime nowUtc) + { + // MQTT 3.1.1 has no retain-handling option, so the broker always replays its retained message + // on subscribe; a tag that opted out of seeding is filtered here rather than at the broker. + if (retained && !def.RetainSeed) + { + return; + } + + DataValueSnapshot snapshot; + try + { + snapshot = Extract(def, payload, nowUtc); + } + catch (Exception ex) + { + // Defence in depth: Extract is written not to throw, but this runs on the library's + // dispatcher and an escape here would stall delivery for every other subscription. + snapshot = new DataValueSnapshot(null, StatusBadDecodingError, null, nowUtc); + _logger?.LogDebug(ex, "MQTT driver '{DriverId}': extraction threw for '{RawPath}'.", _driverId, def.Name); + } + + if (snapshot.StatusCode != StatusGood && _warnedRefs.TryAdd(def.Name, 0)) + { + // Loud once per reference per deploy, quiet thereafter: a mis-authored jsonPath on a 10 Hz + // topic must be diagnosable without drowning the log. + _logger?.LogWarning( + "MQTT driver '{DriverId}': tag '{RawPath}' (topic '{Topic}', {Format}, jsonPath '{JsonPath}', " + + "{DataType}) could not be extracted from its payload; status 0x{Status:X8}.", + _driverId, + def.Name, + def.Topic, + def.PayloadFormat, + def.JsonPath, + def.DataType, + snapshot.StatusCode); + } + + Values.Update(def.Name, snapshot); + Raise(def.Name, snapshot); + } + + /// + /// Raises for a reference, if a live subscription covers it. + /// A throwing subscriber is contained so it cannot starve the rest of the fan-out — or, worse, + /// escape onto MQTTnet's dispatcher thread. + /// + /// The RawPath — the reference the notification is published under. + /// The value/quality/timestamps to publish. + private void Raise(string rawPath, DataValueSnapshot snapshot) + { + if (!_handleByRawPath.TryGetValue(rawPath, out var handle)) + { + // Authored but not subscribed: the value is cached (so a read answers) and nothing is + // published — there is no subscription to attribute it to. + return; + } + + try + { + OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, rawPath, snapshot)); + } + catch (Exception ex) + { + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': a data-change subscriber threw for '{RawPath}'.", + _driverId, + rawPath); + } + } + + /// + /// Extracts the tag's value from a payload according to its . + /// Never throws: every failure becomes a Bad status code on the returned snapshot. + /// + /// The authored tag definition. + /// The message body. + /// Timestamp for the snapshot — MQTT carries no source timestamp of its own. + /// The extracted snapshot. + internal static DataValueSnapshot Extract(MqttTagDefinition def, ReadOnlySpan payload, DateTime nowUtc) + { + switch (def.PayloadFormat) + { + case MqttPayloadFormat.Json: + return ExtractJson(def, payload, nowUtc); + + case MqttPayloadFormat.Raw: + // "Bytes as-is": the payload's UTF-8 text, published verbatim. dataType is deliberately + // NOT applied — Scalar is the format that coerces. Authoring Raw with a numeric + // dataType is a mis-authoring, and one the deploy-time inspection should grow a + // warning for. + return TryDecodeUtf8(payload, out var raw) + ? new DataValueSnapshot(raw, StatusGood, nowUtc, nowUtc) + : Bad(StatusBadDecodingError, nowUtc); + + case MqttPayloadFormat.Scalar: + if (!TryDecodeUtf8(payload, out var text)) + { + return Bad(StatusBadDecodingError, nowUtc); + } + + return TryCoerceText(text, def.DataType, out var scalar) + ? new DataValueSnapshot(scalar, StatusGood, nowUtc, nowUtc) + : Bad(StatusBadTypeMismatch, nowUtc); + + default: + return Bad(StatusBadDecodingError, nowUtc); + } + } + + private static DataValueSnapshot Bad(uint status, DateTime nowUtc) => new(null, status, null, nowUtc); + + /// Parses the payload as JSON, selects the tag's JSONPath, and coerces to its data type. + /// The authored tag definition. + /// The message body. + /// Timestamp for the snapshot. + /// The extracted snapshot. + private static DataValueSnapshot ExtractJson(MqttTagDefinition def, ReadOnlySpan payload, DateTime nowUtc) + { + JsonDocument? doc = null; + try + { + var reader = new Utf8JsonReader(payload); + if (!JsonDocument.TryParseValue(ref reader, out doc)) + { + return Bad(StatusBadDecodingError, nowUtc); + } + + if (!TryEvaluateJsonPath(doc.RootElement, def.JsonPath, out var selected)) + { + return Bad(StatusBadDecodingError, nowUtc); + } + + return TryCoerce(selected, def.DataType, out var value) + ? new DataValueSnapshot(value, StatusGood, nowUtc, nowUtc) + : Bad(StatusBadTypeMismatch, nowUtc); + } + catch (JsonException) + { + return Bad(StatusBadDecodingError, nowUtc); + } + finally + { + doc?.Dispose(); + } + } + + /// + /// Evaluates the supported subset of JSONPath against a document: the root ($), + /// dotted member access ($.a.b), bracketed member access ($['a'] / $["a"]) + /// and non-negative array indexing ($.a[0]). + /// + /// + /// Filters, slices, wildcards and recursive descent ($..x) are deliberately not + /// supported: this repo pins no JSONPath package, adding one for an expression form that + /// selects a set — while a tag needs exactly one scalar — is the wrong trade. An + /// unsupported expression is reported as a miss, so the tag surfaces + /// BadDecodingError plus the one-shot warning rather than a wrong value. + /// + /// The parsed document root. + /// The authored JSONPath. + /// The selected element when this returns . + /// when the path is supported and selects an element. + internal static bool TryEvaluateJsonPath(JsonElement root, string path, out JsonElement value) + { + value = root; + if (string.IsNullOrEmpty(path) || path[0] != '$') + { + return false; + } + + var i = 1; + while (i < path.Length) + { + if (path[i] == '.') + { + i++; + if (i >= path.Length || path[i] == '.') + { + return false; // trailing '.', or recursive descent '..' + } + + var start = i; + while (i < path.Length && path[i] != '.' && path[i] != '[') + { + i++; + } + + var name = path[start..i]; + if (name.Length == 0 || name == "*") + { + return false; + } + + if (value.ValueKind != JsonValueKind.Object || !value.TryGetProperty(name, out var member)) + { + return false; + } + + value = member; + } + else if (path[i] == '[') + { + var close = path.IndexOf(']', i); + if (close < 0) + { + return false; + } + + var inner = path[(i + 1)..close].Trim(); + i = close + 1; + + if (inner.Length >= 2 + && ((inner[0] == '\'' && inner[^1] == '\'') || (inner[0] == '"' && inner[^1] == '"'))) + { + var name = inner[1..^1]; + if (value.ValueKind != JsonValueKind.Object || !value.TryGetProperty(name, out var member)) + { + return false; + } + + value = member; + } + else if (int.TryParse(inner, NumberStyles.Integer, CultureInfo.InvariantCulture, out var index) + && index >= 0) + { + if (value.ValueKind != JsonValueKind.Array || index >= value.GetArrayLength()) + { + return false; + } + + value = value[index]; + } + else + { + return false; // slice, filter or wildcard — outside the supported subset + } + } + else + { + return false; + } + } + + return true; + } + + /// Coerces a selected JSON element to the tag's declared data type. + /// The selected element. + /// The tag's declared type. + /// The coerced value when this returns . + /// when the element fits the declared type. + internal static bool TryCoerce(JsonElement element, DriverDataType type, out object? value) + { + value = null; + + if (element.ValueKind is JsonValueKind.Null or JsonValueKind.Undefined) + { + return false; + } + + if (type is DriverDataType.String or DriverDataType.Reference) + { + // A String tag pointed at an object/array gets that subtree's JSON text — the authored + // intent ("give me this fragment") rather than a refusal. + value = element.ValueKind == JsonValueKind.String ? element.GetString() : element.GetRawText(); + return true; + } + + // A JSON string holding a number ({"v":"21.5"}) is common in the wild; parse it as text rather + // than refusing a payload the operator plainly meant. + if (element.ValueKind == JsonValueKind.String) + { + return TryCoerceText(element.GetString() ?? string.Empty, type, out value); + } + + switch (type) + { + case DriverDataType.Boolean: + if (element.ValueKind == JsonValueKind.True) { value = true; return true; } + if (element.ValueKind == JsonValueKind.False) { value = false; return true; } + if (element.ValueKind == JsonValueKind.Number && element.TryGetDouble(out var b)) { value = b != 0d; return true; } + return false; + + case DriverDataType.Int16: + if (element.ValueKind == JsonValueKind.Number && element.TryGetInt16(out var i16)) { value = i16; return true; } + return false; + + case DriverDataType.Int32: + if (element.ValueKind == JsonValueKind.Number && element.TryGetInt32(out var i32)) { value = i32; return true; } + return false; + + case DriverDataType.Int64: + if (element.ValueKind == JsonValueKind.Number && element.TryGetInt64(out var i64)) { value = i64; return true; } + return false; + + case DriverDataType.UInt16: + if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt16(out var u16)) { value = u16; return true; } + return false; + + case DriverDataType.UInt32: + if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt32(out var u32)) { value = u32; return true; } + return false; + + case DriverDataType.UInt64: + if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt64(out var u64)) { value = u64; return true; } + return false; + + case DriverDataType.Float32: + if (element.ValueKind == JsonValueKind.Number && element.TryGetSingle(out var f32)) { value = f32; return true; } + return false; + + case DriverDataType.Float64: + if (element.ValueKind == JsonValueKind.Number && element.TryGetDouble(out var f64)) { value = f64; return true; } + return false; + + case DriverDataType.DateTime: + // Only ISO-8601 text is accepted; a bare number is ambiguous (epoch seconds? ms?) and + // guessing would silently place samples decades apart. + return false; + + default: + return false; + } + } + + /// Parses a textual payload (or JSON string) into the tag's declared data type. + /// The decoded text. + /// The tag's declared type. + /// The parsed value when this returns . + /// when the text parses as the declared type. + internal static bool TryCoerceText(string text, DriverDataType type, out object? value) + { + value = null; + var inv = CultureInfo.InvariantCulture; + var trimmed = text.Trim(); + + switch (type) + { + case DriverDataType.String: + case DriverDataType.Reference: + // Untrimmed: for a String tag the payload IS the value, whitespace included. + value = text; + return true; + + case DriverDataType.Boolean: + if (bool.TryParse(trimmed, out var b)) { value = b; return true; } + if (long.TryParse(trimmed, NumberStyles.Integer, inv, out var bi)) { value = bi != 0L; return true; } + return false; + + case DriverDataType.Int16: + if (short.TryParse(trimmed, NumberStyles.Integer, inv, out var i16)) { value = i16; return true; } + return false; + + case DriverDataType.Int32: + if (int.TryParse(trimmed, NumberStyles.Integer, inv, out var i32)) { value = i32; return true; } + return false; + + case DriverDataType.Int64: + if (long.TryParse(trimmed, NumberStyles.Integer, inv, out var i64)) { value = i64; return true; } + return false; + + case DriverDataType.UInt16: + if (ushort.TryParse(trimmed, NumberStyles.Integer, inv, out var u16)) { value = u16; return true; } + return false; + + case DriverDataType.UInt32: + if (uint.TryParse(trimmed, NumberStyles.Integer, inv, out var u32)) { value = u32; return true; } + return false; + + case DriverDataType.UInt64: + if (ulong.TryParse(trimmed, NumberStyles.Integer, inv, out var u64)) { value = u64; return true; } + return false; + + case DriverDataType.Float32: + if (float.TryParse(trimmed, NumberStyles.Float, inv, out var f32)) { value = f32; return true; } + return false; + + case DriverDataType.Float64: + if (double.TryParse(trimmed, NumberStyles.Float, inv, out var f64)) { value = f64; return true; } + return false; + + case DriverDataType.DateTime: + if (DateTime.TryParse( + trimmed, + inv, + DateTimeStyles.RoundtripKind | DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal, + out var dt)) + { + value = dt; + return true; + } + + return false; + + default: + return false; + } + } + + /// + /// Decodes a payload as UTF-8, refusing invalid byte sequences rather than rendering a genuinely + /// binary body as a field of replacement characters (the same choice the browse session makes). + /// + /// The message body. + /// The decoded text when this returns . + /// when the payload is valid UTF-8. + private static bool TryDecodeUtf8(ReadOnlySpan payload, out string text) + { + try + { + text = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true) + .GetString(payload); + return true; + } + catch (DecoderFallbackException) + { + text = string.Empty; + return false; + } + } + + /// + /// Builds the handle's diagnostic id: the ingest mode plus the topics it covers, bounded so a + /// thousand-tag subscription does not produce a thousand-topic string in every log line. + /// + /// The definitions the subscription covers. + /// The diagnostic id. + private string BuildDiagnosticId(IReadOnlyCollection defs) + { + const int MaxNamedTopics = 5; + + var id = Interlocked.Increment(ref _handleSeq); + var topics = defs.Select(d => d.Topic).Distinct(StringComparer.Ordinal).ToList(); + var named = string.Join(',', topics.Take(MaxNamedTopics)); + var overflow = topics.Count > MaxNamedTopics ? $",+{topics.Count - MaxNamedTopics}" : string.Empty; + + return $"mqtt:{_mode}:{id}:[{named}{overflow}]"; + } + + /// Opaque subscription identity; the diagnostic id names the mode and its topics. + /// The human-readable identity surfaced for diagnostics. + private sealed record MqttSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle; + + /// + /// The authored tag table and its derived topic index, published as one immutable snapshot so + /// the dispatcher thread reads a consistent view while a redeploy rebuilds it. + /// + /// RawPath → definition; the v3 resolver's backing table. + /// Concrete topic → every definition bound to it (the fan-out list). + /// Definitions whose authored topic carries an MQTT wildcard. + private sealed record AuthoredTable( + FrozenDictionary ByRawPath, + FrozenDictionary ByTopic, + MqttTagDefinition[] Wildcards) + { + public static readonly AuthoredTable Empty = + Build(new Dictionary(StringComparer.Ordinal)); + + /// Builds the snapshot, splitting concrete topics from wildcard-authored ones. + /// The mapped definitions, keyed by RawPath. + /// The immutable snapshot. + public static AuthoredTable Build(IReadOnlyDictionary byRawPath) + { + var concrete = new Dictionary>(StringComparer.Ordinal); + var wildcards = new List(); + + foreach (var def in byRawPath.Values) + { + if (def.Topic.Contains(MqttTopicFilterComparer.SingleLevelWildcard, StringComparison.Ordinal) + || def.Topic.Contains(MqttTopicFilterComparer.MultiLevelWildcard, StringComparison.Ordinal)) + { + wildcards.Add(def); + continue; + } + + if (!concrete.TryGetValue(def.Topic, out var list)) + { + concrete[def.Topic] = list = []; + } + + list.Add(def); + } + + return new AuthoredTable( + byRawPath.ToFrozenDictionary(StringComparer.Ordinal), + concrete.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal), + [.. wildcards]); + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs new file mode 100644 index 00000000..c0f7b3f7 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs @@ -0,0 +1,607 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Covers — the Task-6 plain-MQTT ingest path: authored-tag +/// registration keyed by RawPath, topic dedupe + fan-out, payload extraction, the retained seed, +/// and the subscribe/SUBACK contract. Every test runs without a broker: messages are fed straight +/// through (the method MQTTnet's dispatcher +/// calls) and the subscribe leg runs against a fake . +/// +public sealed class MqttSubscriptionManagerTests +{ + // OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled. + private const uint Good = 0x00000000u; + private const uint BadDecodingError = 0x80070000u; + private const uint BadTypeMismatch = 0x80740000u; + private const uint BadNodeIdUnknown = 0x80340000u; + private const uint BadCommunicationError = 0x80050000u; + + private const string OvenTempPath = "Plant/Mqtt/broker1/OvenTemp"; + private const string OvenPressPath = "Plant/Mqtt/broker1/OvenPressure"; + private const string OvenTopic = "f/oven/temp"; + + private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); + + private static MqttSubscriptionManager NewManager(IMqttSubscribeTransport? transport = null) => + new(new MqttDriverOptions { Mode = MqttMode.Plain }, "mqtt-1", transport); + + // --------------------------------------------------------------------------------- + // Routing + the RawPath identity of the published reference + // --------------------------------------------------------------------------------- + + /// + /// The plan's first snippet, corrected twice: Double is not a + /// member (Float64 is), and the published reference is + /// asserted to BE the RawPath rather than merely to contain the topic — a topic-keyed + /// publish would satisfy the plan's assertion and be silently dead against + /// DriverHostActor's RawPath-keyed dual-namespace fan-out. + /// + [Fact] + public async Task HandleMessage_MatchingTopic_RaisesDataChangeAtJsonPath_UnderRawPath() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + string? gotRef = null; + object? gotVal = null; + mgr.OnDataChange += (_, e) => { gotRef = e.FullReference; gotVal = e.Snapshot.Value; }; + + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + + gotVal.ShouldBe(21.5); + gotRef.ShouldBe(OvenTempPath); + gotRef.ShouldNotBe(OvenTopic); + } + + /// + /// A chatty broker must never auto-provision a tag the operator did not author. Deliberately + /// run with a live, subscribed tag present rather than an empty table: an empty table makes + /// the assertion unfalsifiable — a "deliver to every authored tag" implementation would pass it + /// for want of anything to deliver to. + /// + [Fact] + public async Task HandleMessage_UnauthoredTopic_RaisesNothing() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var fired = false; + mgr.OnDataChange += (_, _) => fired = true; + + mgr.HandleMessage("random/topic", "1"u8.ToArray(), retained: false); + + fired.ShouldBeFalse(); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldNotBe(Good); // and nothing was cached for it either + } + + /// + /// Two tags on ONE topic (different JSONPaths) — the message must reach BOTH. A + /// TryGetValue → first match → return implementation passes every single-tag test and + /// silently starves every tag after the first. + /// + [Fact] + public async Task HandleMessage_TopicSharedByTwoTags_FansOutToEveryMatchingTag() + { + var mgr = NewManager(); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.temp","dataType":"Float64"}"""), + Tag(OvenPressPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.press","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var seen = new Dictionary(); + mgr.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value; + + mgr.HandleMessage(OvenTopic, """{"temp":21.5,"press":1.2}"""u8.ToArray(), retained: false); + + seen.Count.ShouldBe(2); + seen[OvenTempPath].ShouldBe(21.5); + seen[OvenPressPath].ShouldBe(1.2); + } + + /// + /// An authored tag nobody subscribed to still caches its value (so a later + /// IReadable.ReadAsync answers) but must not raise OnDataChange — there is no + /// subscription handle to attribute it to. + /// + [Fact] + public void HandleMessage_AuthoredButUnsubscribedTag_CachesValue_RaisesNothing() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + var fired = false; + mgr.OnDataChange += (_, _) => fired = true; + + mgr.HandleMessage(OvenTopic, "7.5"u8.ToArray(), retained: false); + + fired.ShouldBeFalse(); + mgr.Values.Read(OvenTempPath).Value.ShouldBe(7.5); + } + + /// After UnsubscribeAsync the reference stops publishing. + [Fact] + public async Task UnsubscribeAsync_StopsRaisingForThoseReferences() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var count = 0; + mgr.OnDataChange += (_, _) => count++; + mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false); + + await mgr.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); + mgr.HandleMessage(OvenTopic, "2"u8.ToArray(), retained: false); + + count.ShouldBe(1); + } + + // --------------------------------------------------------------------------------- + // Wildcard-authored topics (accepted by the parser, warned at deploy) + // --------------------------------------------------------------------------------- + + [Theory] + [InlineData("f/+/temp", "f/oven/temp", true)] + [InlineData("f/+/temp", "f/oven/deep/temp", false)] // '+' is single-level + [InlineData("f/#", "f/oven/deep/temp", true)] // '#' is multi-level + [InlineData("f/#", "g/oven/temp", false)] + public async Task HandleMessage_WildcardAuthoredTopic_UsesRealMqttFilterSemantics( + string authoredTopic, string incomingTopic, bool expectMatch) + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{authoredTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var fired = false; + mgr.OnDataChange += (_, _) => fired = true; + + mgr.HandleMessage(incomingTopic, "1"u8.ToArray(), retained: false); + + fired.ShouldBe(expectMatch); + } + + // --------------------------------------------------------------------------------- + // Retained seed + // --------------------------------------------------------------------------------- + + [Fact] + public async Task HandleMessage_RetainedMessage_SeedsWhenRetainSeedIsTrue() + { + var mgr = NewManager(); + // retainSeed absent ⇒ true (the factory default). + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + object? gotVal = null; + mgr.OnDataChange += (_, e) => gotVal = e.Snapshot.Value; + + mgr.HandleMessage(OvenTopic, "3.5"u8.ToArray(), retained: true); + + gotVal.ShouldBe(3.5); + } + + [Fact] + public async Task HandleMessage_RetainedMessage_IgnoredWhenRetainSeedIsFalse() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64","retainSeed":false}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var fired = false; + mgr.OnDataChange += (_, _) => fired = true; + + mgr.HandleMessage(OvenTopic, "3.5"u8.ToArray(), retained: true); + fired.ShouldBeFalse(); + + // A live (non-retained) publish on the same tag still lands — retainSeed gates the seed only. + mgr.HandleMessage(OvenTopic, "3.5"u8.ToArray(), retained: false); + fired.ShouldBeTrue(); + } + + // --------------------------------------------------------------------------------- + // Value extraction — Json / Scalar / Raw, and per-tag degradation + // --------------------------------------------------------------------------------- + + [Theory] + [InlineData("$", """21.5""", 21.5)] + [InlineData("$.value", """{"value":21.5}""", 21.5)] + [InlineData("$.a.b", """{"a":{"b":21.5}}""", 21.5)] + [InlineData("$.arr[1]", """{"arr":[1.5,21.5]}""", 21.5)] + [InlineData("$['value']", """{"value":21.5}""", 21.5)] + public async Task HandleMessage_Json_ExtractsSupportedJsonPathShapes(string jsonPath, string payload, double expected) + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"{{jsonPath.Replace("\\", "\\\\").Replace("\"", "\\\"")}}","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + mgr.HandleMessage(OvenTopic, System.Text.Encoding.UTF8.GetBytes(payload), retained: false); + + snap!.StatusCode.ShouldBe(Good); + snap.Value.ShouldBe(expected); + } + + [Fact] + public async Task HandleMessage_MalformedJsonPayload_PublishesBadDecodingError_NeverThrows() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + Should.NotThrow(() => mgr.HandleMessage(OvenTopic, "{not json"u8.ToArray(), retained: false)); + + snap!.StatusCode.ShouldBe(BadDecodingError); + snap.Value.ShouldBeNull(); + } + + [Theory] + [InlineData("$.missing")] + [InlineData("$..value")] // recursive descent — outside the supported subset + [InlineData("$.arr[9]")] + public async Task HandleMessage_UnresolvableOrUnsupportedJsonPath_PublishesBadDecodingError(string jsonPath) + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"{{jsonPath}}","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + mgr.HandleMessage(OvenTopic, """{"value":21.5,"arr":[1]}"""u8.ToArray(), retained: false); + + snap!.StatusCode.ShouldBe(BadDecodingError); + } + + [Fact] + public async Task HandleMessage_ValueNotCoercibleToDeclaredDataType_PublishesBadTypeMismatch() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + mgr.HandleMessage(OvenTopic, """{"value":"not-a-number"}"""u8.ToArray(), retained: false); + + snap!.StatusCode.ShouldBe(BadTypeMismatch); + snap.Value.ShouldBeNull(); + } + + [Theory] + [InlineData("Boolean", "true", true)] + [InlineData("Int32", "42", 42)] + [InlineData("Int64", "42", 42L)] + [InlineData("Float32", "1.5", 1.5f)] + [InlineData("Float64", "1.5", 1.5)] + [InlineData("String", "hello", "hello")] + public async Task HandleMessage_Scalar_ParsesByDeclaredDataType(string dataType, string payload, object expected) + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"{{dataType}}"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + mgr.HandleMessage(OvenTopic, System.Text.Encoding.UTF8.GetBytes(payload), retained: false); + + snap!.StatusCode.ShouldBe(Good); + snap.Value.ShouldBe(expected); + } + + /// + /// Raw means "bytes as-is": the payload is published as its UTF-8 text and the declared + /// dataType is deliberately NOT applied (Scalar is the member that coerces). + /// + [Fact] + public async Task HandleMessage_Raw_PublishesUtf8TextAndIgnoresDeclaredDataType() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Raw","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false); + + snap!.StatusCode.ShouldBe(Good); + snap.Value.ShouldBe("21.5"); + } + + [Fact] + public async Task HandleMessage_RawNonUtf8Payload_PublishesBadDecodingError_NeverThrows() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Raw","dataType":"String"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + DataValueSnapshot? snap = null; + mgr.OnDataChange += (_, e) => snap = e.Snapshot; + + Should.NotThrow(() => mgr.HandleMessage(OvenTopic, new byte[] { 0xC3, 0x28 }, retained: false)); + + snap!.StatusCode.ShouldBe(BadDecodingError); + } + + /// + /// HandleMessage runs on MQTTnet's dispatcher thread, so nothing it calls — including a + /// misbehaving downstream subscriber — may escape and stall the client's pump. + /// + [Fact] + public async Task HandleMessage_ThrowingSubscriber_DoesNotEscapeToTheDispatcher() + { + var mgr = NewManager(); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var delivered = new List(); + mgr.OnDataChange += (_, e) => + { + delivered.Add(e.FullReference); + throw new InvalidOperationException("downstream blew up"); + }; + + Should.NotThrow(() => mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false)); + + // And a throwing subscriber must not starve the tags behind it in the fan-out. + delivered.Count.ShouldBe(2); + } + + // --------------------------------------------------------------------------------- + // Registration + // --------------------------------------------------------------------------------- + + [Fact] + public void Register_UnmappableTagConfig_IsSkipped_NeverThrows() + { + var mgr = NewManager(); + + var mapped = 0; + Should.NotThrow(() => mapped = mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, """{"payloadFormat":"Scalar"}"""), // no topic ⇒ rejected by the factory + Tag("Plant/Mqtt/broker1/Junk", "not json at all"), + ])); + + mapped.ShouldBe(1); + mgr.TryResolve(OvenTempPath, out _).ShouldBeTrue(); + mgr.TryResolve(OvenPressPath, out _).ShouldBeFalse(); + } + + /// Re-registering replaces the authored table wholesale — a redeploy is not additive. + [Fact] + public void Register_Twice_ReplacesTheAuthoredTable() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + mgr.Register([Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + mgr.TryResolve(OvenTempPath, out _).ShouldBeFalse(); + mgr.TryResolve(OvenPressPath, out _).ShouldBeTrue(); + } + + // --------------------------------------------------------------------------------- + // Filter building — one SUBSCRIBE per distinct topic, strongest QoS, retain handling + // --------------------------------------------------------------------------------- + + [Fact] + public void BuildFilters_DedupesTopics_TakesStrongestQos_AndSeedsWhenAnyTagWants() + { + var a = new MqttTagDefinition("p/a", OvenTopic, MqttPayloadFormat.Scalar, "$", DriverDataType.Float64, Qos: 0, RetainSeed: false); + var b = new MqttTagDefinition("p/b", OvenTopic, MqttPayloadFormat.Scalar, "$", DriverDataType.Float64, Qos: 2, RetainSeed: true); + var c = new MqttTagDefinition("p/c", "f/oven/press", MqttPayloadFormat.Scalar, "$", DriverDataType.Float64, Qos: null, RetainSeed: false); + + var filters = MqttSubscriptionManager.BuildFilters([a, b, c], defaultQos: 1); + + filters.Count.ShouldBe(2); + var oven = filters.Single(f => f.Topic == OvenTopic); + oven.Qos.ShouldBe(2); // strongest wins — nobody gets a weaker guarantee than authored + oven.SeedRetained.ShouldBeTrue(); + var press = filters.Single(f => f.Topic == "f/oven/press"); + press.Qos.ShouldBe(1); // null ⇒ the driver-level default + press.SeedRetained.ShouldBeFalse(); + } + + // --------------------------------------------------------------------------------- + // Subscribe / SUBACK contract + // --------------------------------------------------------------------------------- + + [Fact] + public async Task SubscribeAsync_IssuesTheFirstSubscribeItself_OneFilterPerDistinctTopic() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(1); + transport.Calls[0].Select(f => f.Topic).ShouldBe([OvenTopic]); + } + + [Fact] + public async Task SubscribeAsync_HandleDiagnosticId_NamesModeAndFilters() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + handle.DiagnosticId.ShouldContain("Plain"); + handle.DiagnosticId.ShouldContain(OvenTopic); + } + + /// An already-subscribed topic is not re-SUBSCRIBEd by a second overlapping subscription. + [Fact] + public async Task SubscribeAsync_AlreadySubscribedTopic_IsNotResubscribed() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + await mgr.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(1); + } + + [Fact] + public async Task SubscribeAsync_UnknownReference_MarksItBadNodeIdUnknown_AndDoesNotThrow() + { + var mgr = NewManager(); + mgr.Register([]); + + await Should.NotThrowAsync(() => + mgr.SubscribeAsync(["Plant/Mqtt/broker1/Nope"], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + mgr.Values.Read("Plant/Mqtt/broker1/Nope").StatusCode.ShouldBe(BadNodeIdUnknown); + } + + /// + /// A SUBACK failure on the ISubscribable path degrades the affected references to Bad + /// and returns a handle — it must never hang, and must never throw out of the OPC UA server's + /// subscribe call. + /// + [Fact] + public async Task SubscribeAsync_SubackFailure_DegradesRefsToBad_ReturnsHandle_DoesNotThrow() + { + var transport = new FakeTransport { Reject = _ => true }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + // Not wrapped in Should.NotThrowAsync: an escaping exception fails this test on its own, and + // the handle must still come back. + var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + handle.ShouldNotBeNull(); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + } + + [Fact] + public async Task SubscribeAsync_TransportThrows_DegradesRefsToBad_DoesNotThrow() + { + var transport = new FakeTransport { Throw = new InvalidOperationException("broker gone") }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + await Should.NotThrowAsync(() => + mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + } + + // --------------------------------------------------------------------------------- + // Reconnect re-subscribe + // --------------------------------------------------------------------------------- + + [Fact] + public async Task OnReconnectedAsync_ReSubscribesEveryLiveTopic_Idempotently() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(3); // initial + two reconnects + transport.Calls[1].Select(f => f.Topic).ShouldBe([OvenTopic]); + transport.Calls[2].Select(f => f.Topic).ShouldBe([OvenTopic]); + } + + /// + /// A total re-subscribe failure THROWS: 's documented contract is + /// that a throwing Reconnected subscriber tears the session down and retries under + /// backoff, and a connected-but-deaf client is the one outcome this driver must never serve. + /// + [Fact] + public async Task OnReconnectedAsync_TotalFailure_Throws_SoTheSessionIsTornDownAndRetried() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Throw = new InvalidOperationException("broker gone"); + + await Should.ThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + } + + /// + /// A PARTIAL rejection (e.g. one ACL-denied topic) must NOT throw — tearing the session down + /// forever because one of many topics is denied would take the whole driver dark. Only the + /// denied references degrade. + /// + [Fact] + public async Task OnReconnectedAsync_PartialRejection_DoesNotThrow_OnlyDeniedRefsDegrade() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false); + + transport.Reject = t => t == "f/oven/press"; + + await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + + mgr.Values.Read(OvenPressPath).StatusCode.ShouldBe(BadCommunicationError); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good); + } + + /// Records every subscribe call so dedupe / re-subscribe can be asserted without a broker. + private sealed class FakeTransport : IMqttSubscribeTransport + { + public List> Calls { get; } = []; + + /// Topics for which the fake SUBACK reports a rejection. + public Func Reject { get; set; } = _ => false; + + /// When set, the whole subscribe call throws instead of answering. + public Exception? Throw { get; set; } + + public Task> SubscribeAsync( + IReadOnlyList filters, CancellationToken cancellationToken) + { + Calls.Add(filters); + if (Throw is not null) throw Throw; + IReadOnlyList outcomes = + [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, !Reject(f.Topic), Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1"))]; + return Task.FromResult(outcomes); + } + } +}