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

feat(mqtt): plain subscribe→OnDataChange with retained seed + ref resolver

Browse Source 
MqttSubscriptionManager is the P1 plain-MQTT ingest path: it holds the authored
RawPath → MqttTagDefinition table behind the shared EquipmentTagRefResolver,
establishes one SUBSCRIBE per distinct authored topic, and turns each inbound
message into a LastValueCache update plus an OnDataChange notification.

Load-bearing choices, each pinned by a falsified test:

- The published reference IS the RawPath (def.Name), never the topic and never
  the TagConfig blob — DriverHostActor's dual-namespace fan-out is RawPath-keyed,
  so any other key passes every unit test and delivers nothing in production.
- One message fans out to EVERY tag on its topic; several tags routinely share a
  topic with different jsonPaths, and stopping at the first match silently starves
  the rest.
- An unauthored topic raises nothing: MQTT ingest never auto-provisions.
- Wildcard-authored topics (accepted by the parser, warned at deploy) match via
  MQTTnet's own MqttTopicFilterComparer, so '+'/'#' behave as a broker would.
  Concrete topics stay a single lookup; the wildcard scan runs only when one exists.
- Retained seeds are honoured natively on MQTT 5.0 (retain handling) AND filtered
  client-side on the retain flag, because 3.1.1 brokers always replay.
- Nothing throws on MQTTnet's dispatcher thread: a malformed payload degrades that
  tag (BadDecodingError / BadTypeMismatch), and a throwing OnDataChange subscriber
  is contained without starving the tags behind it.
- The manager issues the FIRST subscribe itself — Reconnected deliberately does not
  fire after the initial ConnectAsync.
- Reconnect re-subscribe: total failure THROWS (MqttConnection then tears the
  session down and retries under backoff, rather than serving a connected-but-deaf
  driver); a partial SUBACK rejection degrades only the denied refs, because
  retrying forever over one ACL-denied topic would take every tag dark.

MqttConnection gains a MessageReceived observer (fired on the dispatcher, throwing
observers contained) and a bounded SubscribeAsync under its own linked-CTS deadline
— deliberately not MqttClientOptions.Timeout, which already governs every MQTTnet
op. A refused filter is an outcome, not an exception; only the SUBSCRIBE itself
failing throws. Classify() is parameterised by operation name (messages unchanged
for connect).

