using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; /// /// Shared doubles for the recorder tests added with the pass-flush / off-mailbox-append work: a FIFO outbox and a /// pass-aware writer that log every call into one ordered event list, so a test can assert the ORDER of writes, /// flushes and acks across both. Thread-safe — the recorder appends and drains off the actor thread. /// internal sealed class EventLog { private readonly Lock _gate = new(); private readonly List _events = new(); public void Add(string e) { lock (_gate) _events.Add(e); } public IReadOnlyList Snapshot() { lock (_gate) return _events.ToArray(); } } /// FIFO outbox honouring the truncate-through contract; appends can be made to yield or block. internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox { private readonly Lock _gate = new(); private readonly List _entries = new(); private readonly List _appended = new(); private int _appendBatchCalls; /// When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync. public TaskCompletionSource? AppendGate { get; set; } /// Completed by an append that has entered (before waiting on ). public TaskCompletionSource AppendEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); /// When set, CountAsync blocks on it — a hung outbox. public TaskCompletionSource? CountGate { get; set; } public int AppendBatchCalls { get { lock (_gate) return _appendBatchCalls; } } public long DroppedCount => 0; public IReadOnlyList Entries { get { lock (_gate) return _entries.ToArray(); } } /// Every entry ever appended, in append order (acks do not remove from it). public IReadOnlyList Appended { get { lock (_gate) return _appended.ToArray(); } } public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct); public async ValueTask AppendBatchAsync(IReadOnlyList entries, CancellationToken ct) { lock (_gate) _appendBatchCalls++; AppendEntered.TrySetResult(); await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, as a real fsync does if (AppendGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false); lock (_gate) { _entries.AddRange(entries); _appended.AddRange(entries); } log.Add($"append:{entries.Count}"); } public ValueTask> PeekBatchAsync(int max, CancellationToken ct) { lock (_gate) return ValueTask.FromResult>(_entries.Take(max).ToArray()); } public ValueTask RemoveAsync(Guid id, CancellationToken ct) { lock (_gate) { var i = _entries.FindIndex(e => e.Id == id); if (i >= 0) _entries.RemoveRange(0, i + 1); } log.Add("ack"); return ValueTask.CompletedTask; } public async ValueTask CountAsync(CancellationToken ct) { if (CountGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false); lock (_gate) return _entries.Count; } public void Dispose() { } } /// A pass-aware writer: logs every in-pass write and flush; plain writes are logged distinctly. internal sealed class LoggingPassWriter(EventLog log) : IHistorianPassWriter { private readonly HashSet _failTags = new(StringComparer.Ordinal); public Exception? FlushThrows { get; set; } public void Fail(string tag) { lock (_failTags) _failTags.Add(tag); } public Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) { log.Add($"plain:{tag}"); return Task.FromResult(true); } public async Task WriteLiveValuesInPassAsync(string tag, IReadOnlyList values, CancellationToken ct) { await Task.Delay(1, ct).ConfigureAwait(false); bool fail; lock (_failTags) fail = _failTags.Contains(tag); log.Add($"write:{tag}"); return !fail; } public async Task FlushAsync(CancellationToken ct) { await Task.Delay(1, ct).ConfigureAwait(false); log.Add("flush"); if (FlushThrows is { } ex) throw ex; } }