using System.Runtime.CompilerServices;
using System.Threading.Channels;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
///
/// The shared test double: an Agent that serves the canned
/// Fixtures/probe.xml + Fixtures/current.xml documents, counts every call, records
/// its own disposal, and lets a test drive the /sample chunk sequence by hand.
///
///
///
/// Deterministic by construction — no timers, no sleeps, no polling. The
/// /sample leg reads from an unbounded that only a test
/// fills, and does not return until the driver has actually consumed
/// the chunk it wrote (each chunk carries its own completion source, signalled after the
/// enumerator's yield return resumes). A subscription test can therefore say "one
/// chunk has now been fully processed" as a fact rather than as a timing hope.
///
///
/// It honours the seam's stream-end contract (see
/// ): cancelling the token is the only way the
/// enumeration ends without throwing. Closing the scripted stream via
/// raises , exactly as the
/// production client does when an Agent drops the connection — so a pump written against the
/// fake cannot silently pass while mishandling the real one.
///
///
/// Disposal is observable () because "did the driver
/// actually release the client?" is a behaviour, not an implementation detail: a
/// ReinitializeAsync that re-points the Agent but leaks the old client keeps a live
/// connection pool per re-deploy, and nothing else in the test surface would notice.
///
///
internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
{
private readonly Channel _chunks = Channel.CreateUnbounded();
private readonly CancellationTokenSource _disposeCts = new();
private int _probeCallCount;
private int _currentCallCount;
private int _sampleCallCount;
private int _disposeCount;
private CannedAgentClient(MTConnectProbeModel probe, MTConnectStreamsResult current)
{
Probe = probe;
Current = current;
}
///
/// Builds a client serving the repo's canned fixtures — Fixtures/probe.xml for
/// /probe and Fixtures/current.xml for /current.
///
/// Probe-document fixture path, relative to the test binaries.
/// Streams-document fixture path, relative to the test binaries.
public static CannedAgentClient FromFixtures(
string probeFixture = "Fixtures/probe.xml",
string currentFixture = "Fixtures/current.xml") =>
new(
MTConnectProbeParser.Parse(File.ReadAllText(probeFixture)),
MTConnectStreamsParser.Parse(File.ReadAllText(currentFixture)));
/// Parses a streams fixture into a chunk a test can script onto the sample stream.
/// Streams-document fixture path, relative to the test binaries.
public static MTConnectStreamsResult Chunk(string fixture) =>
MTConnectStreamsParser.Parse(File.ReadAllText(fixture));
/// The document /probe answers with. Settable so a test can re-shape the model.
public MTConnectProbeModel Probe { get; set; }
///
/// The document /current answers with. Settable so a re-baseline (or an agent-restart
/// instanceId change) can be scripted between calls.
///
public MTConnectStreamsResult Current { get; set; }
/// When set, /probe throws this instead of answering.
public Exception? ProbeFailure { get; set; }
/// When set, /current throws this instead of answering.
public Exception? CurrentFailure { get; set; }
///
/// When set, the next /probe parks here until the test completes it, then the
/// gate clears itself so later calls answer immediately.
///
///
/// This is how a test holds a request "in flight" across a concurrent lifecycle change —
/// a shutdown or a re-initialize landing mid-fetch — with no timers and no races of its own.
/// The answer is captured when the request lands, not when it completes, so a test can
/// change meanwhile and still tell the two documents apart.
///
public TaskCompletionSource? ProbeGate { get; set; }
///
/// When set, /current does not answer until this source completes — an Agent that
/// accepted the request and then went quiet. The wait is cancellation-observing, so the
/// caller's own deadline is what ends it; the test never sleeps and never races a timer it
/// did not set. Leave null for the ordinary immediate answer.
///
public TaskCompletionSource? CurrentGate { get; set; }
/// Number of /probe requests issued against this client.
public int ProbeCallCount => Volatile.Read(ref _probeCallCount);
/// Number of /current requests issued against this client.
public int CurrentCallCount => Volatile.Read(ref _currentCallCount);
/// Number of /sample enumerations started against this client.
public int SampleCallCount => Volatile.Read(ref _sampleCallCount);
/// Number of calls (not clamped — a double-dispose is visible).
public int DisposeCount => Volatile.Read(ref _disposeCount);
/// Whether the client has been disposed at least once.
public bool IsDisposed => DisposeCount > 0;
/// The from sequence the most recent /sample enumeration was opened at.
public long? LastSampleFrom { get; private set; }
///
public async Task ProbeAsync(CancellationToken ct)
{
ObjectDisposedException.ThrowIf(IsDisposed, this);
ct.ThrowIfCancellationRequested();
Interlocked.Increment(ref _probeCallCount);
if (ProbeFailure is not null)
{
throw ProbeFailure;
}
// Captured now: the Agent answers with the document it held when the request landed.
var answer = Probe;
if (ProbeGate is { } gate)
{
ProbeGate = null;
await gate.Task.WaitAsync(ct).ConfigureAwait(false);
// A real client whose handler was disposed mid-request fails the request; so does this.
ObjectDisposedException.ThrowIf(IsDisposed, this);
}
return answer;
}
///
public async Task CurrentAsync(CancellationToken ct)
{
ObjectDisposedException.ThrowIf(IsDisposed, this);
ct.ThrowIfCancellationRequested();
Interlocked.Increment(ref _currentCallCount);
// The request was accepted; the answer is withheld until the test releases the gate or the
// caller's deadline cancels the token.
var gate = CurrentGate;
if (gate is not null)
{
await gate.Task.WaitAsync(ct).ConfigureAwait(false);
}
return CurrentFailure is null ? Current : throw CurrentFailure;
}
///
public async IAsyncEnumerable SampleAsync(
long from, [EnumeratorCancellation] CancellationToken ct)
{
ObjectDisposedException.ThrowIf(IsDisposed, this);
Interlocked.Increment(ref _sampleCallCount);
LastSampleFrom = from;
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
var delivered = 0L;
while (true)
{
ScriptedChunk scripted;
try
{
scripted = await _chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
}
catch (ChannelClosedException)
{
// Mirrors the production client: a stream that stops for any reason other than the
// caller cancelling is an exception, never a quiet end of enumeration.
throw new MTConnectStreamEndedException(
MTConnectStreamEndReason.ConnectionClosed, "canned://agent", delivered);
}
delivered++;
yield return scripted.Result;
scripted.Consumed.TrySetResult();
}
}
///
/// Queues a chunk onto the scripted /sample stream and waits until the consumer has
/// processed it. This is the deterministic replacement for "wait a bit and hope the pump
/// ran".
///
/// The chunk the Agent should send next.
/// A task completing once the enumerating consumer has moved past .
public Task PumpAsync(MTConnectStreamsResult chunk)
{
ArgumentNullException.ThrowIfNull(chunk);
var scripted = new ScriptedChunk(
chunk, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
if (!_chunks.Writer.TryWrite(scripted))
{
throw new InvalidOperationException("The scripted /sample stream is already closed.");
}
return scripted.Consumed.Task;
}
///
/// Closes the scripted /sample stream, which surfaces to the consumer as an
/// — the transient, reconnectable end.
///
public void EndStream() => _chunks.Writer.TryComplete();
///
public void Dispose()
{
Interlocked.Increment(ref _disposeCount);
if (DisposeCount > 1)
{
return;
}
_disposeCts.Cancel();
_chunks.Writer.TryComplete();
_disposeCts.Dispose();
}
private sealed record ScriptedChunk(MTConnectStreamsResult Result, TaskCompletionSource Consumed);
}