/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; namespace Opc.Ua.Client.Subscriptions.Fakes { /// /// Hand-rolled fake for . Records every /// invocation and lets tests override return behaviour via callback /// fields. Replaces Mock<IMessageAckQueue>. /// internal sealed class FakeMessageAckQueue : IMessageAckQueue { /// /// An owned snapshot of queued acknowledgements, captured under the queue's mutation lock. /// public List QueuedAcks { get { lock (m_lock) { return [.. m_queuedAcks]; } } } public List CompletedSubscriptions { get; } = []; /// /// Instances passed to , so tests can /// assert the subscription is retired by identity and not only by /// the server assigned id. /// public List CompletedProcessors { get; } = []; public int UpdateCalls { get; private set; } public int PublishingQuiescenceCalls { get; private set; } /// /// Optional override for . If null, returns /// completed. /// public Func? OnQueueAsync { get; set; } /// /// Optional override for . If null, /// returns completed. /// public Func? OnCompleteAsync { get; set; } /// /// Records an acknowledgement as if a processor had queued it. /// public ValueTask QueueAsync(SubscriptionAcknowledgement ack, CancellationToken ct = default) { return QueueAsync(null, 0, ack, ct); } /// public ValueTask QueueAsync(IMessageProcessor? owner, long generation, SubscriptionAcknowledgement ack, CancellationToken ct = default) { lock (m_lock) { m_queuedAcks.Add(ack); m_queuedGenerations.Add(generation); } return OnQueueAsync?.Invoke(ack, ct) ?? default; } /// /// The sequence numbers whose acknowledgement was tracked before delivery. /// public List TrackedSequenceNumbers { get { lock (m_lock) { return [.. m_tracked]; } } } /// public void TrackAcknowledgement(IMessageProcessor owner, long generation, uint sequenceNumber) { lock (m_lock) { m_tracked.Add(sequenceNumber); } } /// /// The owner generation each entry of was stamped with. /// public List QueuedAckGenerations { get { lock (m_lock) { return [.. m_queuedGenerations]; } } } /// /// Removes every queued acknowledgement for . /// /// The number of removed acknowledgements. public int ClearQueuedAcks(uint subscriptionId) { lock (m_lock) { int removed = 0; for (int i = m_queuedAcks.Count - 1; i >= 0; i--) { if (m_queuedAcks[i].SubscriptionId == subscriptionId) { m_queuedAcks.RemoveAt(i); m_queuedGenerations.RemoveAt(i); removed++; } } return removed; } } /// /// Waits until at least acknowledgements have /// been queued. The processor under test dispatches a notification /// before it queues the acknowledgement for it, so a test that only /// waits for the dispatch races the enqueue that follows it. /// /// /// The number of queued acknowledgements to wait for. /// /// How long to wait before giving up. /// /// The acknowledgements did not arrive in time. Failing here rather /// than returning quietly keeps the diagnosis at the right level: a /// caller that went on to assert the count would only report the /// mismatch, which cannot distinguish an acknowledgement that was /// never queued from one that merely arrived late - and if that /// assertion is ever loosened, a silent return would hide the race /// this helper exists to close. /// public async Task WaitForQueuedAckAsync(int count, int timeoutMs = 5000) { const int kPollIntervalMs = 10; var timeout = TimeSpan.FromMilliseconds(timeoutMs); long start = TimeProvider.System.GetTimestamp(); while (true) { int queued; lock (m_lock) { queued = m_queuedAcks.Count; } if (queued >= count) { return; } // Measure against a monotonic clock rather than accumulating // the poll interval: Task.Delay routinely overshoots, so // counting the requested interval silently stretches the // effective timeout well past what the caller asked for. if (TimeProvider.System.GetElapsedTime(start) >= timeout) { throw new TimeoutException( $"Expected at least {count} queued acknowledgement(s) within " + $"{timeout.TotalSeconds:0.##}s but only {queued} arrived."); } await Task.Delay(kPollIntervalMs).ConfigureAwait(false); } } public ValueTask CompleteAsync(IMessageProcessor subscription, uint subscriptionId, CancellationToken ct = default) { CompletedProcessors.Add(subscription); CompletedSubscriptions.Add(subscriptionId); return OnCompleteAsync?.Invoke(subscriptionId, ct) ?? default; } public ValueTask RunWithPublishingQuiescedAsync( Func operation, CancellationToken ct = default) { PublishingQuiescenceCalls++; return operation(ct); } /// /// Optional override for , /// e.g. to simulate a session that is not available. If null, runs /// the operation. /// public Func, CancellationToken, ValueTask>? OnRunWithSessionAvailableAsync { get; set; } public ValueTask RunWithSessionAvailableAsync( Func operation, CancellationToken ct = default) { return OnRunWithSessionAvailableAsync?.Invoke(operation, ct) ?? operation(ct); } /// /// Records the subscription ids dropped via /// so tests can assert /// stale-ack pruning happened during recovery. /// public List DroppedSubscriptions { get; } = []; public int DropPendingForSubscription(IMessageProcessor owner, uint subscriptionId) { // Like the manager, queued entries are not scanned: they fail send-time // validation because their owner's generation advanced. DroppedSubscriptions.Add(subscriptionId); return 0; } /// /// Ids reported through . /// public List CreatedSubscriptions { get; } = []; /// /// Creation times reported through . /// public List CreatedAt { get; } = []; /// public void OnSubscriptionCreated(IMessageProcessor owner, uint subscriptionId, DateTimeUtc createdAt = default) { CreatedSubscriptions.Add(subscriptionId); CreatedAt.Add(createdAt); } /// /// Test-controlled answer for /// . Defaults to /// true, i.e. the caller still owns the id. /// public Func? OnOwnsSubscriptionId { get; set; } /// /// Ids was asked about. /// public List OwnershipChecks { get; } = []; public bool OwnsSubscriptionId(IMessageProcessor subscription, uint subscriptionId) { OwnershipChecks.Add(subscriptionId); return OnOwnsSubscriptionId?.Invoke(subscription, subscriptionId) ?? true; } public void Update() { UpdateCalls++; } /// /// Test-controlled value for the /// setter the /// subscription dispatcher reads to decide whether to perform the /// pooled-notification reuse walk. /// public bool PoolNotifications { get; set; } /// public CancellationToken ShutdownToken { get; set; } private readonly Lock m_lock = new(); /// The queue's mutable storage, accessed only under . private readonly List m_queuedAcks = []; private readonly List m_queuedGenerations = []; private readonly List m_tracked = []; } }