mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-20 08:41:51 +08:00
Marc Schier
5b9bcedc62
# Description Adds first-class support for the OPC UA ISA-95 companion specifications: - OPC-10030 Common Model. - OPC-10031-4 Job Control V1. - OPC-10031-4 Job Control V2. The change introduces `Opc.Ua.ISA95`, `Opc.Ua.ISA95.Client`, and `Opc.Ua.ISA95.Server`, together with a NativeAOT-compatible minimal server sample and comprehensive tests. ## Highlights - Compiles the three canonical NodeSets and identifier mapping CSV files through the source generator. - Adds NodeSet identifier mapping validation, typed method arguments/results, companion event records, and generated method argument nodes. - Provides typed Common Model builders with normative relationships and an injectable `GeoSpatialLocationType` provider. - Implements lossless V1/V2 Job Control providers over a shared version-neutral store, including state paths, status events, comments, response queries, catalog changes, and cohesive DI registration. - Adds subtype-aware typed clients, standalone encodeable registration, `IIsa95ClientFactory`, and V2 status-event streaming. - Adds the `MinimalIsa95Server` sample, package documentation, and an ISA-95 guide with conformance and HA guidance. - Tracks typed RSL/GPOS integration as follow-up issue #4095. ## Validation - `Opc.Ua.ISA95.Tests`: 137/137 passed on `net10.0` and `net48`. - Source-generator suites: 81/81 model, 90/90 stack, 3718 passed / 8 skipped core. - `EventRecordFieldReaders` feedback tests: 6/6 on `net10.0` and `net48`. - `EventRecordFieldReaders` changed-method patch coverage: 100% line coverage (36/36) and 100% branch coverage (22/22). - `Opc.Ua.ISA95`, `.Client`, and `.Server` build across all configured target frameworks without new diagnostics. - Cobertura line coverage: 88.08% model, 90.04% client, 91.19% server. - NativeAOT sample publishes and passes a live typed V2 Store/status-event validation. - The ISA-95 NativeAOT smoke test passes in the source-generated native test executable. - OPC UA KB NodeSet validation reports zero errors for Common, V1, and V2. The broader NativeAOT executable still reports unrelated existing GDS and complex-type test failures; the ISA-95-specific native test passes. ## Node state generator fixes Generating method arguments inline for imported nodesets regressed every model that already emits `InputArguments` / `OutputArguments` child node factories: - The inline block ran before the child factories. `MethodState.CreateOrReplaceInputArguments` keeps the property that was assigned first, so the child carrying the well-known NodeId was dropped and a NodeId-less placeholder remained in the address space. This broke browsing (reference descriptions without a target) and made method arguments unreachable at their well-known NodeIds. - The `MethodDeclarationId` fallback keyed on an autogenerated display name redirected every declared method of the standard address space to its own NodeId, which made method calls fail. The inline block is now emitted after the child factories and reduced to a value assignment when a child factory exists. The value is only re-emitted when an argument references a namespace other than OPC UA, because the value of an imported property is the verbatim XML of the source nodeset whose namespace indices do not survive the import. The `MethodDeclarationId` fallback is restricted to declarations without a resolvable NodeId. The generated OPC UA address space is byte-identical to `master` again. ## Related Issues - Fixes #4088 - Follow-up: #4095
1123 lines
47 KiB
C#
1123 lines
47 KiB
C#
/* ========================================================================
|
|
* 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/
|
|
* ======================================================================*/
|
|
|
|
#nullable enable
|
|
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Options;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client.Subscriptions;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
using Opc.Ua.Client.Subscriptions.Streaming;
|
|
|
|
namespace Opc.Ua.Subscriptions.Tests
|
|
{
|
|
/// <summary>
|
|
/// Deterministic offline unit-test coverage for
|
|
/// <see cref="StreamingSubscription"/> and
|
|
/// <see cref="StreamingSubscriptionExtensions"/>. Uses hand-rolled
|
|
/// stub implementations of the subscription interfaces (no Moq) and
|
|
/// drives notifications by invoking the captured notification
|
|
/// handler, so every assertion is data-driven and never relies on
|
|
/// wall-clock timing.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("Streaming")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
public sealed class StreamingSubscriptionCoverageTests
|
|
{
|
|
private static readonly TimeSpan s_safetyTimeout = TimeSpan.FromSeconds(10);
|
|
|
|
private static readonly DateTime s_publishTime =
|
|
new(2024, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
|
|
private static readonly IReadOnlyList<string> s_emptyStringTable =
|
|
[];
|
|
|
|
private static readonly NodeId s_nodeA = new("SensorA", 2);
|
|
private static readonly NodeId s_nodeB = new("SensorB", 2);
|
|
private static readonly NodeId s_notifier = new("Area1", 3);
|
|
private static readonly int[] s_oneTwo = [1, 2];
|
|
private static readonly int[] s_oneTwoThree = [1, 2, 3];
|
|
|
|
private const uint GoodStatus = 0x00000000u;
|
|
private const uint UncertainStatus = 0x40920000u;
|
|
|
|
[Test]
|
|
public void ConstructorWithNullManagerThrowsArgumentNullException()
|
|
{
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = new StreamingSubscription(null!));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("subscriptionManager"));
|
|
}
|
|
|
|
[Test]
|
|
public void SubscribeDataChangesWithNullNodeIdThrowsArgumentNullException()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = subscription.SubscribeDataChangesAsync(NodeId.Null));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("nodeId"));
|
|
}
|
|
|
|
[Test]
|
|
public void SubscribeDataChangesWithNullNodeListThrowsArgumentNullException()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = subscription.SubscribeDataChangesAsync(null!));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("nodeIds"));
|
|
}
|
|
|
|
[Test]
|
|
public void SubscribeEventsWithNullNotifierThrowsArgumentNullException()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = subscription.SubscribeEventsAsync(NodeId.Null, new EventFilter()));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("notifierId"));
|
|
}
|
|
|
|
[Test]
|
|
public void SubscribeEventsWithNullFilterThrowsArgumentNullException()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = subscription.SubscribeEventsAsync(s_notifier, null!));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("filter"));
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscribeCreatesSingleSharedSubscriptionAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> firstEnumerator, Task<bool> firstMove) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
(IAsyncEnumerator<DataValueChange> secondEnumerator, Task<bool> secondMove) =
|
|
StartDataChanges(subscription, s_nodeB, CancellationToken.None);
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
|
|
Assert.That(collection.Count, Is.EqualTo(2u));
|
|
Assert.That(collection.Added[0].ClientHandle,
|
|
Is.Not.EqualTo(collection.Added[1].ClientHandle));
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(await WithinTimeoutAsync(firstMove).ConfigureAwait(false), Is.False);
|
|
Assert.That(await WithinTimeoutAsync(secondMove).ConfigureAwait(false), Is.False);
|
|
await firstEnumerator.DisposeAsync().ConfigureAwait(false);
|
|
await secondEnumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DataChangeNotificationFlowsToSubscriberAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
Assert.That(item.Options.StartNodeId, Is.EqualTo(s_nodeA));
|
|
Assert.That(item.Options.AttributeId, Is.EqualTo(Attributes.Value));
|
|
Assert.That(item.Name, Is.EqualTo($"stream_data_1_{s_nodeA}"));
|
|
|
|
var change = new DataValueChange(item,
|
|
new DataValue(Variant.From(42), new StatusCode(UncertainStatus)), null);
|
|
await FireDataChangeAsync(manager, change).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
|
|
DataValueChange received = enumerator.Current;
|
|
Assert.That(received.Value.WrappedValue, Is.EqualTo(Variant.From(42)));
|
|
Assert.That(received.Value.StatusCode.Code, Is.EqualTo(UncertainStatus));
|
|
Assert.That(received.MonitoredItem, Is.SameAs(item));
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(manager.Subscription!.Collection.RemovedHandles,
|
|
Does.Contain(item.ClientHandle));
|
|
}
|
|
|
|
[Test]
|
|
public async Task DataChangeListOverloadAddsAllItemsAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
var nodes = new List<NodeId> { s_nodeA, s_nodeB };
|
|
IAsyncEnumerator<DataValueChange> enumerator =
|
|
subscription.SubscribeDataChangesAsync(nodes).GetAsyncEnumerator();
|
|
Task<bool> move = enumerator.MoveNextAsync().AsTask();
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
|
|
Assert.That(collection.Count, Is.EqualTo(2u));
|
|
StubMonitoredItem second = collection.Added[1];
|
|
Assert.That(second.Options.StartNodeId, Is.EqualTo(s_nodeB));
|
|
|
|
var change = new DataValueChange(second,
|
|
new DataValue(Variant.From(99), new StatusCode(GoodStatus)), null);
|
|
await FireDataChangeAsync(manager, change).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
|
|
Assert.That(enumerator.Current.Value.WrappedValue, Is.EqualTo(Variant.From(99)));
|
|
Assert.That(enumerator.Current.MonitoredItem, Is.SameAs(second));
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EventNotificationFlowsToSubscriberAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
var filter = new EventFilter();
|
|
IAsyncEnumerator<EventNotification> enumerator =
|
|
subscription.SubscribeEventsAsync(s_notifier, filter).GetAsyncEnumerator();
|
|
Task<bool> move = enumerator.MoveNextAsync().AsTask();
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
Assert.That(item.Options.AttributeId, Is.EqualTo(Attributes.EventNotifier));
|
|
Assert.That(item.Options.Filter, Is.SameAs(filter));
|
|
Assert.That(item.Options.QueueSize, Is.EqualTo(10u));
|
|
Assert.That(item.Name, Is.EqualTo($"stream_event_1_{s_notifier}"));
|
|
|
|
var notification = new EventNotification(item,
|
|
ArrayOf.Wrapped(Variant.From("AlarmActive"), Variant.From(7)));
|
|
await FireEventAsync(manager, notification).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
|
|
EventNotification received = enumerator.Current;
|
|
Assert.That(received.Fields.Count, Is.EqualTo(2));
|
|
Assert.That(received.Fields[0], Is.EqualTo(Variant.From("AlarmActive")));
|
|
Assert.That(received.Fields[1], Is.EqualTo(Variant.From(7)));
|
|
Assert.That(received.MonitoredItem, Is.SameAs(item));
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EventUsesProvidedQueueSizeAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
var options = new MonitoredItemOptions { QueueSize = 5 };
|
|
IAsyncEnumerator<EventNotification> enumerator = subscription
|
|
.SubscribeEventsAsync(s_notifier, new EventFilter(), options)
|
|
.GetAsyncEnumerator();
|
|
Task<bool> move = enumerator.MoveNextAsync().AsTask();
|
|
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
Assert.That(item.Options.QueueSize, Is.EqualTo(5u));
|
|
|
|
await DrainAsync(subscription, enumerator, move).ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConcurrentSubscribersReceiveOnlyMatchingNotificationsAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumeratorA, Task<bool> moveA) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
(IAsyncEnumerator<DataValueChange> enumeratorB, Task<bool> moveB) =
|
|
StartDataChanges(subscription, s_nodeB, CancellationToken.None);
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItem itemA = manager.Subscription!.Collection.Added[0];
|
|
|
|
var change = new DataValueChange(itemA,
|
|
new DataValue(Variant.From(11), new StatusCode(GoodStatus)), null);
|
|
await FireDataChangeAsync(manager, change).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(moveA).ConfigureAwait(false), Is.True);
|
|
Assert.That(enumeratorA.Current.Value.WrappedValue, Is.EqualTo(Variant.From(11)));
|
|
Assert.That(moveB.IsCompleted, Is.False);
|
|
|
|
await enumeratorA.DisposeAsync().ConfigureAwait(false);
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(await WithinTimeoutAsync(moveB).ConfigureAwait(false), Is.False);
|
|
await enumeratorB.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task CancellingOneSubscriberRemovesOnlyItsMonitoredItemAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
using var cancelA = new CancellationTokenSource();
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumeratorA, Task<bool> moveA) =
|
|
StartDataChanges(subscription, s_nodeA, cancelA.Token);
|
|
(IAsyncEnumerator<DataValueChange> enumeratorB, Task<bool> moveB) =
|
|
StartDataChanges(subscription, s_nodeB, CancellationToken.None);
|
|
|
|
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
|
|
StubMonitoredItem itemA = collection.Added[0];
|
|
StubMonitoredItem itemB = collection.Added[1];
|
|
|
|
cancelA.Cancel();
|
|
await AwaitFaultAsync<OperationCanceledException>(moveA).ConfigureAwait(false);
|
|
|
|
Assert.That(collection.RemovedHandles, Does.Contain(itemA.ClientHandle));
|
|
Assert.That(collection.RemovedHandles, Does.Not.Contain(itemB.ClientHandle));
|
|
Assert.That(collection.ContainsHandle(itemB.ClientHandle), Is.True);
|
|
|
|
await enumeratorA.DisposeAsync().ConfigureAwait(false);
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(await WithinTimeoutAsync(moveB).ConfigureAwait(false), Is.False);
|
|
await enumeratorB.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncDisposesUnderlyingSubscriptionAndIsIdempotentAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(manager.Subscription!.DisposeCount, Is.EqualTo(1));
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(manager.Subscription!.DisposeCount, Is.EqualTo(1));
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.False);
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscribeAfterDisposeThrowsObjectDisposedExceptionAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
var subscription = new StreamingSubscription(manager);
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
|
|
await AwaitFaultAsync<ObjectDisposedException>(move).ConfigureAwait(false);
|
|
Assert.That(manager.AddCallCount, Is.Zero);
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscriptionCreateFailureSurfacesToCallerAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager(failAdd: true);
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
|
|
InvalidOperationException ex =
|
|
await AwaitFaultAsync<InvalidOperationException>(move).ConfigureAwait(false);
|
|
Assert.That(ex.Message, Is.EqualTo("subscription create failed"));
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task FailedMonitoredItemAddSurfacesToCallerAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager(failItemAdd: true);
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
|
|
InvalidOperationException ex =
|
|
await AwaitFaultAsync<InvalidOperationException>(move).ConfigureAwait(false);
|
|
Assert.That(ex.Message, Does.Contain(s_nodeA.ToString()));
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
|
|
Assert.That(collection.Count, Is.Zero);
|
|
Assert.That(collection.Added, Is.Empty);
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DataChangeWithNullMonitoredItemIsIgnoredAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
|
|
var ignored = new DataValueChange(null,
|
|
new DataValue(Variant.From(1), new StatusCode(GoodStatus)), null);
|
|
await FireDataChangeAsync(manager, ignored).ConfigureAwait(false);
|
|
Assert.That(move.IsCompleted, Is.False);
|
|
|
|
var delivered = new DataValueChange(item,
|
|
new DataValue(Variant.From(55), new StatusCode(GoodStatus)), null);
|
|
await FireDataChangeAsync(manager, delivered).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
|
|
Assert.That(enumerator.Current.Value.WrappedValue, Is.EqualTo(Variant.From(55)));
|
|
Assert.That(enumerator.Current.MonitoredItem, Is.SameAs(item));
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EventWithNullMonitoredItemIsIgnoredAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
IAsyncEnumerator<EventNotification> enumerator =
|
|
subscription.SubscribeEventsAsync(s_notifier, new EventFilter()).GetAsyncEnumerator();
|
|
Task<bool> move = enumerator.MoveNextAsync().AsTask();
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
|
|
var ignored = new EventNotification(null, ArrayOf.Wrapped(Variant.From(1)));
|
|
await FireEventAsync(manager, ignored).ConfigureAwait(false);
|
|
Assert.That(move.IsCompleted, Is.False);
|
|
|
|
var delivered = new EventNotification(item, ArrayOf.Wrapped(Variant.From("Ping")));
|
|
await FireEventAsync(manager, delivered).ConfigureAwait(false);
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
|
|
Assert.That(enumerator.Current.Fields.Count, Is.EqualTo(1));
|
|
Assert.That(enumerator.Current.Fields[0], Is.EqualTo(Variant.From("Ping")));
|
|
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task KeepAliveAndStateChangeNotificationsAreNoOpsAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
|
|
ValueTask keepAlive = manager.Handler!.OnKeepAliveNotificationAsync(
|
|
manager.Subscription!, 7u, s_publishTime, PublishState.None);
|
|
Assert.That(keepAlive.IsCompleted, Is.True);
|
|
await keepAlive.ConfigureAwait(false);
|
|
|
|
ValueTask stateChanged = manager.Handler!.OnSubscriptionStateChangedAsync(
|
|
manager.Subscription!, SubscriptionState.Created, PublishState.None);
|
|
Assert.That(stateChanged.IsCompleted, Is.True);
|
|
await stateChanged.ConfigureAwait(false);
|
|
|
|
Assert.That(move.IsCompleted, Is.False);
|
|
await DrainAsync(subscription, enumerator, move).ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscriptionOptionsMonitorIsQueriedOnCreateAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager();
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
Assert.That(manager.CapturedOptions, Is.Not.Null);
|
|
Assert.That(manager.CapturedOnChange, Is.Null);
|
|
|
|
await DrainAsync(subscription, enumerator, move).ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncSwallowsUnderlyingDisposeFailureAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager(failDispose: true);
|
|
var subscription = new StreamingSubscription(manager);
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, CancellationToken.None);
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(manager.Subscription!.DisposeCount, Is.EqualTo(1));
|
|
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.False);
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DataSubscriberFinallySwallowsRemoveFailureAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager(failRemove: true);
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
using var cancel = new CancellationTokenSource();
|
|
|
|
(IAsyncEnumerator<DataValueChange> enumerator, Task<bool> move) =
|
|
StartDataChanges(subscription, s_nodeA, cancel.Token);
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
|
|
cancel.Cancel();
|
|
await AwaitFaultAsync<OperationCanceledException>(move).ConfigureAwait(false);
|
|
|
|
Assert.That(manager.Subscription!.Collection.RemoveAttempts,
|
|
Does.Contain(item.ClientHandle));
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EventSubscriberFinallySwallowsRemoveFailureAsync()
|
|
{
|
|
var manager = new StubSubscriptionManager(failRemove: true);
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
using var cancel = new CancellationTokenSource();
|
|
|
|
IAsyncEnumerator<EventNotification> enumerator = subscription
|
|
.SubscribeEventsAsync(s_notifier, new EventFilter(), ct: cancel.Token)
|
|
.GetAsyncEnumerator(cancel.Token);
|
|
Task<bool> move = enumerator.MoveNextAsync().AsTask();
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
|
|
|
|
cancel.Cancel();
|
|
await AwaitFaultAsync<OperationCanceledException>(move).ConfigureAwait(false);
|
|
|
|
Assert.That(manager.Subscription!.Collection.RemoveAttempts,
|
|
Does.Contain(item.ClientHandle));
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConcurrentCreatorsShareSingleSubscriptionViaInitLockAsync()
|
|
{
|
|
using var addEntered = new SemaphoreSlim(0, 1);
|
|
using var itemsAdded = new SemaphoreSlim(0, 2);
|
|
using var releaseAdd = new ManualResetEventSlim(false);
|
|
var manager = new StubSubscriptionManager
|
|
{
|
|
OnAddEntered = () => addEntered.Release(),
|
|
AddGate = () => _ = releaseAdd.Wait(s_safetyTimeout),
|
|
OnItemAdded = () => itemsAdded.Release()
|
|
};
|
|
await using var subscription = new StreamingSubscription(manager);
|
|
|
|
// First creator acquires the init lock and blocks inside Add so the
|
|
// shared subscription is still null while a second creator queues.
|
|
IAsyncEnumerator<DataValueChange> firstEnumerator =
|
|
subscription.SubscribeDataChangesAsync(s_nodeA).GetAsyncEnumerator();
|
|
Task<bool> firstMove = Task.Run(() => firstEnumerator.MoveNextAsync().AsTask());
|
|
Assert.That(await addEntered.WaitAsync(s_safetyTimeout).ConfigureAwait(false), Is.True);
|
|
|
|
// Second creator passes the pre-lock null check then suspends on the
|
|
// held init lock; releasing the gate forces it through the
|
|
// double-check path where the subscription now already exists.
|
|
IAsyncEnumerator<DataValueChange> secondEnumerator =
|
|
subscription.SubscribeDataChangesAsync(s_nodeB).GetAsyncEnumerator();
|
|
Task<bool> secondMove = secondEnumerator.MoveNextAsync().AsTask();
|
|
|
|
releaseAdd.Set();
|
|
Assert.That(await itemsAdded.WaitAsync(s_safetyTimeout).ConfigureAwait(false), Is.True);
|
|
Assert.That(await itemsAdded.WaitAsync(s_safetyTimeout).ConfigureAwait(false), Is.True);
|
|
|
|
Assert.That(manager.AddCallCount, Is.EqualTo(1));
|
|
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
|
|
Assert.That(collection.Count, Is.EqualTo(2u));
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(await WithinTimeoutAsync(firstMove).ConfigureAwait(false), Is.False);
|
|
Assert.That(await WithinTimeoutAsync(secondMove).ConfigureAwait(false), Is.False);
|
|
await firstEnumerator.DisposeAsync().ConfigureAwait(false);
|
|
await secondEnumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task TakeUntilAsyncIsInclusiveOfMatchAsync()
|
|
{
|
|
var collected = new List<int>();
|
|
await foreach (int value in RangeAsync(1, 2, 3, 4).TakeUntilAsync(static v => v == 3))
|
|
{
|
|
collected.Add(value);
|
|
}
|
|
Assert.That(collected, Is.EqualTo(s_oneTwoThree));
|
|
}
|
|
|
|
[Test]
|
|
public void TakeUntilAsyncWithNullSourceThrowsArgumentNullException()
|
|
{
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = StreamingSubscriptionExtensions.TakeUntilAsync<int>(
|
|
null!, static v => true));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("source"));
|
|
}
|
|
|
|
[Test]
|
|
public void TakeUntilAsyncWithNullPredicateThrowsArgumentNullException()
|
|
{
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = RangeAsync(1).TakeUntilAsync(null!));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("predicate"));
|
|
}
|
|
|
|
[Test]
|
|
public async Task TakeAsyncYieldsExactCountAsync()
|
|
{
|
|
var collected = new List<int>();
|
|
await foreach (int value in RangeAsync(1, 2, 3, 4, 5).TakeAsync(2))
|
|
{
|
|
collected.Add(value);
|
|
}
|
|
Assert.That(collected, Is.EqualTo(s_oneTwo));
|
|
}
|
|
|
|
[Test]
|
|
public void TakeAsyncWithNonPositiveCountThrowsArgumentOutOfRangeException()
|
|
{
|
|
ArgumentOutOfRangeException? ex = Assert.Throws<ArgumentOutOfRangeException>(
|
|
() => _ = RangeAsync(1).TakeAsync(0));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("count"));
|
|
}
|
|
|
|
[Test]
|
|
public void TakeAsyncWithNullSourceThrowsArgumentNullException()
|
|
{
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = StreamingSubscriptionExtensions.TakeAsync<int>(null!, 1));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("source"));
|
|
}
|
|
|
|
[Test]
|
|
public async Task BufferedAsyncReturnsRequestedCountAsync()
|
|
{
|
|
IReadOnlyList<int> buffer = await RangeAsync(1, 2, 3, 4).BufferedAsync(3).ConfigureAwait(false);
|
|
Assert.That(buffer, Is.EqualTo(s_oneTwoThree));
|
|
}
|
|
|
|
[Test]
|
|
public async Task BufferedAsyncWithNullSourceThrowsArgumentNullExceptionAsync()
|
|
{
|
|
ArgumentNullException ex = await AwaitFaultAsync<ArgumentNullException>(
|
|
StreamingSubscriptionExtensions.BufferedAsync<int>(null!, 1).AsTask()).ConfigureAwait(false);
|
|
Assert.That(ex.ParamName, Is.EqualTo("source"));
|
|
}
|
|
|
|
[Test]
|
|
public async Task BufferedAsyncWithNonPositiveCountThrowsArgumentOutOfRangeExceptionAsync()
|
|
{
|
|
ArgumentOutOfRangeException ex = await AwaitFaultAsync<ArgumentOutOfRangeException>(
|
|
RangeAsync(1).BufferedAsync(0).AsTask()).ConfigureAwait(false);
|
|
Assert.That(ex.ParamName, Is.EqualTo("count"));
|
|
}
|
|
|
|
[Test]
|
|
public void WithTimeoutAsyncWithNullSourceThrowsArgumentNullException()
|
|
{
|
|
ArgumentNullException? ex = Assert.Throws<ArgumentNullException>(
|
|
() => _ = StreamingSubscriptionExtensions.WithTimeoutAsync<int>(
|
|
null!, TimeSpan.FromSeconds(1)));
|
|
Assert.That(ex!.ParamName, Is.EqualTo("source"));
|
|
}
|
|
|
|
[Test]
|
|
public async Task WithTimeoutAsyncPassesThroughItemsBeforeTimeoutAsync()
|
|
{
|
|
var collected = new List<int>();
|
|
await foreach (int value in RangeAsync(1, 2, 3)
|
|
.WithTimeoutAsync(TimeSpan.FromSeconds(30)))
|
|
{
|
|
collected.Add(value);
|
|
}
|
|
Assert.That(collected, Is.EqualTo(s_oneTwoThree));
|
|
}
|
|
|
|
private static (IAsyncEnumerator<DataValueChange> Enumerator, Task<bool> Move)
|
|
StartDataChanges(StreamingSubscription subscription, NodeId nodeId, CancellationToken ct)
|
|
{
|
|
IAsyncEnumerator<DataValueChange> enumerator =
|
|
subscription.SubscribeDataChangesAsync(nodeId, ct: ct).GetAsyncEnumerator(ct);
|
|
return (enumerator, enumerator.MoveNextAsync().AsTask());
|
|
}
|
|
|
|
private static async Task FireDataChangeAsync(
|
|
StubSubscriptionManager manager, DataValueChange change)
|
|
{
|
|
await manager.Handler!.OnDataChangeNotificationAsync(
|
|
manager.Subscription!, 1u, s_publishTime,
|
|
new ReadOnlyMemory<DataValueChange>([change]),
|
|
PublishState.None, s_emptyStringTable).ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task FireEventAsync(
|
|
StubSubscriptionManager manager, EventNotification notification)
|
|
{
|
|
await manager.Handler!.OnEventDataNotificationAsync(
|
|
manager.Subscription!, 1u, s_publishTime,
|
|
new ReadOnlyMemory<EventNotification>([notification]),
|
|
PublishState.None, s_emptyStringTable).ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task DrainAsync<T>(
|
|
StreamingSubscription subscription, IAsyncEnumerator<T> enumerator, Task<bool> move)
|
|
{
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.False);
|
|
await enumerator.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task<T> WithinTimeoutAsync<T>(Task<T> task)
|
|
{
|
|
using var cts = new CancellationTokenSource();
|
|
var delay = Task.Delay(s_safetyTimeout, cts.Token);
|
|
Task winner = await Task.WhenAny(task, delay).ConfigureAwait(false);
|
|
Assert.That(winner, Is.SameAs(task),
|
|
"operation did not complete within the safety timeout");
|
|
cts.Cancel();
|
|
return await task.ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task<TException> AwaitFaultAsync<TException>(Task task)
|
|
where TException : Exception
|
|
{
|
|
using var cts = new CancellationTokenSource();
|
|
var delay = Task.Delay(s_safetyTimeout, cts.Token);
|
|
Task winner = await Task.WhenAny(task, delay).ConfigureAwait(false);
|
|
Assert.That(winner, Is.SameAs(task),
|
|
"operation did not fault within the safety timeout");
|
|
cts.Cancel();
|
|
|
|
TException? caught = null;
|
|
try
|
|
{
|
|
await task.ConfigureAwait(false);
|
|
}
|
|
catch (TException ex)
|
|
{
|
|
caught = ex;
|
|
}
|
|
Assert.That(caught, Is.Not.Null, "expected exception was not thrown");
|
|
return caught!;
|
|
}
|
|
|
|
private static async IAsyncEnumerable<int> RangeAsync(params int[] values)
|
|
{
|
|
foreach (int value in values)
|
|
{
|
|
await Task.Yield();
|
|
yield return value;
|
|
}
|
|
}
|
|
|
|
private sealed class StubSubscriptionManager : ISubscriptionManager
|
|
{
|
|
private readonly bool m_failAdd;
|
|
private readonly bool m_failItemAdd;
|
|
private readonly bool m_failDispose;
|
|
private readonly bool m_failRemove;
|
|
private StubSubscription? m_subscription;
|
|
|
|
public StubSubscriptionManager(
|
|
bool failAdd = false,
|
|
bool failItemAdd = false,
|
|
bool failDispose = false,
|
|
bool failRemove = false)
|
|
{
|
|
m_failAdd = failAdd;
|
|
m_failItemAdd = failItemAdd;
|
|
m_failDispose = failDispose;
|
|
m_failRemove = failRemove;
|
|
}
|
|
|
|
public int AddCallCount { get; private set; }
|
|
|
|
public ISubscriptionNotificationHandler? Handler { get; private set; }
|
|
|
|
public StubSubscription? Subscription => m_subscription;
|
|
|
|
public SubscriptionOptions? CapturedOptions { get; private set; }
|
|
|
|
public IDisposable? CapturedOnChange { get; private set; }
|
|
|
|
public Action? OnAddEntered { get; init; }
|
|
|
|
public Action? AddGate { get; init; }
|
|
|
|
public Action? OnItemAdded { get; init; }
|
|
|
|
public ISubscription Add(
|
|
ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options)
|
|
{
|
|
AddCallCount++;
|
|
Handler = handler;
|
|
CapturedOptions = options.Get(null);
|
|
CapturedOnChange = options.OnChange(static (_, _) => { });
|
|
if (m_failAdd)
|
|
{
|
|
throw new InvalidOperationException("subscription create failed");
|
|
}
|
|
OnAddEntered?.Invoke();
|
|
AddGate?.Invoke();
|
|
m_subscription ??= new StubSubscription(
|
|
m_failItemAdd, m_failDispose, m_failRemove, OnItemAdded);
|
|
return m_subscription;
|
|
}
|
|
|
|
public DiagnosticsMasks ReturnDiagnostics
|
|
{
|
|
get => throw new NotSupportedException();
|
|
set => throw new NotSupportedException();
|
|
}
|
|
|
|
public int MaxPublishWorkerCount
|
|
{
|
|
get => throw new NotSupportedException();
|
|
set => throw new NotSupportedException();
|
|
}
|
|
|
|
public int MinPublishWorkerCount
|
|
{
|
|
get => throw new NotSupportedException();
|
|
set => throw new NotSupportedException();
|
|
}
|
|
|
|
public bool PoolNotifications
|
|
{
|
|
get => throw new NotSupportedException();
|
|
set => throw new NotSupportedException();
|
|
}
|
|
|
|
public int PublishWorkerCount => throw new NotSupportedException();
|
|
|
|
public int GoodPublishRequestCount => throw new NotSupportedException();
|
|
|
|
public int BadPublishRequestCount => throw new NotSupportedException();
|
|
|
|
public long MissingMessageCount => throw new NotSupportedException();
|
|
|
|
public long RepublishMessageCount => throw new NotSupportedException();
|
|
|
|
public int Count => throw new NotSupportedException();
|
|
|
|
public IEnumerable<ISubscription> Items => throw new NotSupportedException();
|
|
|
|
public ValueTask SaveAsync(
|
|
Stream stream,
|
|
IServiceMessageContext messageContext,
|
|
IEnumerable<ISubscription>? subscriptions = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<IReadOnlyList<ISubscription>> LoadAsync(
|
|
Stream stream,
|
|
IServiceMessageContext messageContext,
|
|
Func<string, ISubscriptionNotificationHandler> handlerFactory,
|
|
bool transferSubscriptions = false,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
}
|
|
|
|
private sealed class StubSubscription : ISubscription
|
|
{
|
|
private readonly bool m_failDispose;
|
|
private int m_disposeCount;
|
|
|
|
public StubSubscription(
|
|
bool failItemAdd, bool failDispose, bool failRemove, Action? onItemAdded)
|
|
{
|
|
m_failDispose = failDispose;
|
|
Collection = new StubMonitoredItemCollection(failItemAdd, failRemove, onItemAdded);
|
|
}
|
|
|
|
public StubMonitoredItemCollection Collection { get; }
|
|
|
|
public IMonitoredItemCollection MonitoredItems => Collection;
|
|
|
|
public int DisposeCount => m_disposeCount;
|
|
|
|
public ValueTask DisposeAsync()
|
|
{
|
|
Interlocked.Increment(ref m_disposeCount);
|
|
if (m_failDispose)
|
|
{
|
|
throw new InvalidOperationException("dispose failed");
|
|
}
|
|
return default;
|
|
}
|
|
|
|
public bool Created => throw new NotSupportedException();
|
|
|
|
public TimeSpan CurrentPublishingInterval => throw new NotSupportedException();
|
|
|
|
public byte CurrentPriority => throw new NotSupportedException();
|
|
|
|
public uint CurrentLifetimeCount => throw new NotSupportedException();
|
|
|
|
public uint CurrentKeepAliveCount => throw new NotSupportedException();
|
|
|
|
public bool CurrentPublishingEnabled => throw new NotSupportedException();
|
|
|
|
public uint CurrentMaxNotificationsPerPublish => throw new NotSupportedException();
|
|
|
|
public long MissingMessageCount => throw new NotSupportedException();
|
|
|
|
public long RepublishMessageCount => throw new NotSupportedException();
|
|
|
|
public ValueTask ConditionRefreshAsync(CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask RecreateAsync(CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<TimeSpan> SetAsDurableAsync(
|
|
TimeSpan lifetime, CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<SetTriggeringResult> SetTriggeringAsync(
|
|
IMonitoredItem triggeringItem,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToAdd = null,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToRemove = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
}
|
|
|
|
private sealed class StubMonitoredItemCollection : IMonitoredItemCollection
|
|
{
|
|
private readonly Lock m_lock = new();
|
|
private readonly Dictionary<uint, StubMonitoredItem> m_items = [];
|
|
private readonly List<StubMonitoredItem> m_added = [];
|
|
private readonly List<uint> m_removed = [];
|
|
private readonly List<uint> m_removeAttempts = [];
|
|
private readonly bool m_failItemAdd;
|
|
private readonly bool m_failRemove;
|
|
private readonly Action? m_onItemAdded;
|
|
private uint m_nextHandle = 1000;
|
|
|
|
public StubMonitoredItemCollection(
|
|
bool failItemAdd, bool failRemove, Action? onItemAdded)
|
|
{
|
|
m_failItemAdd = failItemAdd;
|
|
m_failRemove = failRemove;
|
|
m_onItemAdded = onItemAdded;
|
|
}
|
|
|
|
public uint Count
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return (uint)m_items.Count;
|
|
}
|
|
}
|
|
}
|
|
|
|
public IEnumerable<IMonitoredItem> Items
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return [.. m_items.Values];
|
|
}
|
|
}
|
|
}
|
|
|
|
public StubMonitoredItem[] Added
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return [.. m_added];
|
|
}
|
|
}
|
|
}
|
|
|
|
public IReadOnlyList<uint> RemovedHandles
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return [.. m_removed];
|
|
}
|
|
}
|
|
}
|
|
|
|
public IReadOnlyList<uint> RemoveAttempts
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return [.. m_removeAttempts];
|
|
}
|
|
}
|
|
}
|
|
|
|
public bool ContainsHandle(uint clientHandle)
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return m_items.ContainsKey(clientHandle);
|
|
}
|
|
}
|
|
|
|
public bool TryAdd(
|
|
string name,
|
|
IOptionsMonitor<MonitoredItemOptions> options,
|
|
out IMonitoredItem? monitoredItem)
|
|
{
|
|
if (m_failItemAdd)
|
|
{
|
|
monitoredItem = null;
|
|
return false;
|
|
}
|
|
lock (m_lock)
|
|
{
|
|
var item = new StubMonitoredItem(m_nextHandle++, name, options.CurrentValue);
|
|
m_items[item.ClientHandle] = item;
|
|
m_added.Add(item);
|
|
monitoredItem = item;
|
|
}
|
|
m_onItemAdded?.Invoke();
|
|
return true;
|
|
}
|
|
|
|
public bool TryRemove(uint clientHandle)
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
m_removeAttempts.Add(clientHandle);
|
|
if (m_failRemove)
|
|
{
|
|
throw new InvalidOperationException("remove failed");
|
|
}
|
|
m_removed.Add(clientHandle);
|
|
return m_items.Remove(clientHandle);
|
|
}
|
|
}
|
|
|
|
public bool TryGetMonitoredItemByClientHandle(
|
|
uint clientHandle, out IMonitoredItem? monitoredItem)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public bool TryGetMonitoredItemByName(string name, out IMonitoredItem? monitoredItem)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public IReadOnlyList<IMonitoredItem> Update(
|
|
IReadOnlyList<(string Name,
|
|
IOptionsMonitor<MonitoredItemOptions> Options)> state)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
}
|
|
|
|
private sealed class StubMonitoredItem : IMonitoredItem
|
|
{
|
|
public StubMonitoredItem(uint clientHandle, string name, MonitoredItemOptions options)
|
|
{
|
|
ClientHandle = clientHandle;
|
|
Name = name;
|
|
Options = options;
|
|
}
|
|
|
|
public uint ClientHandle { get; }
|
|
|
|
public string Name { get; }
|
|
|
|
public MonitoredItemOptions Options { get; }
|
|
|
|
public uint Order => throw new NotSupportedException();
|
|
|
|
public uint ServerId => throw new NotSupportedException();
|
|
|
|
public bool Created => throw new NotSupportedException();
|
|
|
|
public ServiceResult Error => throw new NotSupportedException();
|
|
|
|
public MonitoringFilterResult? FilterResult => throw new NotSupportedException();
|
|
|
|
public MonitoringMode CurrentMonitoringMode => throw new NotSupportedException();
|
|
|
|
public TimeSpan CurrentSamplingInterval => throw new NotSupportedException();
|
|
|
|
public uint CurrentQueueSize => throw new NotSupportedException();
|
|
|
|
public IEnumerable<IMonitoredItem> TriggeringItems => throw new NotSupportedException();
|
|
|
|
public IEnumerable<IMonitoredItem> TriggeredItems => throw new NotSupportedException();
|
|
|
|
public ValueTask ConditionRefreshAsync(CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
}
|
|
}
|
|
}
|