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;
}
}