GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-29 15:11:53 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
UA-.NETStandard/tests/Opc.Ua.Subscriptions.Tests/StreamingSubscriptionCoverageTests.cs
T
Wolfgang ColsmanandMarc Schier be9ccf912c Client: evaluate the durable subscription lifetime on every physical create (#4882) 
# Description

## Problem

`ISubscription.SetAsDurableAsync(TimeSpan lifetime, CancellationToken
ct)` marks a subscription durable with the Server method
`SetSubscriptionDurable` (OPC 10000-5, "SetSubscriptionDurable Method";
OPC 10000-4, "Durable Subscriptions"). `SetSubscriptionDurable` must be
called before the subscription has monitored items; a conforming server
rejects it afterwards with `Bad_InvalidState`.

`LogicalSubscription` stores the `TimeSpan` and installs an
`OnAfterCreateAsync` hook on each partition. The partition state machine
runs the hook after every `CreateSubscription` and before the first
`CreateMonitoredItems`, so durability survives every later physical
create:

- recreate after a service call reports `BadSubscriptionIdInvalid`
(`Subscription.ApplyStateChangesAsync` → `ResetToRecreateAsync`),
- recreate after `TransferSubscriptions` fails on session recreate
(`SubscriptionManager.RecreateSubscriptionsAsync` → `RecreateAsync`) or
on load (`RestoreTransferAsync` → `ResetToRecreateAsync`),
- recreate after a status-change timeout or an unsolicited
`Good_SubscriptionTransferred`,
- a new secondary partition (`TryApplyDurableToNewPartition`).

The hook always re-applies the lifetime fixed by the first call. An
application whose durable lifetime must not exceed a bound that shrinks
over time (for example an authorization or retention window that ends at
a known instant, after which the server must no longer queue
notifications) cannot keep that bound after a recreate, and it has no
way to supply a new value at the right moment:

- calling `SetAsDurableAsync` again after the recreate is rejected once
monitored items exist;
- `ISubscriptionNotificationHandler.OnSubscriptionStateChangedAsync` is
dispatched fire-and-forget, so a call made from it can arrive after the
first `CreateMonitoredItems`;
- `SubscriptionOptions` has no durable setting and `OnAfterCreateAsync`
is internal.

## API

```csharp
namespace Opc.Ua.Client.Subscriptions
{
    public interface ISubscription
    {
        // existing
        ValueTask<TimeSpan> SetAsDurableAsync(TimeSpan lifetime, CancellationToken ct = default);

        // new
        ValueTask<TimeSpan> SetAsDurableAsync(Func<TimeSpan?> lifetime, CancellationToken ct = default);
    }
}
```

## Behaviour

- The provider is evaluated once for the call itself; the value is sent
to every partition that is already created (unchanged: the call returns
the minimum revised lifetime of those partitions, or the requested value
when none is created yet).
- The provider is stored and evaluated again by the existing
after-create hook before every later physical create, between
`CreateSubscription` and the first `CreateMonitoredItems` of that
create.
- A `null` result requests no durability for that create: no
`SetSubscriptionDurable` is sent and the recreated subscription is a
regular one. A `null` result for the call itself returns `TimeSpan.Zero`
without contacting the server.
- An exception thrown by the provider during a later create is handled
exactly like a rejected `SetSubscriptionDurable` call today: the state
machine logs a warning, reports `SubscriptionState.Modified`, and the
subscription continues (its monitored items are created). An exception
during the call itself propagates to the caller.
- `SetAsDurableAsync(TimeSpan)` now forwards to `SetAsDurableAsync(() =>
lifetime)`; its behaviour is unchanged.
- `docs/Subscriptions.md` describes the provider and corrects the
statement that the hook is cleared after invocation (on master it stays
installed).

Not included: a notification reporting the outcome (revised lifetime or
status) of each re-applied `SetSubscriptionDurable`. Today a failed
re-apply is only logged and surfaced as `SubscriptionState.Modified`;
this could be a follow-up if maintainers want it.

## Tests

New tests in
`tests/Opc.Ua.Client.Tests/Subscription/SubscriptionTests.DurableLifetime.cs`
drive a real partition (`Subscription`) behind
`SubscriptionManager`/`LogicalSubscription` against mocked services and
record `CreateSubscription`, `SetSubscriptionDurable` (requested hours),
`CreateMonitoredItems`, `TransferSubscriptions` and
`DeleteSubscriptions` in call order. The provider returns 10 h for the
initial call and 7 h afterwards:

-
`DurableLifetimeProviderIsEvaluatedOnRecreateAfterSubscriptionIdInvalidAsync`:
`ModifySubscription` returns `BadSubscriptionIdInvalid`; the recreate
sends 7 h between `CreateSubscription` and the first
`CreateMonitoredItems`.
-
`DurableLifetimeProviderIsEvaluatedOnRecreateAfterFailedTransferAsync`:
`TransferSubscriptions` on session recreate returns
`BadSubscriptionIdInvalid`; delete, create, 7 h, items.
- `DurableLifetimeProviderIsEvaluatedForNewPartitionAsync`:
`MaxMonitoredItemsPerPartition = 1`; the second item mints a partition
that sends 7 h before its items.
-
`DurableLifetimeProviderReturningNullSendsNoSetSubscriptionDurableAsync`:
provider returns `null` on recreate; no `SetSubscriptionDurable`, items
are created.
-
`DurableLifetimeProviderThatThrowsLeavesRecreatedSubscriptionRunningAsync`:
provider throws on recreate; no `SetSubscriptionDurable`, items are
created, subscription stays created.
- `DurableFixedLifetimeIsReappliedOnRecreateAsync`: regression of the
`TimeSpan` overload (10 h again on recreate).

Unit tests in `LogicalSubscriptionTests`:
`SetAsDurableWithProviderEvaluatesItOnceForAllCreatedPartitionsAsync`,
`SetAsDurableWithProviderReturningNullSendsNothingAsync`,
`SetAsDurableWithNullProviderThrowsAsync`.

Negative control: with the hook evaluating the provider once at
installation (the previous fixed-value behaviour), the
recreate-after-`BadSubscriptionIdInvalid`, failed-transfer, null and
throwing-provider tests fail (4 of 10).

Local results (Windows, .NET SDK 10.0.402, Release):

| Run | Result |
| --- | --- |
| `Opc.Ua.Client` build, net48/net8.0/net9.0/net10.0 | 0 warnings, 0
errors |
| `SubscriptionTests` + `LogicalSubscriptionTests`, net10.0 | 105/105
passed |
| `SubscriptionTests` + `LogicalSubscriptionTests`, net48 | 105/105
passed |
| `Opc.Ua.Client.Tests`, namespace `Opc.Ua.Client.Subscriptions`,
net10.0 | 353/353 passed |
| `Opc.Ua.Subscriptions.Tests` `SubscriptionManagerCoverageTests` +
`StreamingSubscriptionCoverageTests`, net10.0 | 72/72 passed |
| `Opc.Ua.Subscriptions.Durable.Tests` `SubscriptionDurableTests` +
`DurableSubscriptionTest*` (in-process reference server with durable
subscriptions enabled), net10.0 | 21/21 passed |

## Related observation (not changed here)

A secondary partition minted by `SubscriptionManager.Add`'s partition
factory starts its state machine in the `Subscription` constructor,
before the factory calls `TryApplyDurableToNewPartition`. When
`CreateSubscription` completes before the factory returns (observed with
a synchronously completing mock), `RunAfterCreateHookAsync` finds no
hook and the new partition is created without `SetSubscriptionDurable`,
for the `TimeSpan` overload as well. With a real round trip the hook is
installed first;
`DurableLifetimeProviderIsEvaluatedForNewPartitionAsync` therefore holds
the secondary's `CreateSubscription` response until the factory has
returned. Installing the hook before the partition's first state-manager
pass would remove the window.

## Compatibility

- Source and binary compatible for callers of `ISubscription`.
- `ISubscription` gains a member. Custom implementations of the public
interface (outside the stack; the stack's own implementations are
internal) must add the overload; a default interface implementation is
not possible because the library also targets .NET Framework 4.8. The
three test doubles in the repository were updated.
- No wire, configuration or server change.

## Related Issues

- Independent of #4881; with #4881 the lifetime provider is also applied
to a new partition that is created before the partition factory returns.

## Checklist

- [x] 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.
- [x] 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. (Run
locally: the affected test classes on net48 and net10.0 and the whole
Opc.Ua.Client.Tests project on net10.0; not the whole UA.slnx.)
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.

---------

Co-authored-by: Marc Schier <marcschier@users.noreply.github.com>
2026-10-22 18:51:01 +02:00

1382 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<TimeSpan> SetAsDurableAsync(
Func<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