feat(phase7): wire RingBufferHistoryWriter as production IHistoryWriter for virtual tags (Gap 5)
Closes Phase 7 Gap 5: VirtualTagEngine called IHistoryWriter.Record per evaluation when Historize=true but Phase7EngineComposer always passed NullHistoryWriter, so virtual-tag history was computed but never persisted. The fix: - New RingBufferHistoryWriter implements both IHistoryWriter (write port for the evaluation pipeline) and IHistorianDataSource (read port for IHistoryRouter so OPC UA HistoryRead on virtual-tag nodes resolves here). Maintains one bounded ring buffer (1000 samples, configurable) per tag path; Record() is O(1) and never blocks evaluation. - Phase7EngineComposer.Compose now accepts IHistoryRouter? and, when any VirtualTagDefinition.Historize=true, creates a RingBufferHistoryWriter, passes it to VirtualTagEngine as historyWriter, adds it to the disposables list, and registers it under the "virtual:" prefix in the router for HistoryRead dispatch. - Phase7Composer accepts IHistoryRouter? from DI (already registered as singleton in Program.cs) and threads it through to Phase7EngineComposer.Compose. - NullHistoryWriter remains as fallback when no tags request historization. - 16 new unit tests in RingBufferHistoryWriterTests.cs cover ring-buffer semantics, eviction, per-tag isolation, ReadRawAsync windowing, IHistorianDataSource stubs, router registration, and the Historize=false / null-router fallback paths. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -7,6 +7,7 @@ using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
|||||||
using ZB.MOM.WW.OtOpcUa.Core.Hosting;
|
using ZB.MOM.WW.OtOpcUa.Core.Hosting;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.OpcUa;
|
using ZB.MOM.WW.OtOpcUa.Core.OpcUa;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Server.History;
|
||||||
using ZB.MOM.WW.OtOpcUa.Server.OpcUa;
|
using ZB.MOM.WW.OtOpcUa.Server.OpcUa;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
||||||
@@ -42,6 +43,7 @@ public sealed class Phase7Composer : IAsyncDisposable
|
|||||||
private readonly DriverEquipmentContentRegistry _equipmentRegistry;
|
private readonly DriverEquipmentContentRegistry _equipmentRegistry;
|
||||||
private readonly IAlarmHistorianSink _historianSink;
|
private readonly IAlarmHistorianSink _historianSink;
|
||||||
private readonly IAlarmHistorianWriter? _injectedWriter;
|
private readonly IAlarmHistorianWriter? _injectedWriter;
|
||||||
|
private readonly IHistoryRouter? _historyRouter;
|
||||||
private readonly ILoggerFactory _loggerFactory;
|
private readonly ILoggerFactory _loggerFactory;
|
||||||
private readonly Serilog.ILogger _scriptLogger;
|
private readonly Serilog.ILogger _scriptLogger;
|
||||||
private readonly ILogger<Phase7Composer> _logger;
|
private readonly ILogger<Phase7Composer> _logger;
|
||||||
@@ -61,13 +63,15 @@ public sealed class Phase7Composer : IAsyncDisposable
|
|||||||
ILoggerFactory loggerFactory,
|
ILoggerFactory loggerFactory,
|
||||||
Serilog.ILogger scriptLogger,
|
Serilog.ILogger scriptLogger,
|
||||||
ILogger<Phase7Composer> logger,
|
ILogger<Phase7Composer> logger,
|
||||||
IAlarmHistorianWriter? injectedWriter = null)
|
IAlarmHistorianWriter? injectedWriter = null,
|
||||||
|
IHistoryRouter? historyRouter = null)
|
||||||
{
|
{
|
||||||
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
|
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
|
||||||
_driverHost = driverHost ?? throw new ArgumentNullException(nameof(driverHost));
|
_driverHost = driverHost ?? throw new ArgumentNullException(nameof(driverHost));
|
||||||
_equipmentRegistry = equipmentRegistry ?? throw new ArgumentNullException(nameof(equipmentRegistry));
|
_equipmentRegistry = equipmentRegistry ?? throw new ArgumentNullException(nameof(equipmentRegistry));
|
||||||
_historianSink = historianSink ?? throw new ArgumentNullException(nameof(historianSink));
|
_historianSink = historianSink ?? throw new ArgumentNullException(nameof(historianSink));
|
||||||
_injectedWriter = injectedWriter;
|
_injectedWriter = injectedWriter;
|
||||||
|
_historyRouter = historyRouter;
|
||||||
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
|
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
|
||||||
_scriptLogger = scriptLogger ?? throw new ArgumentNullException(nameof(scriptLogger));
|
_scriptLogger = scriptLogger ?? throw new ArgumentNullException(nameof(scriptLogger));
|
||||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||||
@@ -117,7 +121,8 @@ public sealed class Phase7Composer : IAsyncDisposable
|
|||||||
alarmStateStore: new InMemoryAlarmStateStore(),
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
historianSink: historianSink,
|
historianSink: historianSink,
|
||||||
rootScriptLogger: _scriptLogger,
|
rootScriptLogger: _scriptLogger,
|
||||||
loggerFactory: _loggerFactory);
|
loggerFactory: _loggerFactory,
|
||||||
|
historyRouter: _historyRouter);
|
||||||
|
|
||||||
_logger.LogInformation(
|
_logger.LogInformation(
|
||||||
"Phase 7: composed engines from generation {Gen} — {Vt} virtual tag(s), {Al} scripted alarm(s), {Sc} script(s)",
|
"Phase 7: composed engines from generation {Gen} — {Vt} virtual tag(s), {Al} scripted alarm(s), {Sc} script(s)",
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
|||||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Server.History;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
||||||
|
|
||||||
@@ -32,6 +33,14 @@ namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
|||||||
/// </remarks>
|
/// </remarks>
|
||||||
public static class Phase7EngineComposer
|
public static class Phase7EngineComposer
|
||||||
{
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Prefix used when registering the virtual-tag ring-buffer history source in
|
||||||
|
/// <see cref="IHistoryRouter"/>. All virtual-tag UNS paths are prefixed with this
|
||||||
|
/// string so the router resolves reads to the <see cref="RingBufferHistoryWriter"/>
|
||||||
|
/// rather than a driver-owned historian.
|
||||||
|
/// </summary>
|
||||||
|
public const string VirtualTagHistoryPrefix = "virtual:";
|
||||||
|
|
||||||
public static Phase7ComposedSources Compose(
|
public static Phase7ComposedSources Compose(
|
||||||
IReadOnlyList<Script> scripts,
|
IReadOnlyList<Script> scripts,
|
||||||
IReadOnlyList<VirtualTag> virtualTags,
|
IReadOnlyList<VirtualTag> virtualTags,
|
||||||
@@ -40,7 +49,8 @@ public static class Phase7EngineComposer
|
|||||||
IAlarmStateStore alarmStateStore,
|
IAlarmStateStore alarmStateStore,
|
||||||
IAlarmHistorianSink historianSink,
|
IAlarmHistorianSink historianSink,
|
||||||
Serilog.ILogger rootScriptLogger,
|
Serilog.ILogger rootScriptLogger,
|
||||||
ILoggerFactory loggerFactory)
|
ILoggerFactory loggerFactory,
|
||||||
|
IHistoryRouter? historyRouter = null)
|
||||||
{
|
{
|
||||||
ArgumentNullException.ThrowIfNull(scripts);
|
ArgumentNullException.ThrowIfNull(scripts);
|
||||||
ArgumentNullException.ThrowIfNull(virtualTags);
|
ArgumentNullException.ThrowIfNull(virtualTags);
|
||||||
@@ -64,10 +74,40 @@ public static class Phase7EngineComposer
|
|||||||
// Engines take Serilog.ILogger — each engine gets its own so rolling-file emissions
|
// Engines take Serilog.ILogger — each engine gets its own so rolling-file emissions
|
||||||
// stay keyed to the right source in the scripts-*.log.
|
// stay keyed to the right source in the scripts-*.log.
|
||||||
VirtualTagSource? vtSource = null;
|
VirtualTagSource? vtSource = null;
|
||||||
|
RingBufferHistoryWriter? ringWriter = null;
|
||||||
if (virtualTags.Count > 0)
|
if (virtualTags.Count > 0)
|
||||||
{
|
{
|
||||||
var vtDefs = ProjectVirtualTags(virtualTags, scriptById).ToList();
|
var vtDefs = ProjectVirtualTags(virtualTags, scriptById).ToList();
|
||||||
var vtEngine = new VirtualTagEngine(upstream, scriptLoggerFactory, rootScriptLogger);
|
|
||||||
|
// Gap 5 closure (task #28): wire a real IHistoryWriter when any tag has
|
||||||
|
// Historize=true. RingBufferHistoryWriter is a bounded in-process ring buffer
|
||||||
|
// that also implements IHistorianDataSource so OPC UA HistoryRead on virtual
|
||||||
|
// nodes resolves here. Register with the IHistoryRouter under a dedicated
|
||||||
|
// prefix so the DriverNodeManager's history dispatch finds it.
|
||||||
|
IHistoryWriter historyWriter = NullHistoryWriter.Instance;
|
||||||
|
var hasHistorize = vtDefs.Any(d => d.Historize);
|
||||||
|
if (hasHistorize)
|
||||||
|
{
|
||||||
|
ringWriter = new RingBufferHistoryWriter();
|
||||||
|
historyWriter = ringWriter;
|
||||||
|
disposables.Add(ringWriter);
|
||||||
|
|
||||||
|
if (historyRouter is not null)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
historyRouter.Register(VirtualTagHistoryPrefix, ringWriter);
|
||||||
|
}
|
||||||
|
catch (InvalidOperationException)
|
||||||
|
{
|
||||||
|
// Already registered (e.g. engine reload). Leave the existing entry —
|
||||||
|
// both registrations refer to the same in-process buffer type.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var vtEngine = new VirtualTagEngine(upstream, scriptLoggerFactory, rootScriptLogger,
|
||||||
|
historyWriter: historyWriter);
|
||||||
vtEngine.Load(vtDefs);
|
vtEngine.Load(vtDefs);
|
||||||
vtSource = new VirtualTagSource(vtEngine);
|
vtSource = new VirtualTagSource(vtEngine);
|
||||||
disposables.Add(vtEngine);
|
disposables.Add(vtEngine);
|
||||||
|
|||||||
@@ -0,0 +1,235 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Production <see cref="IHistoryWriter"/> for virtual-tag evaluations (Gap 5 closure,
|
||||||
|
/// task #28). Every evaluation result from a <c>Historize=true</c> virtual tag is
|
||||||
|
/// appended to an in-process per-tag ring buffer. The same instance also implements
|
||||||
|
/// <see cref="IHistorianDataSource"/> so it can be registered on the server-level
|
||||||
|
/// <see cref="ZB.MOM.WW.OtOpcUa.Server.History.IHistoryRouter"/> — OPC UA HistoryRead
|
||||||
|
/// requests for virtual-tag nodes then resolve here.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// Design rationale:
|
||||||
|
/// <list type="bullet">
|
||||||
|
/// <item><description>
|
||||||
|
/// No new external process or DB is needed: the ring buffer lives in-process
|
||||||
|
/// and survives across evaluation cycles for the lifetime of the server.
|
||||||
|
/// </description></item>
|
||||||
|
/// <item><description>
|
||||||
|
/// <see cref="Record"/> is synchronous and O(1) — it never blocks the
|
||||||
|
/// evaluation pipeline. Per <see cref="IHistoryWriter"/> contract the caller
|
||||||
|
/// treats it as fire-and-forget.
|
||||||
|
/// </description></item>
|
||||||
|
/// <item><description>
|
||||||
|
/// The buffer is bounded by <see cref="MaxSamplesPerTag"/> (default 1 000).
|
||||||
|
/// Older samples are evicted silently when the tag writes past the limit, which
|
||||||
|
/// matches the "hot ring" semantics typical of in-process historian buffers:
|
||||||
|
/// the most-recent N seconds of data is always accessible for HistoryRead even
|
||||||
|
/// in the absence of a persistent historian.
|
||||||
|
/// </description></item>
|
||||||
|
/// <item><description>
|
||||||
|
/// Thread safety: <see cref="Record"/> uses an interlocked write pointer so
|
||||||
|
/// concurrent evaluation callbacks (rare but possible during cascade + timer
|
||||||
|
/// races) don't corrupt the buffer.
|
||||||
|
/// </description></item>
|
||||||
|
/// </list>
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <see cref="ReadRawAsync"/> supports the OPC UA HistoryRead service's raw-values
|
||||||
|
/// mode. Processed (aggregate), at-time, and event read modes return empty results
|
||||||
|
/// with no error — virtual tags are scalar real-time values, not time-series stored
|
||||||
|
/// at external resolution.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// Lifecycle: <see cref="Dispose"/> clears every buffer. The caller
|
||||||
|
/// (<see cref="Phase7Composer.DisposeAsync"/>) disposes this instance after the
|
||||||
|
/// engine that writes to it has been disposed, so no further <see cref="Record"/>
|
||||||
|
/// calls can arrive after dispose.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class RingBufferHistoryWriter : IHistoryWriter, IHistorianDataSource
|
||||||
|
{
|
||||||
|
/// <summary>Maximum samples retained per tag path. Older samples are evicted when full.</summary>
|
||||||
|
public const int MaxSamplesPerTag = 1_000;
|
||||||
|
|
||||||
|
private readonly ConcurrentDictionary<string, TagRingBuffer> _buffers =
|
||||||
|
new(StringComparer.Ordinal);
|
||||||
|
private readonly int _capacity;
|
||||||
|
private bool _disposed;
|
||||||
|
|
||||||
|
/// <param name="capacity">
|
||||||
|
/// Per-tag sample capacity. Tests inject a smaller value to keep fixtures
|
||||||
|
/// small; production uses the default <see cref="MaxSamplesPerTag"/>.
|
||||||
|
/// </param>
|
||||||
|
public RingBufferHistoryWriter(int capacity = MaxSamplesPerTag)
|
||||||
|
{
|
||||||
|
if (capacity < 1) throw new ArgumentOutOfRangeException(nameof(capacity), "Capacity must be ≥ 1.");
|
||||||
|
_capacity = capacity;
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== IHistoryWriter =====
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Records one evaluation result. Called by <see cref="VirtualTagEngine"/> on every
|
||||||
|
/// evaluation when <c>Historize=true</c>. O(1), never blocks.
|
||||||
|
/// </summary>
|
||||||
|
public void Record(string path, DataValueSnapshot value)
|
||||||
|
{
|
||||||
|
if (_disposed) return; // graceful shutdown — silently drop
|
||||||
|
|
||||||
|
var buffer = _buffers.GetOrAdd(path, _ => new TagRingBuffer(_capacity));
|
||||||
|
buffer.Write(value);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== IHistorianDataSource =====
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Returns samples in the ring buffer whose source timestamp falls within
|
||||||
|
/// [<paramref name="startUtc"/>, <paramref name="endUtc"/>), newest-first.
|
||||||
|
/// <paramref name="maxValuesPerNode"/> caps the result count.
|
||||||
|
/// </summary>
|
||||||
|
public Task<HistoryReadResult> ReadRawAsync(
|
||||||
|
string fullReference,
|
||||||
|
DateTime startUtc,
|
||||||
|
DateTime endUtc,
|
||||||
|
uint maxValuesPerNode,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (!_buffers.TryGetValue(fullReference, out var buffer))
|
||||||
|
return Task.FromResult(new HistoryReadResult([], null));
|
||||||
|
|
||||||
|
var all = buffer.Snapshot();
|
||||||
|
var limit = (int)Math.Min(maxValuesPerNode, (uint)all.Length);
|
||||||
|
var result = new List<DataValueSnapshot>(limit);
|
||||||
|
|
||||||
|
foreach (var snap in all)
|
||||||
|
{
|
||||||
|
if (result.Count >= limit) break;
|
||||||
|
var ts = snap.SourceTimestampUtc ?? snap.ServerTimestampUtc;
|
||||||
|
if (ts >= startUtc && ts < endUtc)
|
||||||
|
result.Add(snap);
|
||||||
|
}
|
||||||
|
|
||||||
|
return Task.FromResult(new HistoryReadResult(result, null));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
/// <remarks>
|
||||||
|
/// Virtual tags do not carry aggregate history — returns an empty result rather than
|
||||||
|
/// failing so OPC UA clients that request processed history receive a graceful empty
|
||||||
|
/// response instead of a Bad status for the whole node.
|
||||||
|
/// </remarks>
|
||||||
|
public Task<HistoryReadResult> ReadProcessedAsync(
|
||||||
|
string fullReference,
|
||||||
|
DateTime startUtc,
|
||||||
|
DateTime endUtc,
|
||||||
|
TimeSpan interval,
|
||||||
|
HistoryAggregateType aggregate,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
=> Task.FromResult(new HistoryReadResult([], null));
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<HistoryReadResult> ReadAtTimeAsync(
|
||||||
|
string fullReference,
|
||||||
|
IReadOnlyList<DateTime> timestampsUtc,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
=> Task.FromResult(new HistoryReadResult([], null));
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
||||||
|
string? sourceName,
|
||||||
|
DateTime startUtc,
|
||||||
|
DateTime endUtc,
|
||||||
|
int maxEvents,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
=> Task.FromResult(new HistoricalEventsResult([], null));
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public HistorianHealthSnapshot GetHealthSnapshot() => new(
|
||||||
|
TotalQueries: 0,
|
||||||
|
TotalSuccesses: 0,
|
||||||
|
TotalFailures: 0,
|
||||||
|
ConsecutiveFailures: 0,
|
||||||
|
LastSuccessTime: null,
|
||||||
|
LastFailureTime: null,
|
||||||
|
LastError: null,
|
||||||
|
ProcessConnectionOpen: true,
|
||||||
|
EventConnectionOpen: false,
|
||||||
|
ActiveProcessNode: null,
|
||||||
|
ActiveEventNode: null,
|
||||||
|
Nodes: []);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Number of distinct tag paths that have received at least one recorded sample.
|
||||||
|
/// Exposed for diagnostics and tests.
|
||||||
|
/// </summary>
|
||||||
|
public int TagCount => _buffers.Count;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Returns a snapshot of all samples currently in the buffer for <paramref name="path"/>,
|
||||||
|
/// or an empty array when the path has no recorded samples. Exposed for diagnostics.
|
||||||
|
/// </summary>
|
||||||
|
public DataValueSnapshot[] GetSnapshots(string path)
|
||||||
|
=> _buffers.TryGetValue(path, out var buf) ? buf.Snapshot() : [];
|
||||||
|
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
if (_disposed) return;
|
||||||
|
_disposed = true;
|
||||||
|
_buffers.Clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== Inner ring buffer =====
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Bounded FIFO ring buffer with O(1) write. Reads return a snapshot (array copy)
|
||||||
|
/// of all current samples in insertion order, oldest first.
|
||||||
|
/// </summary>
|
||||||
|
private sealed class TagRingBuffer
|
||||||
|
{
|
||||||
|
private readonly DataValueSnapshot?[] _slots;
|
||||||
|
private int _head; // next write position (wraps)
|
||||||
|
private int _count; // how many valid entries (≤ capacity)
|
||||||
|
|
||||||
|
public TagRingBuffer(int capacity)
|
||||||
|
{
|
||||||
|
_slots = new DataValueSnapshot[capacity];
|
||||||
|
}
|
||||||
|
|
||||||
|
public void Write(DataValueSnapshot value)
|
||||||
|
{
|
||||||
|
lock (_slots)
|
||||||
|
{
|
||||||
|
_slots[_head] = value;
|
||||||
|
_head = (_head + 1) % _slots.Length;
|
||||||
|
if (_count < _slots.Length) _count++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Returns all current samples in insertion order (oldest → newest), as a
|
||||||
|
/// snapshot array. Thread-safe: takes the lock for the copy.
|
||||||
|
/// </summary>
|
||||||
|
public DataValueSnapshot[] Snapshot()
|
||||||
|
{
|
||||||
|
lock (_slots)
|
||||||
|
{
|
||||||
|
if (_count == 0) return [];
|
||||||
|
|
||||||
|
var result = new DataValueSnapshot[_count];
|
||||||
|
// The oldest entry is at (_head - _count + capacity) % capacity.
|
||||||
|
var start = (_head - _count + _slots.Length) % _slots.Length;
|
||||||
|
for (var i = 0; i < _count; i++)
|
||||||
|
{
|
||||||
|
result[i] = _slots[(start + i) % _slots.Length]!;
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,308 @@
|
|||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Serilog;
|
||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Server.History;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Task #28 — Gap 5 closure: verifies that <see cref="RingBufferHistoryWriter"/>
|
||||||
|
/// correctly records virtual-tag evaluation results and returns them via the
|
||||||
|
/// <see cref="IHistorianDataSource"/> read interface; and that
|
||||||
|
/// <see cref="Phase7EngineComposer.Compose"/> wires the writer and registers it
|
||||||
|
/// with an <see cref="IHistoryRouter"/> when <c>Historize=true</c> tags are present.
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "Unit")]
|
||||||
|
public sealed class RingBufferHistoryWriterTests
|
||||||
|
{
|
||||||
|
private static readonly DateTime T0 = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||||
|
private static readonly DateTime T1 = T0.AddSeconds(1);
|
||||||
|
private static readonly DateTime T2 = T0.AddSeconds(2);
|
||||||
|
private static readonly DateTime T3 = T0.AddSeconds(3);
|
||||||
|
|
||||||
|
private static DataValueSnapshot Snap(double value, DateTime ts) =>
|
||||||
|
new(value, 0u, ts, ts);
|
||||||
|
|
||||||
|
// ===== RingBufferHistoryWriter unit tests =====
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Record_stores_sample_retrievable_via_GetSnapshots()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/area/line/eq/Tag1", Snap(42.0, T0));
|
||||||
|
|
||||||
|
var snaps = writer.GetSnapshots("/area/line/eq/Tag1");
|
||||||
|
snaps.Length.ShouldBe(1);
|
||||||
|
snaps[0].Value.ShouldBe(42.0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Record_multiple_samples_preserves_insertion_order()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
writer.Record("/t", Snap(2.0, T1));
|
||||||
|
writer.Record("/t", Snap(3.0, T2));
|
||||||
|
|
||||||
|
var snaps = writer.GetSnapshots("/t");
|
||||||
|
snaps.Length.ShouldBe(3);
|
||||||
|
snaps[0].Value.ShouldBe(1.0);
|
||||||
|
snaps[1].Value.ShouldBe(2.0);
|
||||||
|
snaps[2].Value.ShouldBe(3.0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Record_evicts_oldest_when_capacity_exceeded()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter(capacity: 3);
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
writer.Record("/t", Snap(2.0, T1));
|
||||||
|
writer.Record("/t", Snap(3.0, T2));
|
||||||
|
writer.Record("/t", Snap(4.0, T3)); // evicts 1.0
|
||||||
|
|
||||||
|
var snaps = writer.GetSnapshots("/t");
|
||||||
|
snaps.Length.ShouldBe(3);
|
||||||
|
snaps[0].Value.ShouldBe(2.0, "oldest evicted");
|
||||||
|
snaps[2].Value.ShouldBe(4.0, "newest present");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Record_maintains_separate_buffers_per_tag_path()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/area/eq/TagA", Snap(10.0, T0));
|
||||||
|
writer.Record("/area/eq/TagB", Snap(20.0, T0));
|
||||||
|
|
||||||
|
writer.GetSnapshots("/area/eq/TagA").Single().Value.ShouldBe(10.0);
|
||||||
|
writer.GetSnapshots("/area/eq/TagB").Single().Value.ShouldBe(20.0);
|
||||||
|
writer.TagCount.ShouldBe(2);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void GetSnapshots_returns_empty_for_unknown_path()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.GetSnapshots("/not/a/path").ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Dispose_clears_buffers_and_subsequent_Record_is_silently_ignored()
|
||||||
|
{
|
||||||
|
var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
writer.Dispose();
|
||||||
|
|
||||||
|
// After dispose, Record must silently drop (no exception).
|
||||||
|
Should.NotThrow(() => writer.Record("/t", Snap(2.0, T1)));
|
||||||
|
// GetSnapshots post-dispose returns empty (buffers cleared).
|
||||||
|
writer.GetSnapshots("/t").ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== IHistorianDataSource.ReadRawAsync tests =====
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadRawAsync_returns_empty_for_unknown_path()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
var result = await writer.ReadRawAsync("notexists", T0, T3, 100, default);
|
||||||
|
result.Samples.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadRawAsync_returns_samples_in_time_window()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
writer.Record("/t", Snap(2.0, T1));
|
||||||
|
writer.Record("/t", Snap(3.0, T2));
|
||||||
|
|
||||||
|
// Window [T0, T2) — T2 excluded (half-open interval).
|
||||||
|
var result = await writer.ReadRawAsync("/t", T0, T2, 100, default);
|
||||||
|
result.Samples.Count.ShouldBe(2);
|
||||||
|
result.Samples[0].Value.ShouldBe(1.0);
|
||||||
|
result.Samples[1].Value.ShouldBe(2.0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadRawAsync_respects_maxValuesPerNode_cap()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
writer.Record("/t", Snap(2.0, T1));
|
||||||
|
writer.Record("/t", Snap(3.0, T2));
|
||||||
|
|
||||||
|
var result = await writer.ReadRawAsync("/t", T0, T3, maxValuesPerNode: 2, default);
|
||||||
|
result.Samples.Count.ShouldBe(2);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadProcessedAsync_returns_empty_result()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
var result = await writer.ReadProcessedAsync("/t", T0, T3, TimeSpan.FromSeconds(1),
|
||||||
|
HistoryAggregateType.Average, default);
|
||||||
|
result.Samples.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadAtTimeAsync_returns_empty_result()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
writer.Record("/t", Snap(1.0, T0));
|
||||||
|
var result = await writer.ReadAtTimeAsync("/t", [T0, T1], default);
|
||||||
|
result.Samples.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadEventsAsync_returns_empty_result()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
var result = await writer.ReadEventsAsync(null, T0, T3, 100, default);
|
||||||
|
result.Events.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void GetHealthSnapshot_returns_connected_non_null_snapshot()
|
||||||
|
{
|
||||||
|
using var writer = new RingBufferHistoryWriter();
|
||||||
|
var health = writer.GetHealthSnapshot();
|
||||||
|
health.ShouldNotBeNull();
|
||||||
|
health.ProcessConnectionOpen.ShouldBeTrue("ring buffer is always available in-process");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== Phase7EngineComposer wiring tests =====
|
||||||
|
|
||||||
|
private static Script ScriptRow(string id, string source) => new()
|
||||||
|
{
|
||||||
|
ScriptRowId = Guid.NewGuid(), GenerationId = 1,
|
||||||
|
ScriptId = id, Name = id, SourceCode = source, SourceHash = "h",
|
||||||
|
};
|
||||||
|
|
||||||
|
private static VirtualTag VtRow(string id, string scriptId, bool historize = false) => new()
|
||||||
|
{
|
||||||
|
VirtualTagRowId = Guid.NewGuid(), GenerationId = 1,
|
||||||
|
VirtualTagId = id, EquipmentId = "eq-1", Name = id,
|
||||||
|
DataType = "Float32", ScriptId = scriptId,
|
||||||
|
Historize = historize,
|
||||||
|
};
|
||||||
|
|
||||||
|
private static ScriptedAlarm AlarmRow(string id, string scriptId) => new()
|
||||||
|
{
|
||||||
|
ScriptedAlarmRowId = Guid.NewGuid(), GenerationId = 1,
|
||||||
|
ScriptedAlarmId = id, EquipmentId = "eq-1", Name = id,
|
||||||
|
AlarmType = "LimitAlarm", Severity = 500,
|
||||||
|
MessageTemplate = "x", PredicateScriptId = scriptId,
|
||||||
|
};
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Compose_without_Historize_uses_NullHistoryWriter_and_skips_router_registration()
|
||||||
|
{
|
||||||
|
using var router = new HistoryRouter();
|
||||||
|
var scripts = new[] { ScriptRow("s1", "return 1;") };
|
||||||
|
var vtags = new[] { VtRow("vt-1", "s1", historize: false) };
|
||||||
|
|
||||||
|
var result = Phase7EngineComposer.Compose(
|
||||||
|
scripts, vtags, [],
|
||||||
|
upstream: new CachedTagUpstreamSource(),
|
||||||
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
|
historianSink: NullAlarmHistorianSink.Instance,
|
||||||
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
||||||
|
loggerFactory: NullLoggerFactory.Instance,
|
||||||
|
historyRouter: router);
|
||||||
|
|
||||||
|
// Router should not have a "virtual:" prefix entry when no Historize=true tags.
|
||||||
|
router.Resolve(Phase7EngineComposer.VirtualTagHistoryPrefix + "vt-1").ShouldBeNull();
|
||||||
|
result.VirtualReadable.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Compose_with_Historize_true_registers_RingBufferHistoryWriter_in_router()
|
||||||
|
{
|
||||||
|
using var router = new HistoryRouter();
|
||||||
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
||||||
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
||||||
|
|
||||||
|
var result = Phase7EngineComposer.Compose(
|
||||||
|
scripts, vtags, [],
|
||||||
|
upstream: new CachedTagUpstreamSource(),
|
||||||
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
|
historianSink: NullAlarmHistorianSink.Instance,
|
||||||
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
||||||
|
loggerFactory: NullLoggerFactory.Instance,
|
||||||
|
historyRouter: router);
|
||||||
|
|
||||||
|
// The "virtual:" prefix must resolve to a RingBufferHistoryWriter instance.
|
||||||
|
var source = router.Resolve(Phase7EngineComposer.VirtualTagHistoryPrefix + "vt-hist");
|
||||||
|
source.ShouldNotBeNull("router should have the ring-buffer source registered under 'virtual:' prefix");
|
||||||
|
source.ShouldBeOfType<RingBufferHistoryWriter>();
|
||||||
|
|
||||||
|
result.VirtualReadable.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Compose_with_Historize_true_but_no_router_does_not_throw()
|
||||||
|
{
|
||||||
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
||||||
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
||||||
|
|
||||||
|
// historyRouter = null — should still work, just no registration.
|
||||||
|
Should.NotThrow(() => Phase7EngineComposer.Compose(
|
||||||
|
scripts, vtags, [],
|
||||||
|
upstream: new CachedTagUpstreamSource(),
|
||||||
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
|
historianSink: NullAlarmHistorianSink.Instance,
|
||||||
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
||||||
|
loggerFactory: NullLoggerFactory.Instance,
|
||||||
|
historyRouter: null));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Compose_with_Historize_true_router_already_registered_does_not_throw()
|
||||||
|
{
|
||||||
|
// Simulate a reload scenario where the prefix is already registered.
|
||||||
|
using var router = new HistoryRouter();
|
||||||
|
using var priorWriter = new RingBufferHistoryWriter();
|
||||||
|
router.Register(Phase7EngineComposer.VirtualTagHistoryPrefix, priorWriter);
|
||||||
|
|
||||||
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
||||||
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
||||||
|
|
||||||
|
// Second compose call — should tolerate the duplicate without throwing.
|
||||||
|
Should.NotThrow(() => Phase7EngineComposer.Compose(
|
||||||
|
scripts, vtags, [],
|
||||||
|
upstream: new CachedTagUpstreamSource(),
|
||||||
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
|
historianSink: NullAlarmHistorianSink.Instance,
|
||||||
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
||||||
|
loggerFactory: NullLoggerFactory.Instance,
|
||||||
|
historyRouter: router));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Compose_RingBufferHistoryWriter_is_in_disposables_list()
|
||||||
|
{
|
||||||
|
using var router = new HistoryRouter();
|
||||||
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
||||||
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
||||||
|
|
||||||
|
var result = Phase7EngineComposer.Compose(
|
||||||
|
scripts, vtags, [],
|
||||||
|
upstream: new CachedTagUpstreamSource(),
|
||||||
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
||||||
|
historianSink: NullAlarmHistorianSink.Instance,
|
||||||
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
||||||
|
loggerFactory: NullLoggerFactory.Instance,
|
||||||
|
historyRouter: router);
|
||||||
|
|
||||||
|
// The RingBufferHistoryWriter must be tracked in Disposables so Phase7Composer.DisposeAsync
|
||||||
|
// clears the ring buffer on shutdown.
|
||||||
|
result.Disposables.ShouldContain(d => d is RingBufferHistoryWriter);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user