GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-18 14:43:08 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
UA-.NETStandard/tests/Opc.Ua.Subscriptions.Tests/StreamingSubscriptionCoverageTests.cs
T
romanett 9b034f8291 Enable nullable reference types centrally; migrate all unset and annotations-only projects (#4700) 
# Description

Nullable reference types are now enabled once, in `common.props`, and
every project in the repository builds warning-clean with them.

**Configuration**
- `<Nullable>enable</Nullable>` lives in `common.props`. All per-project
`<Nullable>` settings are removed: 178 project files (enable,
annotations, and the `disable` in the three Aot test projects).
- 419 per-file `#nullable enable`/`restore` directives and the 4
`#nullable disable` directives are removed.
- Kept on purpose:
- The 5 `<auto-generated>` schema files (`UANodeSet.cs`, `UA Type
Dictionary.cs`, `OPCBinarySchema.cs`, `SecuredApplication.cs`,
`PInvokeHelper.cs`) keep `#nullable enable`. The compiler ignores the
project setting in generated code, so the directive is required there.
- `tests/Opc.Ua.Interop.LegacyPeer` keeps `disable`. It builds against
the 1.5.378 packages and does not import the repository props.
- `docs/DeveloperGuide.md` and `.github/copilot-instructions.md` now
describe the central setting.

**Migration** (the 43 unset and 13 annotations-only projects, roughly
10k diagnostics)
- **Production code: annotated at the declarations.** Covers
`tools/Opc.Ua.SourceGeneration.Core`, `src/Opc.Ua.Lds.Server` and the
LDS sample.
- Members, parameters and returns are nullable where null actually
flows.
- Null-check helpers carry `[NotNullWhen]` (`IsNull`, `IsValidXmlName`,
`IsMethodTypeDesign`).
- `CoreUtils.Format`, `Utils.Format`, `ITemplateWriter.WriteLine` take
`params object?[]`. `Template.AddReplacement` takes `string?` (rendering
already skips null replacements).
- `IRegisteredServerStore.RegisterAsync` returns
`Task<RegistrationEntry?>`: it returns null when a registration is
removed.
- Dereferences backed by an invariant (for example, the validated model)
use `!`, so runtime behaviour is unchanged.
- **Behavioural changes in production code** (small, deliberate):
- `NodeSetToModelDesign` throws `InvalidOperationException` when the
file is not a NodeSet. Before, it threw a NullReferenceException on the
next line.
- `ModelDesignValidator` starts its lookup tables empty instead of null.
A redundant `if (m_nodes == null) return;` guard is gone.
- **Test code: minimal annotations.**
  - `?` on values that can be null, `!` where the test asserts non-null.
  - `= null!` for fixture members assigned in SetUp.
- `TryGetStructure<T>(out T? x)` where the inferred `T?` broke the
`IEncodeable` constraint.
- A few redundant `== null` checks on structs (`NodeId`, `ArrayOf<T>`)
folded into the adjacent `IsNull`/`Count` checks.

**Notes for reviewers**
- The diff is large (~1,500 files) but almost entirely mechanical: `?`,
`!` and directive removal. The non-annotation changes are listed above.
- Known follow-up, not done here:
`IModelDesign.Namespaces/TargetNamespace/NamespaceUris` stay
non-nullable even though the validator returns null before it loads a
dictionary. Making them nullable would ripple through every generator
call site.

## Related Issues

- Fixes #4439

## Checklist

- [ ] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [ ] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [x] I have added all necessary documentation.
- [x] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [ ] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.

The "no new warnings" box covers net10.0 and net48: full non-incremental
builds of `UA.slnx`, `tools/SourceGeneration.slnx`,
`tools/MigrationAnalyzer.slnx` and the Aot test projects finished with 0
warnings and 0 errors under the repository's warnings-as-errors
settings. net9.0 and net8.0 also had no nullable diagnostics, but they
were not rebuilt after the final fixes. Tests have not been run locally
yet.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-08 08:48:13 +02:00

1376 lines
58 KiB
C#
Raw Permalink Blame History

/* ========================================================================
* 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.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);
}
[TestCase(true, 2, 3)]
[TestCase(false, 1, 2)]
public async Task BoundedEventChannelHonorsDiscardOldestOptionAsync(
bool discardOldest,
int firstExpected,
int secondExpected)
{
var manager = new StubSubscriptionManager();
await using var subscription = new StreamingSubscription(manager);
var options = new MonitoredItemOptions
{
QueueSize = 2,
DiscardOldest = discardOldest
};
var itemReady = new TaskCompletionSource<bool>(
TaskCreationOptions.RunContinuationsAsynchronously);
var releaseItemReady = new TaskCompletionSource<bool>(
TaskCreationOptions.RunContinuationsAsynchronously);
var readiness = (IStreamingSubscriptionReadiness)subscription;
await using IAsyncEnumerator<EventNotification> enumerator = readiness
.SubscribeEventsAsync(
s_notifier,
new EventFilter(),
options,
async (_, _) =>
{
itemReady.SetResult(true);
await releaseItemReady.Task.ConfigureAwait(false);
})
.GetAsyncEnumerator();
Task<bool> move = enumerator.MoveNextAsync().AsTask();
try
{
await itemReady.Task.WaitAsync(s_safetyTimeout).ConfigureAwait(false);
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
await FireEventsAsync(
manager,
new EventNotification(item, ArrayOf.Wrapped(Variant.From(1))),
new EventNotification(item, ArrayOf.Wrapped(Variant.From(2))),
new EventNotification(item, ArrayOf.Wrapped(Variant.From(3))))
.ConfigureAwait(false);
releaseItemReady.SetResult(true);
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
Assert.That(
enumerator.Current.Fields[0],
Is.EqualTo(Variant.From(firstExpected)));
move = enumerator.MoveNextAsync().AsTask();
Assert.That(await WithinTimeoutAsync(move).ConfigureAwait(false), Is.True);
Assert.That(
enumerator.Current.Fields[0],
Is.EqualTo(Variant.From(secondExpected)));
Assert.That(subscription.DroppedNotificationCount, Is.EqualTo(1));
}
finally
{
releaseItemReady.TrySetResult(true);
await subscription.DisposeAsync().ConfigureAwait(false);
await WithinTimeoutAsync(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);
}
[TestCase(true, 4)]
[TestCase(false, 1)]
public async Task DefaultDataChangeBufferBoundsSlowReadersAsync(bool discardOldest, int expected)
{
var manager = new StubSubscriptionManager();
await using var subscription = new StreamingSubscription(manager);
MonitoredItemOptions? options = discardOldest
? null
: new MonitoredItemOptions { DiscardOldest = false };
await using IAsyncEnumerator<DataValueChange> enumerator = subscription
.SubscribeDataChangesAsync(s_nodeA, options).GetAsyncEnumerator();
Task<bool> firstMove = enumerator.MoveNextAsync().AsTask();
StubMonitoredItem item = manager.Subscription!.Collection.Added[0];
await FireDataChangeAsync(manager, new DataValueChange(
item, new DataValue(Variant.From(0)), null)).ConfigureAwait(false);
Assert.That(await WithinTimeoutAsync(firstMove).ConfigureAwait(false), Is.True);
Assert.That(enumerator.Current.Value.WrappedValue, Is.EqualTo(Variant.From(0)));
Assert.That(subscription.DroppedNotificationCount, Is.Zero);
for (int value = 1; value <= 4; value++)
{
await FireDataChangeAsync(manager, new DataValueChange(
item, new DataValue(Variant.From(value)), null)).ConfigureAwait(false);
}
await subscription.DisposeAsync().ConfigureAwait(false);
Assert.That(await enumerator.MoveNextAsync().ConfigureAwait(false), Is.True);
Assert.That(enumerator.Current.Value.WrappedValue, Is.EqualTo(Variant.From(expected)));
Assert.That(await enumerator.MoveNextAsync().ConfigureAwait(false), Is.False);
Assert.That(subscription.DroppedNotificationCount, Is.EqualTo(3));
}
[TestCase(0u, true, new[] { 5, 6 })]
[TestCase(2u, true, new[] { 3, 4, 5, 6 })]
[TestCase(2u, false, new[] { 1, 2, 3, 4 })]
public async Task MultiNodeDataChangeBufferHonorsBoundAndDiscardPolicyAsync(
uint queueSize,
bool discardOldest,
int[] expected)
{
var manager = new StubSubscriptionManager();
await using var subscription = new StreamingSubscription(manager);
await using IAsyncEnumerator<DataValueChange> enumerator = subscription
.SubscribeDataChangesAsync(
[s_nodeA, s_nodeB],
new MonitoredItemOptions { QueueSize = queueSize, DiscardOldest = discardOldest })
.GetAsyncEnumerator();
Task<bool> firstMove = enumerator.MoveNextAsync().AsTask();
StubMonitoredItemCollection collection = manager.Subscription!.Collection;
Assert.That(collection.Count, Is.EqualTo(2));
await FireDataChangeAsync(manager, new DataValueChange(
collection.Added[0], new DataValue(Variant.From(0)), null)).ConfigureAwait(false);
Assert.That(await WithinTimeoutAsync(firstMove).ConfigureAwait(false), Is.True);
for (int value = 1; value <= 6; value++)
{
await FireDataChangeAsync(manager, new DataValueChange(
collection.Added[value == 1 ? 1 : 0], new DataValue(Variant.From(value)), null))
.ConfigureAwait(false);
}
await subscription.DisposeAsync().ConfigureAwait(false);
var values = new List<int>();
while (await enumerator.MoveNextAsync().ConfigureAwait(false))
{
Assert.That(enumerator.Current.Value.WrappedValue.TryGetValue(out int value), Is.True);
Assert.That(enumerator.Current.MonitoredItem,
Is.SameAs(collection.Added[value == 1 ? 1 : 0]));
values.Add(value);
}
Assert.That(values, Is.EqualTo(expected));
Assert.That(subscription.DroppedNotificationCount, Is.EqualTo(6 - expected.Length));
Assert.That(manager.Subscription.DisposeCount, Is.EqualTo(1));
}
[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 async Task TakeUntilAsyncForwardsCancellationToSourceEnumeratorAsync()
{
using var cts = new CancellationTokenSource();
var probe = new CancellationProbeEnumerable();
Task consumption = ConsumeAsync(
probe.TakeUntilAsync(static _ => false, cts.Token));
Assert.That(probe.EnumeratorCreated.Wait(s_safetyTimeout), Is.True);
cts.Cancel();
await AwaitFaultAsync<OperationCanceledException>(consumption)
.ConfigureAwait(false);
Assert.That(probe.EnumeratorCancellationObserved, Is.True);
}
[Test]
public async Task TakeAsyncForwardsCancellationToSourceEnumeratorAsync()
{
using var cts = new CancellationTokenSource();
var probe = new CancellationProbeEnumerable();
Task consumption = ConsumeAsync(probe.TakeAsync(1, cts.Token));
Assert.That(probe.EnumeratorCreated.Wait(s_safetyTimeout), Is.True);
cts.Cancel();
await AwaitFaultAsync<OperationCanceledException>(consumption)
.ConfigureAwait(false);
Assert.That(probe.EnumeratorCancellationObserved, Is.True);
}
[Test]
public async Task BufferedAsyncForwardsCancellationToSourceEnumeratorAsync()
{
using var cts = new CancellationTokenSource();
var probe = new CancellationProbeEnumerable();
Task consumption = ConsumeBufferedAsync(probe, cts.Token);
Assert.That(probe.EnumeratorCreated.Wait(s_safetyTimeout), Is.True);
cts.Cancel();
await AwaitFaultAsync<OperationCanceledException>(consumption)
.ConfigureAwait(false);
Assert.That(probe.EnumeratorCancellationObserved, Is.True);
}
[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 FireEventsAsync(
StubSubscriptionManager manager,
params EventNotification[] notifications)
{
await manager.Handler!.OnEventDataNotificationAsync(
manager.Subscription!, 1u, s_publishTime,
new ReadOnlyMemory<EventNotification>(notifications),
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 Task ConsumeAsync(IAsyncEnumerable<int> source)
{
await foreach (int _ in source.ConfigureAwait(false))
{
}
}
private static async Task ConsumeBufferedAsync(
IAsyncEnumerable<int> source,
CancellationToken ct)
{
await source.BufferedAsync(1, ct).ConfigureAwait(false);
}
private static async IAsyncEnumerable<int> RangeAsync(params int[] values)
{
foreach (int value in values)
{
await Task.Yield();
yield return value;
}
}
private sealed class CancellationProbeEnumerable : IAsyncEnumerable<int>
{
public ManualResetEventSlim EnumeratorCreated { get; } = new();
public bool EnumeratorCancellationObserved { get; private set; }
public IAsyncEnumerator<int> GetAsyncEnumerator(
CancellationToken cancellationToken = default)
{
return new CancellationProbeEnumerator(this, cancellationToken);
}
private sealed class CancellationProbeEnumerator : IAsyncEnumerator<int>
{
private readonly CancellationProbeEnumerable m_owner;
private readonly CancellationToken m_ct;
public CancellationProbeEnumerator(
CancellationProbeEnumerable owner,
CancellationToken ct)
{
m_owner = owner;
m_ct = ct;
owner.EnumeratorCreated.Set();
}
public int Current => 0;
public ValueTask DisposeAsync()
{
return default;
}
public async ValueTask<bool> MoveNextAsync()
{
try
{
await Task.Delay(Timeout.Infinite, m_ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
m_owner.EnumeratorCancellationObserved = true;
throw;
}
return false;
}
}
}
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();
}
}
}
}
Reference in New Issue View Git Blame Copy Permalink