diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs new file mode 100644 index 00000000..f759707f --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs @@ -0,0 +1,63 @@ +using System.Collections.Concurrent; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// Thread-safe last-observed-value store backing for MQTT/Sparkplug B. +/// MQTT is a subscribe-first protocol — the driver holds one live broker connection and values +/// arrive pushed, not polled. But the OPC UA server still issues polled IReadable.ReadAsync +/// batches, so this cache is the bridge: the subscription path (message handler) calls +/// with the newest observed value per reference, and the read path serves +/// the most recent one via . +/// +/// +/// Keyed by RawPath — the v3 driver-reference identity (see +/// EquipmentTagRefResolver<TDef> and , +/// which is the RawPath) — never a topic/JSON-path-derived key. The parameter is +/// therefore named rawPath throughout rather than "topic" or "key" to keep that identity +/// fact visible at every call site. +/// +public sealed class LastValueCache +{ + // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value has + // been observed yet" (mirrored by CalculationDriver before its first evaluation, + // VirtualTagEngine for a freshly-materialised node, and OtOpcUaNodeManager / + // AddressSpaceApplier for a just-deployed variable). MQTT is subscribe-first, so an unseen + // RawPath is in exactly that state until its first publish arrives. Deliberately NOT + // GoodNoData — that code is reserved (see NullHistorianDataSource, OtOpcUaNodeManager + // HistoryRead paths) for "the historian window held no samples", a different question from + // "has this live reference ever been observed". + private const uint BadWaitingForInitialData = 0x80320000u; + + private readonly ConcurrentDictionary _values = new(); + + /// + /// Record the newest observed value for . Called from the MQTT + /// subscription/message-handling path. A null or empty + /// is a no-op — never throws. + /// + /// The RawPath identifying the tag. + /// The newest observed value/quality/timestamps for that tag. + public void Update(string rawPath, DataValueSnapshot snapshot) + { + if (string.IsNullOrEmpty(rawPath)) return; + _values[rawPath] = snapshot; + } + + /// + /// Read the last observed value for . Never throws: a + /// null/empty or a RawPath never observed both return a + /// snapshot rather than an exception, so a batch read + /// covering many references degrades per-reference instead of failing the whole call. + /// + /// The RawPath identifying the tag. + /// The last observed snapshot, or a BadWaitingForInitialData snapshot if none has been observed. + public DataValueSnapshot Read(string rawPath) + { + if (!string.IsNullOrEmpty(rawPath) && _values.TryGetValue(rawPath, out var snapshot)) + return snapshot; + + return new DataValueSnapshot(null, BadWaitingForInitialData, null, DateTime.UtcNow); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs new file mode 100644 index 00000000..b656fcd5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs @@ -0,0 +1,88 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +public sealed class LastValueCacheTests +{ + // OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value + // observed yet" (mirrored by CalculationDriver, VirtualTagEngine, FOCAS, and + // OtOpcUaNodeManager for freshly-materialised / not-yet-evaluated nodes). MQTT is + // subscribe-first, so an unseen RawPath is in exactly that state until its first publish. + private const uint BadWaitingForInitialData = 0x80320000u; + + [Fact] + public void Read_UnseenRef_ReturnsPerRefWaitingForInitialData_DoesNotThrow() + { + var cache = new LastValueCache(); + + var snap = cache.Read("factory/oven/temp"); + + snap.StatusCode.ShouldBe(BadWaitingForInitialData); + snap.Value.ShouldBeNull(); + } + + [Fact] + public void Update_ThenRead_ReturnsLastValue() + { + var cache = new LastValueCache(); + var now = DateTime.UtcNow; + + cache.Update("factory/oven/temp", new DataValueSnapshot(42.0, 0u, now, now)); + + var snap = cache.Read("factory/oven/temp"); + snap.Value.ShouldBe(42.0); + snap.StatusCode.ShouldBe(0u); + } + + [Fact] + public void Update_OverwritesPreviousValue_ForSameRawPath() + { + var cache = new LastValueCache(); + var t1 = DateTime.UtcNow; + var t2 = t1.AddSeconds(1); + + cache.Update("factory/oven/temp", new DataValueSnapshot(1.0, 0u, t1, t1)); + cache.Update("factory/oven/temp", new DataValueSnapshot(2.0, 0u, t2, t2)); + + cache.Read("factory/oven/temp").Value.ShouldBe(2.0); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + public void Read_NullOrEmptyRawPath_DoesNotThrow_ReturnsWaitingForInitialData(string? rawPath) + { + var cache = new LastValueCache(); + + var snap = cache.Read(rawPath!); + + snap.StatusCode.ShouldBe(BadWaitingForInitialData); + } + + [Fact] + public void Update_NullRawPath_DoesNotThrow() + { + var cache = new LastValueCache(); + + Should.NotThrow(() => cache.Update(null!, new DataValueSnapshot(1.0, 0u, DateTime.UtcNow, DateTime.UtcNow))); + } + + [Fact] + public void Read_DistinguishesUnseenFromGenuinelyBadObservedValue() + { + var cache = new LastValueCache(); + var now = DateTime.UtcNow; + const uint badDeviceFailure = 0x808B0000u; + + cache.Update("factory/oven/temp", new DataValueSnapshot(null, badDeviceFailure, null, now)); + + var observedBad = cache.Read("factory/oven/temp"); + var neverObserved = cache.Read("factory/oven/other"); + + observedBad.StatusCode.ShouldBe(badDeviceFailure); + neverObserved.StatusCode.ShouldBe(BadWaitingForInitialData); + observedBad.StatusCode.ShouldNotBe(neverObserved.StatusCode); + } +}