feat(mqtt): last-value cache backing IReadable (per-ref no-data)
LastValueCache bridges MQTT's subscribe-first push model to the OPC UA server's polled IReadable.ReadAsync: the subscription path calls Update() per RawPath, the read path calls Read(). Read never throws; an unseen RawPath returns BadWaitingForInitialData (0x80320000) rather than an exception or null, so a batch covering many references degrades per-ref instead of failing wholesale. Deviates from the plan's GoodNoData snippet: GoodNoData is reserved repo-wide for "the historian window held no samples" (NullHistorianDataSource, OtOpcUaNodeManager HistoryRead paths); BadWaitingForInitialData is the established convention for "no live value observed yet" (CalculationDriver, VirtualTagEngine, FOCAS, AddressSpaceApplier). Keyed by RawPath per the v3 driver-reference identity, not a topic/JSON-path-derived key. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,63 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
|
||||
|
||||
/// <summary>
|
||||
/// Thread-safe last-observed-value store backing <see cref="IReadable"/> 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 <c>IReadable.ReadAsync</c>
|
||||
/// batches, so this cache is the bridge: the subscription path (message handler) calls
|
||||
/// <see cref="Update"/> with the newest observed value per reference, and the read path serves
|
||||
/// the most recent one via <see cref="Read"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Keyed by <c>RawPath</c> — the v3 driver-reference identity (see
|
||||
/// <c>EquipmentTagRefResolver<TDef></c> and <see cref="MqttTagDefinition.Name"/>,
|
||||
/// which <b>is</b> the RawPath) — never a topic/JSON-path-derived key. The parameter is
|
||||
/// therefore named <c>rawPath</c> throughout rather than "topic" or "key" to keep that identity
|
||||
/// fact visible at every call site.
|
||||
/// </remarks>
|
||||
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<string, DataValueSnapshot> _values = new();
|
||||
|
||||
/// <summary>
|
||||
/// Record the newest observed value for <paramref name="rawPath"/>. Called from the MQTT
|
||||
/// subscription/message-handling path. A <c>null</c> or empty <paramref name="rawPath"/>
|
||||
/// is a no-op — never throws.
|
||||
/// </summary>
|
||||
/// <param name="rawPath">The RawPath identifying the tag.</param>
|
||||
/// <param name="snapshot">The newest observed value/quality/timestamps for that tag.</param>
|
||||
public void Update(string rawPath, DataValueSnapshot snapshot)
|
||||
{
|
||||
if (string.IsNullOrEmpty(rawPath)) return;
|
||||
_values[rawPath] = snapshot;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Read the last observed value for <paramref name="rawPath"/>. Never throws: a
|
||||
/// <c>null</c>/empty <paramref name="rawPath"/> or a RawPath never observed both return a
|
||||
/// <see cref="BadWaitingForInitialData"/> snapshot rather than an exception, so a batch read
|
||||
/// covering many references degrades per-reference instead of failing the whole call.
|
||||
/// </summary>
|
||||
/// <param name="rawPath">The RawPath identifying the tag.</param>
|
||||
/// <returns>The last observed snapshot, or a BadWaitingForInitialData snapshot if none has been observed.</returns>
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user