JSONPath is a documented subset ($, $.a.b, $['a'], $.a[0]); filters/slices/
recursive descent report a miss rather than pulling in a JSONPath package for an
expression form that selects a set where a tag needs one scalar.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 16:10:07 -04:00
parent f2a9ee5d93
commit d421487bcd
3 changed files with 1861 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
+216 -9
View File
@@ -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;
/// <summary>
/// Receives one inbound MQTT application message. Invoked on <b>MQTTnet's own dispatcher
/// thread</b>: 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.
/// </summary>
/// <param name="topic">The concrete topic the message arrived on (never a filter).</param>
/// <param name="payload">
/// 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.
/// </param>
/// <param name="retained">
/// The message's MQTT <c>retain</c> flag as delivered to <i>this</i> client, i.e. <c>true</c> for
/// the broker's stored last-known value replayed at subscribe time, and <c>false</c> for an
/// ordinary live publish. This is the retained <b>seed</b> signal.
/// </param>
public delegate void MqttMessageObserver(string topic, ReadOnlySpan<byte> payload, bool retained);
/// <summary>One topic filter to SUBSCRIBE, as the subscription manager wants it established.</summary>
/// <param name="Topic">The topic filter (concrete, or carrying MQTT wildcards).</param>
/// <param name="Qos">Requested QoS, 0–2.</param>
/// <param name="SeedRetained">
/// 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.
/// </param>
public sealed record MqttTopicSubscription(string Topic, int Qos, bool SeedRetained);
/// <summary>The broker's SUBACK verdict for one requested filter.</summary>
/// <param name="Topic">The filter this outcome answers.</param>
/// <param name="Granted">Whether the broker granted the subscription.</param>
/// <param name="Reason">The broker's reason code / string, for logs and diagnostics.</param>
public sealed record MqttSubscribeOutcome(string Topic, bool Granted, string Reason);
/// <summary>
/// The one seam through which the subscription manager establishes MQTT subscriptions.
/// <see cref="MqttConnection"/> is the production implementation; the interface exists so the
/// manager's SUBACK handling (per-filter grant / rejection, total failure) is exercisable without
/// a broker.
/// </summary>
public interface IMqttSubscribeTransport
{
/// <summary>
/// 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.
/// </summary>
/// <param name="filters">The filters to establish; never empty.</param>
/// <param name="cancellationToken">Caller cancellation, linked with the implementation's deadline.</param>
/// <returns>One outcome per requested filter.</returns>
Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
IReadOnlyList<MqttTopicSubscription> filters,
CancellationToken cancellationToken);
}
/// <summary>Lifecycle state of an <see cref="MqttConnection"/>.</summary>
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 <see cref="ObjectDisposedException"/>.
/// </para>
/// Subscription (Task 6) and Sparkplug rebirth-on-reconnect (Task 21, which hangs off
/// <see cref="Reconnected"/>) are deliberately not implemented here.
/// Sparkplug rebirth-on-reconnect (Task 21, which hangs off <see cref="Reconnected"/>) is
/// deliberately not implemented here.
/// </remarks>
public sealed class MqttConnection : IAsyncDisposable
public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport
{
private readonly string _driverId;
@@ -217,6 +273,19 @@ public sealed class MqttConnection : IAsyncDisposable
/// </remarks>
public event Func<CancellationToken, Task>? Reconnected;
/// <summary>
/// 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 <see cref="LastMessageUtc"/> and
/// handed on.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
public event MqttMessageObserver? MessageReceived;
/// <summary>Whether the underlying client currently holds an established MQTT session.</summary>
public bool IsConnected => _client?.IsConnected ?? false;
@@ -572,20 +641,28 @@ public sealed class MqttConnection : IAsyncDisposable
}
/// <summary>
/// Maps a connect failure onto this type's documented contract. MQTTnet wraps a cancelled
/// connect in <c>MqttConnectingFailedException</c> rather than letting the
/// Maps a bounded-operation failure onto this type's documented contract. MQTTnet wraps a
/// cancelled connect in <c>MqttConnectingFailedException</c> rather than letting the
/// <see cref="OperationCanceledException"/> surface, so the legs are told apart by which
/// token fired, not by exception type. Precedence: teardown, then caller intent, then the
/// deadline.
/// </summary>
private Exception Classify(Exception ex, CancellationToken cancellationToken, CancellationTokenSource deadline)
/// <param name="ex">The failure to classify.</param>
/// <param name="cancellationToken">The caller's token — cancelled means caller intent.</param>
/// <param name="deadline">The operation's own deadline source.</param>
/// <param name="operation">The operation name for the message ("connect", "subscribe").</param>
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<byte> 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;
}
/// <summary>
/// Issues one SUBSCRIBE carrying every requested filter, bounded by
/// <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> 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.
/// </summary>
/// <remarks>
/// The deadline is this method's own linked source, deliberately <b>not</b>
/// <see cref="MqttClientOptions.Timeout"/>: 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.
/// <para>
/// A rejected filter is <b>not</b> 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.
/// </para>
/// </remarks>
/// <param name="filters">The filters to establish.</param>
/// <param name="cancellationToken">Caller cancellation; linked with the subscribe deadline.</param>
/// <returns>One outcome per requested filter, in request order.</returns>
/// <exception cref="TimeoutException">The subscribe deadline elapsed.</exception>
/// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
/// <exception cref="ObjectDisposedException">The connection was disposed.</exception>
/// <exception cref="InvalidOperationException">There is no established session to subscribe on.</exception>
public async Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
IReadOnlyList<MqttTopicSubscription> 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<MqttClientSubscribeResultItem> ?? [.. result.Items];
var outcomes = new List<MqttSubscribeOutcome>(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;
}
/// <summary>Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value.</summary>
private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch
{
<= 0 => MqttQualityOfServiceLevel.AtMostOnce,
1 => MqttQualityOfServiceLevel.AtLeastOnce,
_ => MqttQualityOfServiceLevel.ExactlyOnce,
};
/// <summary>
/// The reconnect loop MQTTnet v5 no longer provides. Started on the first successful connect
/// and stopped by <see cref="DisposeAsync"/> cancelling the lifetime token; it owns every