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

perf(mqtt): suppress unchanged-value republish fan-out

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 21:42:00 -04:00
parent 8d91833fed
commit 2c708f5560
2 changed files with 126 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
+33 -1
View File
@@ -1,3 +1,4 @@
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Frozen;
using System.Globalization;
@@ -748,8 +749,39 @@ public sealed class MqttSubscriptionManager
snapshot.StatusCode);
}
// Unchanged-value suppression. Every poll driver gets this dedupe for free at the edge from
// PollGroupEngine; MQTT is push-based, so a broker that timer-republishes an identical payload
// would otherwise re-trigger the ENTIRE downstream fan-out (actor hop → dependency mux → OPC UA
// sink write) per message. The cache is still freshened — a read must serve the current
// timestamp — and only the notification is skipped. "Cached but not raised" is already a valid
// state on this path (see Raise: an authored-but-unsubscribed tag does exactly that).
var prior = Values.Read(def.Name);
var unchanged = prior.StatusCode == snapshot.StatusCode && !ValuesAreDifferent(prior.Value, snapshot.Value);
Values.Update(def.Name, snapshot);
Raise(def.Name, snapshot);
if (!unchanged) Raise(def.Name, snapshot);
}
/// <summary>
/// Returns <c>true</c> when <paramref name="previous"/> and <paramref name="current"/> represent
/// different values. Mirrors <c>PollGroupEngine.ValuesAreDifferent</c> — deliberately duplicated
/// rather than shared, because that one is private to the poll-loop engine this push-based driver
/// does not run through.
/// </summary>
/// <remarks>
/// Arrays are compared structurally: a JSON array payload decodes to a fresh array instance on
/// every message, so reference equality would report every republish as a change and suppress
/// nothing at all for exactly the tags whose payloads are largest.
/// </remarks>
/// <param name="previous">The last cached value for the reference.</param>
/// <param name="current">The value just extracted from the inbound message.</param>
/// <returns><c>true</c> when the two differ.</returns>
private static bool ValuesAreDifferent(object? previous, object? current)
{
if (previous is Array prevArr && current is Array currArr)
return !StructuralComparisons.StructuralEqualityComparer.Equals(prevArr, currArr);
return !Equals(previous, current);
}
/// <summary>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs
+93
View File
@@ -964,6 +964,99 @@ public sealed class MqttSubscriptionManagerTests
transport.Calls.Count.ShouldBe(1);
}
// ---------------------------------------------------------------------------------
// Unchanged-value suppression — MQTT is push-based, so the edge dedupe every poll driver
// gets from PollGroupEngine has to live here instead.
// ---------------------------------------------------------------------------------
/// <summary>
/// A broker that timer-republishes an identical payload must not re-trigger the whole
/// downstream fan-out (actor hop → mux → OPC UA sink write). The value is still <i>cached</i>
/// — the republish freshens the snapshot so <c>IReadable.ReadAsync</c> stays current — but no
/// <c>OnDataChange</c> is raised. "Cached but not raised" is an established state here; see
/// <see cref="HandleMessage_AuthoredButUnsubscribedTag_CachesValue_RaisesNothing"/>.
/// </summary>
[Fact]
public async Task An_identical_republish_raises_nothing()
{
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);
var raised = new List<DataValueSnapshot>();
mgr.OnDataChange += (_, e) => raised.Add(e.Snapshot);
mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
raised.Count.ShouldBe(1);
raised[0].Value.ShouldBe(21.5);
// The suppressed republish still WROTE the cache: the cached snapshot is the second message's
// instance, not the one that was raised. Asserting on the timestamp alone would be flaky —
// two dispatcher-thread calls can share a clock tick — while the instance identity is exact.
var cached = mgr.Values.Read(OvenTempPath);
cached.Value.ShouldBe(21.5);
cached.StatusCode.ShouldBe(Good);
cached.ShouldNotBeSameAs(raised[0]);
cached.ServerTimestampUtc.ShouldBeGreaterThanOrEqualTo(raised[0].ServerTimestampUtc);
}
/// <summary>
/// The falsification control for <see cref="An_identical_republish_raises_nothing"/>:
/// suppression must be <i>value</i>-based, not once-only. An implementation that simply
/// raised the first message per reference and went quiet passes that test and takes the
/// driver permanently dark.
/// </summary>
[Fact]
public async Task A_changed_value_after_a_suppressed_republish_raises()
{
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);
var raised = new List<object?>();
mgr.OnDataChange += (_, e) => raised.Add(e.Snapshot.Value);
mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // suppressed
mgr.HandleMessage(OvenTopic, """{"value":22.5}"""u8.ToArray(), retained: false);
mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // back again — a real change
raised.ShouldBe([21.5, 22.5, 21.5]);
}
/// <summary>
/// Suppression compares status as well as value, so a quality transition on an otherwise
/// unchanged reference still publishes. The case that matters is
/// <c>DegradeTopic</c>: a Good value followed by a refused subscription must reach the
/// subscriber as Bad, or the client keeps a stale Good forever.
/// </summary>
[Fact]
public async Task A_status_transition_still_raises()
{
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, "21.5"u8.ToArray(), retained: false);
var raised = new List<(string Reference, uint Status)>();
mgr.OnDataChange += (_, e) => raised.Add((e.FullReference, e.Snapshot.StatusCode));
// Only the oven-temp topic is refused, so the re-subscribe is a PARTIAL rejection and does
// not throw — one topic is still granted.
transport.Reject = t => t == OvenTopic;
await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
raised.ShouldContain((OvenTempPath, BadCommunicationError));
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError);
}
/// <summary>Records every subscribe call so dedupe / re-subscribe can be asserted without a broker.</summary>
private sealed class FakeTransport : IMqttSubscribeTransport
{