diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
index abf760f6..8a4ed603 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
@@ -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);
+ }
+
+ ///
+ /// Returns true when and represent
+ /// different values. Mirrors PollGroupEngine.ValuesAreDifferent — deliberately duplicated
+ /// rather than shared, because that one is private to the poll-loop engine this push-based driver
+ /// does not run through.
+ ///
+ ///
+ /// 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.
+ ///
+ /// The last cached value for the reference.
+ /// The value just extracted from the inbound message.
+ /// true when the two differ.
+ 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);
}
///
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
index 9c80c78d..6ee9255a 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs
@@ -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.
+ // ---------------------------------------------------------------------------------
+
+ ///
+ /// 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 cached
+ /// — the republish freshens the snapshot so IReadable.ReadAsync stays current — but no
+ /// OnDataChange is raised. "Cached but not raised" is an established state here; see
+ /// .
+ ///
+ [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();
+ 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);
+ }
+
+ ///
+ /// The falsification control for :
+ /// suppression must be value-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.
+ ///
+ [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();
+ 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]);
+ }
+
+ ///
+ /// Suppression compares status as well as value, so a quality transition on an otherwise
+ /// unchanged reference still publishes. The case that matters is
+ /// DegradeTopic: a Good value followed by a refused subscription must reach the
+ /// subscriber as Bad, or the client keeps a stale Good forever.
+ ///
+ [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);
+ }
+
/// Records every subscribe call so dedupe / re-subscribe can be asserted without a broker.
private sealed class FakeTransport : IMqttSubscribeTransport
{