mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 07:07:24 +08:00
Marc Schier
49305d9085
# Description V2 subscription engine feature parity round to make the classic engine eligible for deprecation. Splits the test surface so the classic and V2 engines can evolve independently, ports the gaps the classic-engine integration tests had been exercising, and lands the public V2 API additions needed to close them. - Splits `Tests/Opc.Ua.Subscriptions.Tests` into a classic (`Opc.Ua.Subscriptions.Classic.Tests`) and a V2 (`Opc.Ua.Subscriptions.Tests`) project to shrink CI time and let the two engines diverge cleanly. - Adds the V2 public-surface gaps required by the test ports (`Subscription.ServerId` on the class, `SetSubscriptionDurableAsync`, `SendInitialValuesOnTransfer`, unified `OnSubscriptionStateChangedAsync` handler callback, per-item `ConditionRefreshAsync`, fluent `LoadSubscriptionsAsync(Stream, …)` and `Subscription.Snapshot()` / `RestoreAsync()` for save/load + transfer). - Lands handler-centric design (no `PublishingStopped` property handlers maintain derived state); always-sequential per-subscription delivery documented + tested. - **Adds short and long haul stability tests for the V2 `ManagedSession` + subscription engine. Closes #3744** (see "Issue #3744 Haul tests" below). - **Migrates SubscriptionStateSnapshot / MonitoredItemStateSnapshot to `[DataType]` source-generator codecs**. - `SubscriptionDurableV2Tests.cs` 5 V2 ports of the classic durable test patterns. - `V2FollowUpCoverageTests.cs` handler state-change callback, fluent stream-based `LoadSubscriptionsAsync`, `SendInitialValuesOnTransfer`, snapshot edge cases (empty / `DataChangeFilter` round-trip / concurrent mutation). - `SubscriptionFailoverV2Tests.cs` forced channel-break with and without `WithTransferSubscriptionsOnRecreate`. - `MonitoredItemConditionRefreshLiveV2Tests.cs` — live `ConditionRefresh` against the reference server's event source; asserts `RefreshStart` / `RefreshEnd` events flow through the handler. ## Fixes - #3744 Haul tests (V2 ManagedSession + subscription engine) `Tests/Opc.Ua.Sessions.Tests/ManagedSessionStabilityTest.cs` adds the short/long haul coverage requested by [#3744](https://github.com/OPCFoundation/UA-.NETStandard/issues/3744) plus the bonus fault-injection variant: | Test | Duration | Category | Purpose | |---|---|---|---| | `ShortHaulManagedSessionV2Async` | 2 minutes | `ManagedSessionHaul` (`[Explicit]`) | Quick local sanity check of the V2 monotonicity contract. | | `LongHaulManagedSessionV2Async` | 90 minutes (configurable via `TEST_DURATION_MINUTES`) | `ManagedSessionHaul` (`[Explicit]`) | Long-running V2 stability + monotonicity. | | `LongHaulManagedSessionWithFaultInjectionV2Async` | 90 minutes (configurable) + fault every `FAULT_INJECTION_INTERVAL_SECONDS` (default 60s) | `ManagedSessionHaul` (`[Explicit]`) | **Bonus** — force-closes the subscriber transport channel periodically; verifies V2 manager transfers/recreates the subscription and continues monotonic delivery across reconnects. | **Pattern.** A writer `ManagedSession` increments a `uint32` counter and writes it into `Scalar_Static_Mass_UInt32_UInt32_00` every 250ms. A subscriber `ManagedSession` runs a V2 subscription with a single monitored item on the same node. `MonotonicCounterHandler` asserts each received sample is *strictly greater than* the previous one. V2 `ISubscriptionNotificationHandler` per-subscription ordering guarantee. Skipped values (sampling rate < write rate) are tolerated; duplicates and reorderings are violations.
565 lines
25 KiB
C#
565 lines
25 KiB
C#
/* ========================================================================
|
|
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
|
|
*
|
|
* OPC Foundation MIT License 1.00
|
|
*
|
|
* Permission is hereby granted, free of charge, to any person
|
|
* obtaining a copy of this software and associated documentation
|
|
* files (the "Software"), to deal in the Software without
|
|
* restriction, including without limitation the rights to use,
|
|
* copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
* copies of the Software, and to permit persons to whom the
|
|
* Software is furnished to do so, subject to the following
|
|
* conditions:
|
|
*
|
|
* The above copyright notice and this permission notice shall be
|
|
* included in all copies or substantial portions of the Software.
|
|
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
|
|
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
|
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
|
|
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
|
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
|
|
* OTHER DEALINGS IN THE SOFTWARE.
|
|
*
|
|
* The complete license agreement can be found here:
|
|
* http://opcfoundation.org/License/MIT/1.00/
|
|
* ======================================================================*/
|
|
|
|
#nullable enable
|
|
|
|
#pragma warning disable CA2016
|
|
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client;
|
|
using Opc.Ua.Client.Subscriptions;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
|
|
using Opc.Ua.Client.TestFramework;
|
|
|
|
namespace Opc.Ua.Subscriptions.Tests
|
|
{
|
|
/// <summary>
|
|
/// V2 follow-up coverage: handler state-change callback,
|
|
/// fluent stream-based LoadSubscriptionsAsync, SendInitialValuesOnTransfer
|
|
/// behavior, snapshot edge cases (empty / with-filter / concurrent).
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("V2")]
|
|
[Category("FollowUp")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
public class SubscriptionCoverageTests : ClientTestFramework
|
|
{
|
|
[OneTimeSetUp]
|
|
public override Task OneTimeSetUpAsync()
|
|
{
|
|
SupportsExternalServerUrl = true;
|
|
SingleSession = false;
|
|
return OneTimeSetUpCoreAsync(securityNone: true);
|
|
}
|
|
|
|
[OneTimeTearDown]
|
|
public override Task OneTimeTearDownAsync()
|
|
{
|
|
return base.OneTimeTearDownAsync();
|
|
}
|
|
|
|
[SetUp]
|
|
public override Task SetUpAsync()
|
|
{
|
|
return base.SetUpAsync();
|
|
}
|
|
|
|
[TearDown]
|
|
public override Task TearDownAsync()
|
|
{
|
|
return base.TearDownAsync();
|
|
}
|
|
|
|
// ===== 1. OnSubscriptionStateChangedAsync fires on lifecycle =====
|
|
|
|
[Test]
|
|
[Order(100)]
|
|
[CancelAfter(60_000)]
|
|
public async Task HandlerStateChangedFiresOnLifecycleV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(HandlerStateChangedFiresOnLifecycleV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
bool created = await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(created, Is.True);
|
|
|
|
bool sawStateChanges = await WaitForAsync(
|
|
() => handler.StateChangedCount > 0,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(sawStateChanges, Is.True,
|
|
"OnSubscriptionStateChangedAsync should fire during subscription lifecycle");
|
|
|
|
IReadOnlyList<RecordedStateChange> snap = handler.GetStateChangeSnapshot();
|
|
Assert.That(snap, Is.Not.Empty);
|
|
|
|
// Sub-test: derive a "is currently created" indicator from
|
|
// the callback stream — proves the handler-maintains-state
|
|
// pattern that replaced the dropped PublishingStopped
|
|
// property on ISubscription.
|
|
bool sawCreated = false;
|
|
foreach (RecordedStateChange c in snap)
|
|
{
|
|
if (c.State == Opc.Ua.Client.Subscriptions.SubscriptionState.Created ||
|
|
c.State == Opc.Ua.Client.Subscriptions.SubscriptionState.Modified ||
|
|
c.State == Opc.Ua.Client.Subscriptions.SubscriptionState.Opened)
|
|
{
|
|
sawCreated = true;
|
|
}
|
|
}
|
|
Assert.That(sawCreated, Is.True,
|
|
"State stream should include at least one Created / " +
|
|
"Modified / Opened transition");
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
// ===== 2. Fluent stream-based LoadSubscriptionsAsync =====
|
|
|
|
[Test]
|
|
[Order(200)]
|
|
[CancelAfter(60_000)]
|
|
public async Task FluentLoadSubscriptionsAsyncStreamV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession originSession = await ConnectV2Async(
|
|
nameof(FluentLoadSubscriptionsAsyncStreamV2Async) + "_origin", ct)
|
|
.ConfigureAwait(false);
|
|
ManagedSession? targetSession = null;
|
|
try
|
|
{
|
|
var originHandler = new RecordingSubscriptionHandler();
|
|
ISubscription origin = originSession.AddSubscription(originHandler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
Assert.That(await WaitForAsync(() => origin.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(origin.TryAddMonitoredItem("Time",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.FromMilliseconds(200) },
|
|
out _), Is.True);
|
|
Assert.That(await originHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
using var ms = new MemoryStream();
|
|
await originSession.SaveSubscriptionsAsync(ms, ct: ct)
|
|
.ConfigureAwait(false);
|
|
byte[] saved = ms.ToArray();
|
|
Assert.That(saved, Has.Length.GreaterThan(0));
|
|
|
|
targetSession = await ConnectV2Async(
|
|
nameof(FluentLoadSubscriptionsAsyncStreamV2Async) + "_target", ct)
|
|
.ConfigureAwait(false);
|
|
var targetHandler = new RecordingSubscriptionHandler();
|
|
using var input = new MemoryStream(saved);
|
|
IReadOnlyList<ISubscription> loaded = await targetSession
|
|
.LoadSubscriptionsAsync(input, _ => targetHandler,
|
|
transferSubscriptions: false, ct)
|
|
.ConfigureAwait(false);
|
|
Assert.That(loaded, Has.Count.EqualTo(1));
|
|
Assert.That(await WaitForAsync(() => loaded[0].Created,
|
|
TimeSpan.FromSeconds(15), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(await targetHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(15), ct).ConfigureAwait(false), Is.True);
|
|
|
|
await loaded[0].DisposeAsync().ConfigureAwait(false);
|
|
await origin.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
try { await originSession.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await originSession.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
if (targetSession != null)
|
|
{
|
|
try { await targetSession.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await targetSession.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
}
|
|
}
|
|
}
|
|
|
|
// ===== 3. SendInitialValuesOnTransfer behavior =====
|
|
|
|
[Test]
|
|
[Order(300)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SendInitialValuesOnTransferV2Async(CancellationToken ct)
|
|
{
|
|
// When the test runs against a reference server that denies
|
|
// cross-anonymous-session transfer with BadUserAccessDenied,
|
|
// the V2 manager falls back to recreate. Recreate always
|
|
// delivers an initial value (server first-sample behavior),
|
|
// so the SendInitialValuesOnTransfer option's effect can't
|
|
// be distinguished. Document that and verify the option
|
|
// propagates structurally + the subscription works after
|
|
// restore regardless of which path the server took.
|
|
string saveFile = Path.Combine(Path.GetTempPath(),
|
|
$"V2InitVals-{Guid.NewGuid():N}.bin");
|
|
ManagedSession originSession = await ConnectV2Async(
|
|
nameof(SendInitialValuesOnTransferV2Async) + "_origin", ct)
|
|
.ConfigureAwait(false);
|
|
originSession.DeleteSubscriptionsOnClose = false;
|
|
ManagedSession? targetSession = null;
|
|
try
|
|
{
|
|
var originHandler = new RecordingSubscriptionHandler();
|
|
ISubscription origin = originSession.AddSubscription(originHandler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true,
|
|
SendInitialValuesOnTransfer = true
|
|
});
|
|
Assert.That(await WaitForAsync(() => origin.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(origin.TryAddMonitoredItem("Time",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.FromMilliseconds(200) },
|
|
out _), Is.True);
|
|
Assert.That(await originHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
// Snapshot includes the SendInitialValuesOnTransfer
|
|
// option through SubscriptionStateSnapshot.ToOptions().
|
|
SubscriptionStateSnapshot snap = ((Opc.Ua.Client.Subscriptions.Subscription)origin).Snapshot();
|
|
Assert.That(snap.SendInitialValuesOnTransfer, Is.True,
|
|
"SendInitialValuesOnTransfer option must round-trip through Snapshot");
|
|
|
|
using (var output = File.Create(saveFile))
|
|
{
|
|
await originSession.SaveSubscriptionsAsync(output, ct: ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
Assert.That(await originSession.CloseAsync().ConfigureAwait(false),
|
|
Is.EqualTo(StatusCodes.Good));
|
|
|
|
targetSession = await ConnectV2Async(
|
|
nameof(SendInitialValuesOnTransferV2Async) + "_target", ct)
|
|
.ConfigureAwait(false);
|
|
var targetHandler = new RecordingSubscriptionHandler();
|
|
using var input = File.OpenRead(saveFile);
|
|
IReadOnlyList<ISubscription> loaded = await targetSession
|
|
.LoadSubscriptionsAsync(input, _ => targetHandler,
|
|
transferSubscriptions: true, ct)
|
|
.ConfigureAwait(false);
|
|
Assert.That(loaded, Has.Count.EqualTo(1));
|
|
bool gotData = await targetHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(15), ct).ConfigureAwait(false);
|
|
Assert.That(gotData, Is.True,
|
|
"Target subscription should publish (transfer or recreate)");
|
|
|
|
await loaded[0].DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
try { await originSession.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
if (targetSession != null)
|
|
{
|
|
try { await targetSession.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await targetSession.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
}
|
|
try { File.Delete(saveFile); } catch { /* best effort */ }
|
|
}
|
|
}
|
|
|
|
// ===== 4-6. Snapshot/Restore edge cases =====
|
|
|
|
[Test]
|
|
[Order(400)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SnapshotEmptySubscriptionRoundTripV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SnapshotEmptySubscriptionRoundTripV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
ManagedSession? target = null;
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
Assert.That(await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(sub.MonitoredItems.Count, Is.Zero);
|
|
|
|
SubscriptionStateSnapshot snap = ((Opc.Ua.Client.Subscriptions.Subscription)sub).Snapshot();
|
|
Assert.That(snap.MonitoredItems.Count, Is.Zero);
|
|
|
|
target = await ConnectV2Async(
|
|
nameof(SnapshotEmptySubscriptionRoundTripV2Async) + "_target", ct)
|
|
.ConfigureAwait(false);
|
|
ISubscription restored = await ((Opc.Ua.Client.Subscriptions.SubscriptionManager)target.SubscriptionManager)
|
|
.RestoreAsync(
|
|
new RecordingSubscriptionHandler(),
|
|
snap,
|
|
transferSubscriptions: false,
|
|
ct)
|
|
.ConfigureAwait(false);
|
|
Assert.That(await WaitForAsync(() => restored.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(restored.MonitoredItems.Count, Is.Zero);
|
|
|
|
await restored.DisposeAsync().ConfigureAwait(false);
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
try { await session.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await session.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
if (target != null)
|
|
{
|
|
try { await target.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await target.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(500)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SnapshotWithDataChangeFilterRoundTripV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SnapshotWithDataChangeFilterRoundTripV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
ManagedSession? target = null;
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
Assert.That(await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
var dataChangeFilter = new DataChangeFilter
|
|
{
|
|
Trigger = DataChangeTrigger.StatusValueTimestamp,
|
|
DeadbandType = (uint)DeadbandType.None,
|
|
DeadbandValue = 0.0
|
|
};
|
|
Assert.That(sub.TryAddMonitoredItem("Trigger",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with
|
|
{
|
|
SamplingInterval = TimeSpan.FromMilliseconds(250),
|
|
Filter = dataChangeFilter
|
|
},
|
|
out Opc.Ua.Client.Subscriptions.MonitoredItems.IMonitoredItem? item), Is.True);
|
|
Assert.That(await WaitForAsync(() => item!.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
SubscriptionStateSnapshot snap = ((Opc.Ua.Client.Subscriptions.Subscription)sub).Snapshot();
|
|
Assert.That(snap.MonitoredItems.Count, Is.EqualTo(1));
|
|
MonitoredItemStateSnapshot itemSnap = snap.MonitoredItems[0];
|
|
Assert.That(itemSnap.Filter, Is.Not.Null);
|
|
Assert.That(itemSnap.Filter, Is.InstanceOf<DataChangeFilter>());
|
|
DataChangeFilter restoredFilter = (DataChangeFilter)itemSnap.Filter!;
|
|
Assert.That(restoredFilter.Trigger,
|
|
Is.EqualTo(dataChangeFilter.Trigger));
|
|
Assert.That(restoredFilter.DeadbandType,
|
|
Is.EqualTo(dataChangeFilter.DeadbandType));
|
|
|
|
// Round-trip through the manager
|
|
target = await ConnectV2Async(
|
|
nameof(SnapshotWithDataChangeFilterRoundTripV2Async) + "_target", ct)
|
|
.ConfigureAwait(false);
|
|
ISubscription restored = await ((Opc.Ua.Client.Subscriptions.SubscriptionManager)target.SubscriptionManager)
|
|
.RestoreAsync(
|
|
new RecordingSubscriptionHandler(),
|
|
snap,
|
|
transferSubscriptions: false,
|
|
ct)
|
|
.ConfigureAwait(false);
|
|
Assert.That(await WaitForAsync(() => restored.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
Assert.That(restored.MonitoredItems.Count, Is.EqualTo(1u));
|
|
|
|
await restored.DisposeAsync().ConfigureAwait(false);
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
try { await session.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await session.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
if (target != null)
|
|
{
|
|
try { await target.CloseAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
try { await target.DisposeAsync().ConfigureAwait(false); }
|
|
catch { /* best effort */ }
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(600)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SnapshotUnderConcurrentMutationV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SnapshotUnderConcurrentMutationV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Opc.Ua.Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
Assert.That(await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
// Concurrently add items while taking snapshots. The
|
|
// snapshot must always return an internally consistent
|
|
// count (the names must all exist) — proves the
|
|
// manager-lock contract.
|
|
const int itemCount = 20;
|
|
var addTasks = new Task[itemCount];
|
|
for (int i = 0; i < itemCount; i++)
|
|
{
|
|
int idx = i;
|
|
addTasks[i] = Task.Run(() => sub.TryAddMonitoredItem(
|
|
string.Format(System.Globalization.CultureInfo.InvariantCulture,
|
|
"item-{0}", idx),
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.Zero },
|
|
out _), ct);
|
|
}
|
|
var subConcrete = (Opc.Ua.Client.Subscriptions.Subscription)sub;
|
|
var snapTasks = new Task<SubscriptionStateSnapshot>[5];
|
|
for (int i = 0; i < snapTasks.Length; i++)
|
|
{
|
|
snapTasks[i] = Task.Run(() => subConcrete.Snapshot(), ct);
|
|
}
|
|
await Task.WhenAll(addTasks).ConfigureAwait(false);
|
|
SubscriptionStateSnapshot[] snaps = await Task
|
|
.WhenAll(snapTasks).ConfigureAwait(false);
|
|
|
|
foreach (SubscriptionStateSnapshot s in snaps)
|
|
{
|
|
int count = s.MonitoredItems.Count;
|
|
// Snapshot is captured at a point in time — count is
|
|
// between 0 and itemCount inclusive. The critical
|
|
// invariant: each captured item has a non-empty name
|
|
// (i.e. no partially-constructed entries leaked).
|
|
foreach (MonitoredItemStateSnapshot it in s.MonitoredItems)
|
|
{
|
|
Assert.That(it.Name, Is.Not.Null.And.Not.Empty);
|
|
}
|
|
}
|
|
|
|
Assert.That(sub.MonitoredItems.Count, Is.EqualTo((uint)itemCount));
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
// ===== helpers =====
|
|
|
|
private async Task<ManagedSession> ConnectV2Async(
|
|
string sessionName, CancellationToken ct)
|
|
{
|
|
ConfiguredEndpoint endpoint = await ClientFixture
|
|
.GetEndpointAsync(ServerUrl, SecurityPolicies.None)
|
|
.ConfigureAwait(false);
|
|
return await new ManagedSessionBuilder(ClientFixture.Config, Telemetry)
|
|
.UseEndpoint(endpoint)
|
|
.WithSessionName(sessionName)
|
|
.WithSessionTimeout(TimeSpan.FromSeconds(120))
|
|
.ConnectAsync(ct).ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task<bool> WaitForAsync(
|
|
Func<bool> predicate, TimeSpan timeout, CancellationToken ct)
|
|
{
|
|
DateTime deadline = DateTime.UtcNow + timeout;
|
|
while (DateTime.UtcNow < deadline)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
if (predicate())
|
|
{
|
|
return true;
|
|
}
|
|
await Task.Delay(50, ct).ConfigureAwait(false);
|
|
}
|
|
return predicate();
|
|
}
|
|
}
|
|
}
|