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
{
///
/// Default — 1 MiB. Generous for MQTT telemetry (which is
/// kilobytes) while still bounding the dispatcher-thread work any one publisher can impose.
///
public const int DefaultMaxPayloadBytes = 1024 * 1024;
// 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 .
///
///
/// Ceiling on an inbound message body; see . Non-positive values
/// fall back to — "unbounded" is not offered, because the
/// bound exists to protect a shared dispatcher thread.
///
public MqttSubscriptionManager(
MqttDriverOptions options,
string driverId,
IMqttSubscribeTransport? transport = null,
ILogger? logger = null,
LastValueCache? cache = null,
int maxPayloadBytes = DefaultMaxPayloadBytes)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverId);
MaxPayloadBytes = maxPayloadBytes > 0 ? maxPayloadBytes : DefaultMaxPayloadBytes;
_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; }
///
/// Ceiling on an inbound message body. A larger message is refused before any decode or parse
/// and degrades its own tags to BadDecodingError.
///
///
/// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their cost
/// is paid by every subscription, not just the offending topic. Without a bound, one
/// publisher shipping a multi-megabyte body — or a hostile one shipping a huge nested document
/// — stalls the whole driver's delivery for as long as the parse takes. Sized for telemetry
/// payloads, which are kilobytes; a deployment that legitimately needs more should raise this
/// deliberately rather than run unbounded.
///
/// Follow-up: this is a constructor knob, not yet an operator-facing one. Promoting
/// it to an MqttDriverOptions key (and pairing it with MQTTnet's
/// WithMaximumPacketSize, so an oversized body is refused at the protocol layer
/// rather than after the client has already buffered it) needs an edit to the
/// .Contracts project that this task was scoped out of.
///
///
public int MaxPayloadBytes { 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.
///
///
/// Binding to a session forgets the previous session's established filters — the same
/// reset performs, and for the same reason:
/// this manager outlives a session rebuild. MqttDriver.AdoptOptions only
/// replaces it when the ingest identity changed, so a driver-host reconnect (ShutdownAsync +
/// InitializeAsync, #527's watchdog path) and a ReinitializeAsync that changes only the
/// endpoint/credentials both re-attach this instance to a brand-new broker session.
/// _subscribedTopics is a claim about the session that has just gone: carried across,
/// the host's post-reconnect re-subscribe resolves an empty pending set, issues no SUBSCRIBE
/// at all, and leaves a driver that reports Connected/Healthy and receives nothing until the
/// new session happens to drop — the silent-death outcome every other path in this
/// type is written to prevent.
///
///
/// The connection to observe.
public void AttachTo(MqttConnection connection)
{
ArgumentNullException.ThrowIfNull(connection);
// FIRST, before delivery is wired: a new session holds none of the old one's subscriptions.
_subscribedTopics.Clear();
_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);
}
}
var table = AuthoredTable.Build(byRawPath);
_authored = table;
_warnedRefs.Clear();
// Drop state belonging to tags this deploy removed. Without this, a topic whose last tag was
// deleted stays in _subscribedTopics and is dutifully re-SUBSCRIBEd on every future reconnect,
// forever — and its RawPaths keep a handle mapping that can never fire again.
foreach (var topic in _subscribedTopics.Keys)
{
if (!table.ByFilter.ContainsKey(topic))
{
_subscribedTopics.TryRemove(topic, out _);
}
}
foreach (var rawPath in _handleByRawPath.Keys)
{
if (!table.ByRawPath.ContainsKey(rawPath))
{
_handleByRawPath.TryRemove(rawPath, out _);
}
}
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 subscribed = _subscribedTopics.Keys.ToList();
// ByFilter, NOT ByExactTopic: these keys are subscribed FILTER strings, so a wildcard-authored
// tag's key is its pattern ("f/+/temp"). Reading the concrete-only index here misses every
// wildcard tag, drops it out of the re-subscribe set, and leaves it dark forever behind a
// connection that reports healthy — see the AuthoredTable remarks.
var live = subscribed
.Select(topic => authored.ByFilter.TryGetValue(topic, out var defs) ? defs : [])
.SelectMany(defs => defs)
.ToList();
var filters = BuildFilters(live, _defaultQos);
if (filters.Count == 0)
{
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. If any
// topic was live, resolving zero filters for it is an anomaly and must be loud.
if (subscribed.Count > 0)
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': reconnect resolved NO filters for {TopicCount} live topic(s) "
+ "({Topics}); those subscriptions are not being re-established.",
_driverId,
subscribed.Count,
string.Join(", ", subscribed));
}
else
{
_logger?.LogDebug(
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
_driverId);
}
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, 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.
///
///
/// Bounded work. A body larger than is refused
/// before any decode or parse: the whole point of the dispatcher discipline is that
/// one topic cannot stall delivery for every other subscription, and an unbounded
/// GetString / JsonDocument.Parse over a multi-megabyte body does exactly
/// that. The oversized message degrades its own tags to BadDecodingError.
///
///
/// Parsed once. The documented shape is several tags sharing one JSON document with
/// a JSONPath each, so the document is parsed at most once per message and shared across
/// the whole fan-out rather than re-parsed per 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;
var oversized = payload.Length > MaxPayloadBytes;
// Zero-alloc until a Json-format tag actually matches; disposed in the finally below.
var json = default(PayloadJson);
try
{
// 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.ByExactTopic.TryGetValue(topic, out var exact))
{
foreach (var def in exact)
{
Deliver(def, payload, retained, now, oversized, ref json);
}
}
// 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, oversized, ref json);
}
}
}
finally
{
json.Dispose();
}
}
///
/// 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.
///
/// is a subscribed filter string (it came from a requested
/// filter or a SUBACK outcome), so the lookup must go through ByFilter — the
/// concrete-only index misses every wildcard-authored tag and this method would then return
/// having silently degraded nothing, leaving the tag at BadWaitingForInitialData (or
/// worse, its last Good value) after the broker refused its subscription.
///
/// 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.ByFilter.TryGetValue(topic, out var defs))
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': cannot degrade tags for unestablished filter '{Topic}' ({Reason}) — "
+ "no authored tag resolves to it.",
_driverId,
topic,
reason);
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.
/// Whether the body exceeded .
/// The fan-out's shared JSON parse, created on first use.
private void Deliver(
MqttTagDefinition def,
ReadOnlySpan payload,
bool retained,
DateTime nowUtc,
bool oversized,
ref PayloadJson json)
{
// 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
{
// The cap is enforced before any decode/parse, so an oversized body costs one comparison
// per matched tag rather than a full pass over the payload.
snapshot = oversized
? new DataValueSnapshot(null, StatusBadDecodingError, null, nowUtc)
: Extract(def, payload, nowUtc, ref json);
}
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 fan-out's shared JSON parse, created on first use by a Json-format tag.
/// The extracted snapshot.
internal static DataValueSnapshot Extract(
MqttTagDefinition def,
ReadOnlySpan payload,
DateTime nowUtc,
ref PayloadJson json)
{
switch (def.PayloadFormat)
{
case MqttPayloadFormat.Json:
return ExtractJson(def, payload, nowUtc, ref json);
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);
///
/// Selects the tag's JSONPath out of the fan-out's shared parse and coerces it to the
/// tag's data type. The document is parsed at most once per message however many tags share the
/// topic — the documented "one JSON document, one JSONPath per signal" shape would otherwise
/// re-parse it once per tag, on the dispatcher thread.
///
/// The authored tag definition.
/// The message body.
/// Timestamp for the snapshot.
/// The shared parse; parses on first call and caches the outcome.
/// The extracted snapshot.
private static DataValueSnapshot ExtractJson(
MqttTagDefinition def,
ReadOnlySpan payload,
DateTime nowUtc,
ref PayloadJson json)
{
if (!json.TryGetRoot(payload, out var root))
{
return Bad(StatusBadDecodingError, nowUtc);
}
if (!TryEvaluateJsonPath(root, 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);
}
///
/// One message's JSON parse, shared across the whole fan-out. A mutable struct passed by
/// ref so the common case — a message no Json-format tag matched — allocates nothing at
/// all, while a message several Json tags share is parsed exactly once.
///
internal struct PayloadJson : IDisposable
{
private JsonDocument? _doc;
private bool _attempted;
private bool _parsed;
///
/// The parsed document root, parsing on first call. A malformed payload is remembered as a
/// failure so the fan-out's remaining tags do not each retry the same doomed parse.
///
/// The message body; only read on the first call.
/// The document root when this returns .
/// when the payload is valid JSON.
public bool TryGetRoot(ReadOnlySpan payload, out JsonElement root)
{
if (!_attempted)
{
_attempted = true;
try
{
var reader = new Utf8JsonReader(payload);
_parsed = JsonDocument.TryParseValue(ref reader, out _doc);
}
catch (JsonException)
{
_parsed = false;
}
}
root = _parsed ? _doc!.RootElement : default;
return _parsed;
}
///
public void Dispose()
{
_doc?.Dispose();
_doc = null;
}
}
///
/// 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:
case DriverDataType.Int32:
case DriverDataType.Int64:
case DriverDataType.UInt16:
case DriverDataType.UInt32:
case DriverDataType.UInt64:
if (element.ValueKind != JsonValueKind.Number)
{
return false;
}
// Fast path: the JSON number is already written as an integer.
if (TryExactInteger(element, type, out value))
{
return true;
}
// Fallback: an integral-valued REAL. JS/Python edge gateways routinely serialize an
// integer as 5.0 (or 5E2), and System.Text.Json's TryGetInt32/64 are syntactic — they
// refuse it. Refusing here would mean an integer tag fed by such a gateway silently
// never receives data, so an exactly-integral real is accepted; a genuinely fractional
// value (5.5) is still refused rather than rounded.
return TryIntegralReal(element.GetRawText(), type, out value);
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:
case DriverDataType.Int32:
case DriverDataType.Int64:
case DriverDataType.UInt16:
case DriverDataType.UInt32:
case DriverDataType.UInt64:
// Same two-step as the JSON path: exact integer text first, then an integral-valued
// real ("5.0"), which a Scalar-format edge gateway emits just as readily.
if (TryExactIntegerText(trimmed, type, out value))
{
return true;
}
return TryIntegralReal(trimmed, type, out value);
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;
}
}
/// Reads a JSON number that is written as an exact integer of the requested type.
/// The JSON number.
/// The target integer type.
/// The boxed value when this returns .
/// when the number is an exact integer in range.
private static bool TryExactInteger(JsonElement element, DriverDataType type, out object? value)
{
value = type switch
{
DriverDataType.Int16 => element.TryGetInt16(out var v) ? v : null,
DriverDataType.Int32 => element.TryGetInt32(out var v) ? v : null,
DriverDataType.Int64 => element.TryGetInt64(out var v) ? v : null,
DriverDataType.UInt16 => element.TryGetUInt16(out var v) ? v : null,
DriverDataType.UInt32 => element.TryGetUInt32(out var v) ? v : null,
DriverDataType.UInt64 => element.TryGetUInt64(out var v) ? v : (object?)null,
_ => null,
};
return value is not null;
}
/// Parses text written as an exact integer of the requested type.
/// The trimmed text.
/// The target integer type.
/// The boxed value when this returns .
/// when the text is an exact integer in range.
private static bool TryExactIntegerText(string text, DriverDataType type, out object? value)
{
var inv = CultureInfo.InvariantCulture;
value = type switch
{
DriverDataType.Int16 => short.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.Int32 => int.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.Int64 => long.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt16 => ushort.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt32 => uint.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt64 => ulong.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : (object?)null,
_ => null,
};
return value is not null;
}
///
/// Accepts a real that is exactly integral (5.0, 5E2) as an integer of the
/// requested type. A fractional value is refused, never rounded — silently turning 5.5 into 5
/// or 6 would be a wrong value, which is worse than no value.
///
///
/// Parsed as rather than so the integrality test and
/// the conversion are exact across the whole Int64/UInt64 range, where double loses precision
/// above 2^53. Out-of-range values overflow the conversion and are refused.
///
/// The number's text (JSON raw text, or a Scalar payload).
/// The target integer type.
/// The boxed value when this returns .
/// when the text is an integral real in range.
private static bool TryIntegralReal(string text, DriverDataType type, out object? value)
{
value = null;
if (!decimal.TryParse(text, NumberStyles.Float, CultureInfo.InvariantCulture, out var d)
|| d != decimal.Truncate(d))
{
return false;
}
try
{
value = type switch
{
DriverDataType.Int16 => (short)d,
DriverDataType.Int32 => (int)d,
DriverDataType.Int64 => (long)d,
DriverDataType.UInt16 => (ushort)d,
DriverDataType.UInt32 => (uint)d,
DriverDataType.UInt64 => (ulong)d,
_ => null,
};
}
catch (OverflowException)
{
// decimal → integer conversions throw rather than wrap; out of range is a refusal.
value = null;
}
return value is not null;
}
///
/// 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 indexes, published as one immutable snapshot so the
/// dispatcher thread reads a consistent view while a redeploy rebuilds it.
///
///
/// Two topic indexes, and the difference is load-bearing. Delivery matches an
/// incoming published topic, which never carries a wildcard, so it uses
/// (concrete only) plus a comparer scan of .
/// Every path that instead reconstructs state from a subscribed filter string —
/// reconnect re-subscribe and per-topic degradation — must use , which is
/// keyed by the tag's authored topic including wildcard patterns, because that is what
/// lands in _subscribedTopics and in a SUBACK outcome. Looking a wildcard filter up in
/// the concrete index always misses, and a miss on those paths is silent: the tag simply stops
/// updating behind a healthy-looking connection, keeping its last Good value forever.
///
/// RawPath → definition; the v3 resolver's backing table.
///
/// Authored topic filter string → every definition bound to it, wildcard patterns
/// included. Keyed the same way _subscribedTopics and SUBACK outcomes are.
///
/// Concrete topic → its definitions; the delivery fast path.
/// Definitions whose authored topic carries an MQTT wildcard.
private sealed record AuthoredTable(
FrozenDictionary ByRawPath,
FrozenDictionary ByFilter,
FrozenDictionary ByExactTopic,
MqttTagDefinition[] Wildcards)
{
public static readonly AuthoredTable Empty =
Build(new Dictionary(StringComparer.Ordinal));
/// True when a topic string carries an MQTT wildcard and is therefore a filter, not a name.
/// The authored topic.
/// when it contains + or #.
public static bool IsWildcard(string topic) =>
topic.Contains(MqttTopicFilterComparer.SingleLevelWildcard, StringComparison.Ordinal)
|| topic.Contains(MqttTopicFilterComparer.MultiLevelWildcard, StringComparison.Ordinal);
/// Builds the snapshot: every def into ByFilter, concrete/wildcard split for delivery.
/// The mapped definitions, keyed by RawPath.
/// The immutable snapshot.
public static AuthoredTable Build(IReadOnlyDictionary byRawPath)
{
var byFilter = new Dictionary>(StringComparer.Ordinal);
var concrete = new Dictionary>(StringComparer.Ordinal);
var wildcards = new List();
foreach (var def in byRawPath.Values)
{
// EVERY def lands here, wildcard or not — this is the index the reconnect and degrade
// paths read, and omitting wildcards from it is exactly the C1 silent-death defect.
if (!byFilter.TryGetValue(def.Topic, out var all))
{
byFilter[def.Topic] = all = [];
}
all.Add(def);
if (IsWildcard(def.Topic))
{
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),
byFilter.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
concrete.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
[.. wildcards]);
}
}
}