UA-.NETStandard/tests/Opc.Ua.Client.Tests/ModelChange/ModelChangeTrackerTests.cs

/* ========================================================================
 * 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.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Moq;
using NUnit.Framework;
using Opc.Ua.Client.ModelChange;
using Opc.Ua.Client.Subscriptions;
using Opc.Ua.Client.Subscriptions.MonitoredItems;
using Opc.Ua.Client.Subscriptions.Streaming;
using MonitoringOptions = Opc.Ua.Client.Subscriptions.MonitoredItems.MonitoredItemOptions;

// Tests run on the default TaskScheduler so CA2007's sync-context
// risk does not apply. The local FakeStreamingSubscription is a
// no-op IAsyncDisposable fake with nothing to dispose; CA2000's leak
// warning does not apply.
#pragma warning disable CA2007, CA2000

namespace Opc.Ua.Client.Tests.ModelChange
{
    /// <summary>
    /// Unit tests for <see cref="ModelChangeTracker"/>. Drives the
    /// tracker through a fake <see cref="IStreamingSubscription"/>
    /// that pushes <see cref="EventNotification"/>s on demand and
    /// uses a <see cref="Mock{INodeCache}"/> to observe cache
    /// invalidation calls.
    /// </summary>
    [TestFixture]
    [Category("Client")]
    [Category("ModelChange")]
    [Parallelizable]
    public sealed class ModelChangeTrackerTests
    {
        [Test]
        public void ConstructorWithNullStreamingThrowsArgumentNullException()
        {
            Assert.That(
                () => new ModelChangeTracker(null!),
                Throws.ArgumentNullException);
        }

        [Test]
        public async Task IsTrackingIsFalseBeforeStartAsync()
        {
            var fake = new FakeStreamingSubscription();
            await using var tracker = new ModelChangeTracker(fake);

            Assert.That(tracker.IsTracking, Is.False);
        }

        [Test]
        public async Task IsTrackingIsTrueAfterStartAsync()
        {
            var fake = new FakeStreamingSubscription();
            await using var tracker = new ModelChangeTracker(fake);

            await tracker.StartTrackingAsync().ConfigureAwait(false);

            Assert.That(tracker.IsTracking, Is.True);
        }

        [Test]
        public async Task StartTrackingAsyncWaitsUntilEventMonitoredItemIsCreated()
        {
            var fake = new FakeStreamingSubscription(eventMonitoredItemCreated: false);
            await using var tracker = new ModelChangeTracker(fake);

            int raised = 0;
            tracker.ModelChanged += (_, _) => Interlocked.Increment(ref raised);

            Task startTask = tracker.StartTrackingAsync().AsTask();
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);
            await Task.Delay(50).ConfigureAwait(false);

            Assert.That(startTask.IsCompleted, Is.False);

            fake.CreateEventMonitoredItem();
            await startTask.ConfigureAwait(false);

            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(default, default, Variant.From("ready"))));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(raised, Is.EqualTo(1));
        }

        [Test]
        public async Task StartTrackingAsyncIsIdempotent()
        {
            var fake = new FakeStreamingSubscription();
            await using var tracker = new ModelChangeTracker(fake);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);
            await tracker.StartTrackingAsync().ConfigureAwait(false);

            Assert.That(tracker.IsTracking, Is.True);

            // The pump task is started exactly once — a second
            // call must not subscribe again.
            Assert.That(fake.SubscribeCallCount, Is.EqualTo(1));
        }

        [Test]
        public async Task StopTrackingAsyncCompletesPumpTaskAndFlipsIsTrackingToFalse()
        {
            var fake = new FakeStreamingSubscription();
            await using var tracker = new ModelChangeTracker(fake);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);
            await tracker.StopTrackingAsync().ConfigureAwait(false);

            Assert.That(tracker.IsTracking, Is.False);
            Assert.That(fake.PumpCancellationObserved, Is.True);
        }

        [Test]
        public async Task DisposeAsyncIsIdempotent()
        {
            var fake = new FakeStreamingSubscription();
            var tracker = new ModelChangeTracker(fake);
            await tracker.StartTrackingAsync().ConfigureAwait(false);

            await tracker.DisposeAsync().ConfigureAwait(false);
            // Second dispose must be a no-op (does not throw).
            await tracker.DisposeAsync().ConfigureAwait(false);

            Assert.That(tracker.IsTracking, Is.False);
        }

        [Test]
        public async Task DisposeAsyncCallsStopTracking()
        {
            var fake = new FakeStreamingSubscription();
            var tracker = new ModelChangeTracker(fake);
            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            await tracker.DisposeAsync().ConfigureAwait(false);

            Assert.That(tracker.IsTracking, Is.False);
            Assert.That(fake.PumpCancellationObserved, Is.True);
        }

        [Test]
        public async Task NotificationWithFewerThanThreeFieldsDoesNotRaiseModelChanged()
        {
            var fake = new FakeStreamingSubscription();
            var cache = new Mock<INodeCache>(MockBehavior.Strict);
            await using var tracker = new ModelChangeTracker(fake, cache.Object);

            int raised = 0;
            tracker.ModelChanged += (_, _) => Interlocked.Increment(ref raised);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            // Fields.Count < 3 — fast-path return, no event, no cache call.
            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default)));

            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(raised, Is.Zero);
            cache.VerifyNoOtherCalls();
        }

        [Test]
        public async Task NotificationWithMalformedFieldsTriggersFullCacheInvalidation()
        {
            var fake = new FakeStreamingSubscription();
            var cache = new Mock<INodeCache>(MockBehavior.Loose);
            await using var tracker = new ModelChangeTracker(fake, cache.Object);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            // fields[2] is a scalar string — definitively NOT an
            // ExtensionObject[] of ModelChangeStructureDataType, so the
            // tracker must fall back to full cache invalidation.
            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(
                    default,
                    default,
                    Variant.From("not-an-extension-object-array"))));

            await fake.QuiesceAsync().ConfigureAwait(false);

            cache.Verify(c => c.Clear(), Times.Once);
            cache.Verify(c => c.InvalidateNode(It.IsAny<NodeId>()), Times.Never);

            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.RequiresFullCacheInvalidation, Is.True);
            Assert.That(observed.Changes, Is.Empty);
        }

        [Test]
        public async Task NotificationWithNullChangesFieldTriggersFullCacheInvalidation()
        {
            // changesVariant.IsNull -> boxed is null -> still NOT
            // ExtensionObject[] -> full invalidation path.
            var fake = new FakeStreamingSubscription();
            var cache = new Mock<INodeCache>(MockBehavior.Loose);
            await using var tracker = new ModelChangeTracker(fake, cache.Object);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped<Variant>(default, default, default)));

            await fake.QuiesceAsync().ConfigureAwait(false);

            cache.Verify(c => c.Clear(), Times.Once);
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.RequiresFullCacheInvalidation, Is.True);
        }

        [Test]
        public async Task CacheInvalidationExceptionsAreLoggedButDoNotStopTracking()
        {
            var fake = new FakeStreamingSubscription();
            var cache = new Mock<INodeCache>();
            cache.Setup(c => c.Clear())
                .Throws(new InvalidOperationException("boom"));

            await using var tracker = new ModelChangeTracker(fake, cache.Object);

            int raised = 0;
            tracker.ModelChanged += (_, _) => Interlocked.Increment(ref raised);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            // First malformed notification — Clear() throws, must be swallowed,
            // ModelChanged must still fire.
            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(
                    default, default, Variant.From("garbage"))));
            await fake.QuiesceAsync().ConfigureAwait(false);

            // Second notification — pump must still be alive.
            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(
                    default, default, Variant.From("garbage-2"))));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(raised, Is.EqualTo(2));
            Assert.That(tracker.IsTracking, Is.True);
        }

        [Test]
        public async Task SubscriberHandlerExceptionsAreSwallowed()
        {
            var fake = new FakeStreamingSubscription();
            await using var tracker = new ModelChangeTracker(fake);

            int raised = 0;
            tracker.ModelChanged += (_, _) =>
            {
                Interlocked.Increment(ref raised);
                throw new InvalidOperationException("subscriber failure");
            };

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(
                    default, default, Variant.From("garbage-1"))));
            await fake.QuiesceAsync().ConfigureAwait(false);

            // Pump still alive — a second notification is delivered.
            fake.Push(new EventNotification(
                null,
                ArrayOf.Wrapped(
                    default, default, Variant.From("garbage-2"))));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(raised, Is.EqualTo(2));
            Assert.That(tracker.IsTracking, Is.True);
        }

        /// <summary>
        /// Regression for #4100. A server that appends a namespace uri
        /// while running reports the change as a
        /// <c>BaseModelChangeEvent</c> — no per-node detail — so the
        /// tracker must re-read the namespace table.
        /// </summary>
        [Test]
        public async Task UnqualifiedModelChangeRefreshesNamespaceTable()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher();
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            namespaces.OnRefresh = table => table.Append("urn:live:model");

            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default, default)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.EqualTo(1));
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.NamespaceTableRefreshed, Is.True);
            Assert.That(namespaces.NamespaceUris.GetIndex("urn:live:model"),
                Is.EqualTo(1));
        }

        [Test]
        public async Task ResolvableModelChangeDoesNotRefreshNamespaceTable()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher("urn:one");
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(CreateGeneralModelChangeNotification(
                new NodeId(42u, 1), ObjectTypeIds.BaseObjectType));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.Zero);
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.NamespaceTableRefreshed, Is.False);
            Assert.That(observed.Changes, Has.Count.EqualTo(1));
            Assert.That(observed.Changes[0].AffectedNode,
                Is.EqualTo(new NodeId(42u, 1)));
        }

        [Test]
        public async Task ChangeAffectingServerNamespaceArrayRefreshesNamespaceTable()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher();
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(CreateGeneralModelChangeNotification(
                VariableIds.Server_NamespaceArray, NodeId.Null));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.EqualTo(1));
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.NamespaceTableRefreshed, Is.True);
        }

        [Test]
        public async Task ChangeWithUnknownNamespaceIndexRefreshesNamespaceTable()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher();
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            // Namespace index 3 is beyond the single known uri, so the
            // client provably cannot resolve the affected node.
            fake.Push(CreateGeneralModelChangeNotification(
                new NodeId(42u, 3), NodeId.Null));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.EqualTo(1));
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.NamespaceTableRefreshed, Is.True);
        }

        [Test]
        public async Task ChangeWithUnknownTypeDefinitionNamespaceRefreshesNamespaceTable()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher();
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(CreateGeneralModelChangeNotification(
                new NodeId(42u), new NodeId(99u, 2)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.EqualTo(1));
        }

        [Test]
        public async Task NamespaceTableIsRefreshedBeforeTheNodeCacheIsInvalidated()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher();
            var order = new List<string>();
            var cache = new Mock<INodeCache>(MockBehavior.Loose);
            cache.Setup(c => c.Clear()).Callback(() => order.Add("clear"));
            namespaces.OnRefresh = _ => order.Add("refresh");

            await using var tracker = new ModelChangeTracker(
                fake, cache.Object, logger: null, namespaceTables: namespaces);

            tracker.ModelChanged += (_, _) => order.Add("event");

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default, default)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(order, Is.EqualTo(s_refreshBeforeInvalidateOrder));
        }

        [Test]
        public async Task NamespaceTableRefreshFailureIsSwallowedAndTrackingContinues()
        {
            var fake = new FakeStreamingSubscription();
            var namespaces = new FakeNamespaceTableRefresher
            {
                OnRefresh = _ => throw new ServiceResultException(
                    StatusCodes.BadNotConnected)
            };
            await using var tracker = new ModelChangeTracker(
                fake, nodeCache: null, logger: null, namespaceTables: namespaces);

            var observed = new List<ModelChangedEventArgs>();
            tracker.ModelChanged += (_, e) => observed.Add(e);

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default, default)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            // Pump must still be alive for the next notification.
            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default, default)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            Assert.That(namespaces.RefreshCount, Is.EqualTo(2));
            Assert.That(observed, Has.Count.EqualTo(2));
            Assert.That(observed[0].NamespaceTableRefreshed, Is.False);
            Assert.That(tracker.IsTracking, Is.True);
        }

        [Test]
        public async Task WithoutANamespaceTableRefresherNoRefreshIsAttempted()
        {
            var fake = new FakeStreamingSubscription();
            var cache = new Mock<INodeCache>(MockBehavior.Loose);
            await using var tracker = new ModelChangeTracker(fake, cache.Object);

            ModelChangedEventArgs? observed = null;
            tracker.ModelChanged += (_, e) => observed = e;

            await tracker.StartTrackingAsync().ConfigureAwait(false);
            await fake.WaitForSubscribeAsync().ConfigureAwait(false);

            fake.Push(new EventNotification(
                null, ArrayOf.Wrapped<Variant>(default, default, default)));
            await fake.QuiesceAsync().ConfigureAwait(false);

            cache.Verify(c => c.Clear(), Times.Once);
            Assert.That(observed, Is.Not.Null);
            Assert.That(observed!.NamespaceTableRefreshed, Is.False);
        }

        /// <summary>
        /// Expected call order when a model change triggers both a
        /// namespace refresh and a full cache invalidation.
        /// </summary>
        private static readonly string[] s_refreshBeforeInvalidateOrder =
            ["refresh", "clear", "event"];

        /// <summary>
        /// Builds a <c>GeneralModelChangeEvent</c> shaped notification
        /// carrying a single change entry.
        /// </summary>
        private static EventNotification CreateGeneralModelChangeNotification(
            NodeId affected,
            NodeId affectedType)
        {
            var change = new ModelChangeStructureDataType
            {
                Affected = affected,
                AffectedType = affectedType,
                Verb = (byte)ModelChangeVerb.NodeAdded
            };
            ArrayOf<ExtensionObject> changes = new ExtensionObject[]
            {
                new(change)
            };
            return new EventNotification(
                null,
                ArrayOf.Wrapped(default, default, Variant.From(changes)));
        }

        /// <summary>
        /// Hand-rolled fake for <see cref="INamespaceTableRefresher"/>.
        /// Counts refresh calls and lets a test mutate the table (or
        /// throw) from inside the refresh.
        /// </summary>
        private sealed class FakeNamespaceTableRefresher : INamespaceTableRefresher
        {
            public FakeNamespaceTableRefresher(params string[] additionalNamespaceUris)
            {
                NamespaceUris = new NamespaceTable(
                    new[] { Namespaces.OpcUa }.Concat(additionalNamespaceUris));
            }

            public NamespaceTable NamespaceUris { get; }

            public int RefreshCount => Volatile.Read(ref m_refreshCount);

            /// <summary>
            /// Invoked on every refresh so a test can simulate the
            /// server side table growing, or a failure.
            /// </summary>
            public Action<NamespaceTable>? OnRefresh { get; set; }

            public Task FetchNamespaceTablesAsync(CancellationToken ct = default)
            {
                Interlocked.Increment(ref m_refreshCount);
                OnRefresh?.Invoke(NamespaceUris);
                return Task.CompletedTask;
            }

            private int m_refreshCount;
        }

        /// <summary>
        /// Fake streaming subscription that exposes a deterministic
        /// <see cref="EventNotification"/> channel for tests. Each call
        /// to <see cref="SubscribeEventsAsync"/> increments
        /// <see cref="SubscribeCallCount"/>, signals
        /// <see cref="WaitForSubscribeAsync"/>, and then yields every
        /// notification pushed via <see cref="Push"/>. The enumerator
        /// completes when <see cref="Complete"/> is called or the
        /// supplied cancellation token fires.
        /// </summary>
        private sealed class FakeStreamingSubscription :
            IStreamingSubscription,
            IStreamingSubscriptionReadiness
        {
            private readonly Channel<EventNotification> m_channel =
                Channel.CreateUnbounded<EventNotification>(
                    new UnboundedChannelOptions
                    {
                        SingleReader = true,
                        SingleWriter = false
                    });

            private readonly TaskCompletionSource<bool> m_subscribed =
                new(TaskCreationOptions.RunContinuationsAsynchronously);

            private readonly FakeMonitoredItem m_eventMonitoredItem;
            private int m_subscribeCallCount;
            private int m_pumpCancellationObserved;
            private int m_pushed;
            private int m_handled;

            public FakeStreamingSubscription(bool eventMonitoredItemCreated = true)
            {
                m_eventMonitoredItem = new FakeMonitoredItem(eventMonitoredItemCreated);
            }

            public int SubscribeCallCount => m_subscribeCallCount;

            public bool PumpCancellationObserved =>
                Volatile.Read(ref m_pumpCancellationObserved) != 0;

            public void CreateEventMonitoredItem()
            {
                m_eventMonitoredItem.Create();
            }

            public void Push(EventNotification notification)
            {
                Interlocked.Increment(ref m_pushed);
                m_channel.Writer.TryWrite(notification);
            }

            public void Complete()
            {
                m_channel.Writer.TryComplete();
            }

            public Task<bool> WaitForSubscribeAsync()
            {
                return m_subscribed.Task;
            }

            /// <summary>
            /// Spins briefly until every <see cref="Push"/>ed notification
            /// has been observed by the production-code consumer. The
            /// counter is incremented in the iterator <i>after</i> the
            /// <c>yield return</c>, i.e. once the consumer's
            /// <c>HandleNotification</c> call has returned and the
            /// foreach loop has requested the next item.
            /// </summary>
            public async Task QuiesceAsync()
            {
                for (int i = 0; i < 400; i++)
                {
                    if (Volatile.Read(ref m_handled) >= Volatile.Read(ref m_pushed))
                    {
                        return;
                    }
                    await Task.Delay(5).ConfigureAwait(false);
                }
            }

            public IAsyncEnumerable<EventNotification> SubscribeEventsAsync(
                NodeId notifierId,
                EventFilter filter,
                MonitoringOptions? options,
                Func<IMonitoredItem, CancellationToken, ValueTask> onMonitoredItemReady,
                CancellationToken ct = default)
            {
                return SubscribeEventsCoreAsync(onMonitoredItemReady, ct);
            }

            public async IAsyncEnumerable<EventNotification> SubscribeEventsAsync(
                NodeId notifierId,
                EventFilter filter,
                MonitoringOptions? options = null,
                [EnumeratorCancellation] CancellationToken ct = default)
            {
                await foreach (EventNotification notification in SubscribeEventsCoreAsync(null, ct)
                    .ConfigureAwait(false))
                {
                    yield return notification;
                }
            }

            private async IAsyncEnumerable<EventNotification> SubscribeEventsCoreAsync(
                Func<IMonitoredItem, CancellationToken, ValueTask>? onMonitoredItemReady,
                [EnumeratorCancellation] CancellationToken ct = default)
            {
                Interlocked.Increment(ref m_subscribeCallCount);
                m_subscribed.TrySetResult(true);
                if (onMonitoredItemReady != null)
                {
                    await onMonitoredItemReady(m_eventMonitoredItem, ct).ConfigureAwait(false);
                }

                try
                {
                    await foreach (EventNotification n in m_channel.Reader
                        .ReadAllAsync(ct).ConfigureAwait(false))
                    {
                        yield return n;
                        // Reached only after the consumer's
                        // HandleNotification returned and asked for the
                        // next item, so we can declare this push handled.
                        Interlocked.Increment(ref m_handled);
                    }
                }
                finally
                {
                    if (ct.IsCancellationRequested)
                    {
                        Interlocked.Exchange(ref m_pumpCancellationObserved, 1);
                    }
                }
            }

            public async IAsyncEnumerable<DataValueChange> SubscribeDataChangesAsync(
                NodeId nodeId,
                MonitoringOptions? options = null,
                [EnumeratorCancellation] CancellationToken ct = default)
            {
                await Task.Yield();
                yield break;
            }

            public async IAsyncEnumerable<DataValueChange> SubscribeDataChangesAsync(
                IReadOnlyList<NodeId> nodeIds,
                MonitoringOptions? options = null,
                [EnumeratorCancellation] CancellationToken ct = default)
            {
                await Task.Yield();
                yield break;
            }

            public ValueTask DisposeAsync()
            {
                Complete();
                return default;
            }
        }

        private sealed class FakeMonitoredItem : IMonitoredItem, IMonitoredItemApplyState
        {
            private int m_created;

            public FakeMonitoredItem(bool created)
            {
                m_created = created ? 1 : 0;
                Error = ServiceResult.Good;
            }

            public string Name => "fake_event";

            public uint Order => 0;

            public uint ServerId => Created ? 1u : 0u;

            public bool Created => Volatile.Read(ref m_created) != 0;

            public ServiceResult Error { get; }

            public bool HasPendingChanges => !Created && ServiceResult.IsGood(Error);

            public MonitoringFilterResult? FilterResult => null;

            public MonitoringMode CurrentMonitoringMode => MonitoringMode.Reporting;

            public TimeSpan CurrentSamplingInterval => TimeSpan.Zero;

            public uint CurrentQueueSize => 0;

            public uint ClientHandle => 1;

            public IEnumerable<IMonitoredItem> TriggeringItems => Array.Empty<IMonitoredItem>();

            public IEnumerable<IMonitoredItem> TriggeredItems => Array.Empty<IMonitoredItem>();

            public void Create()
            {
                Interlocked.Exchange(ref m_created, 1);
            }

            public ValueTask ConditionRefreshAsync(CancellationToken ct = default)
            {
                return default;
            }
        }
    }
}