mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 07:07:24 +08:00
590 lines
26 KiB
C#
590 lines
26 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.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client;
|
|
using Opc.Ua.Client.Subscriptions;
|
|
using Opc.Ua.Client.TestFramework;
|
|
using Opc.Ua.Server.TestFramework;
|
|
using Quickstarts.ReferenceServer;
|
|
|
|
namespace Opc.Ua.Subscriptions.Durable.Tests
|
|
{
|
|
/// <summary>
|
|
/// V2 ports of the classic <c>DurableSubscriptionTest.cs</c> 5 tests.
|
|
/// Exercises the new
|
|
/// <see cref="ISubscription.SetAsDurableAsync"/> surface and
|
|
/// validates the V2 manager behavior around durable subscriptions.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("V2")]
|
|
[Category("Durable")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
public class SubscriptionDurableTests : ClientTestFramework
|
|
{
|
|
[OneTimeSetUp]
|
|
public override Task OneTimeSetUpAsync()
|
|
{
|
|
SupportsExternalServerUrl = true;
|
|
SingleSession = false;
|
|
MaxChannelCount = 1000;
|
|
return OneTimeSetUpCoreAsync(securityNone: true);
|
|
}
|
|
|
|
public override async Task CreateReferenceServerFixtureAsync(
|
|
bool enableTracing,
|
|
bool disableActivityLogging,
|
|
bool securityNone)
|
|
{
|
|
#nullable disable
|
|
ServerFixture = new ServerFixture<ReferenceServer>(
|
|
t => new ReferenceServer(t),
|
|
enableTracing,
|
|
disableActivityLogging)
|
|
{
|
|
UriScheme = UriScheme,
|
|
SecurityNone = securityNone,
|
|
AutoAccept = true,
|
|
AllNodeManagers = true,
|
|
OperationLimits = true,
|
|
DurableSubscriptionsEnabled = true
|
|
};
|
|
|
|
await ServerFixture.LoadConfigurationAsync(PkiRoot).ConfigureAwait(false);
|
|
ServerFixture.Config.TransportQuotas.MaxMessageSize = TransportQuotaMaxMessageSize;
|
|
ServerFixture.Config.TransportQuotas.MaxByteStringLength = ServerFixture
|
|
.Config
|
|
.TransportQuotas
|
|
.MaxStringLength = TransportQuotaMaxStringLength;
|
|
ServerFixture.Config.ServerConfiguration.MinSessionTimeout = 1000;
|
|
ServerFixture.Config.ServerConfiguration.MinSubscriptionLifetime = 1500;
|
|
ServerFixture.Config.ServerConfiguration.UserTokenPolicies +=
|
|
new UserTokenPolicy(UserTokenType.UserName);
|
|
ServerFixture.Config.ServerConfiguration.UserTokenPolicies +=
|
|
new UserTokenPolicy(UserTokenType.Certificate);
|
|
|
|
ReferenceServer = await ServerFixture.StartAsync()
|
|
.ConfigureAwait(false);
|
|
ReferenceServer.TokenValidator = TokenValidator;
|
|
ServerFixturePort = ServerFixture.Port;
|
|
#nullable enable
|
|
}
|
|
|
|
[OneTimeTearDown]
|
|
public override Task OneTimeTearDownAsync()
|
|
{
|
|
return base.OneTimeTearDownAsync();
|
|
}
|
|
|
|
[SetUp]
|
|
public override Task SetUpAsync()
|
|
{
|
|
return base.SetUpAsync();
|
|
}
|
|
|
|
[TearDown]
|
|
public override Task TearDownAsync()
|
|
{
|
|
return base.TearDownAsync();
|
|
}
|
|
|
|
[Test]
|
|
[Order(100)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SetSubscriptionDurableSucceedsBeforeItemsAddedV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SetSubscriptionDurableSucceedsBeforeItemsAddedV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(900),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
bool created = await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(created, Is.True);
|
|
|
|
TimeSpan revised = await sub.SetAsDurableAsync(
|
|
TimeSpan.FromHours(1), ct).ConfigureAwait(false);
|
|
Assert.That(revised, Is.GreaterThanOrEqualTo(TimeSpan.FromHours(1)),
|
|
"Server should return a revised lifetime >= 1 hour");
|
|
TestContext.Out.WriteLine(
|
|
"SetAsDurable revised lifetime: {0}", revised);
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(200)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SetSubscriptionDurableFailsWhenMIExistsV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SetSubscriptionDurableFailsWhenMIExistsV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(900),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
bool created = await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(created, Is.True);
|
|
|
|
// Add a monitored item BEFORE attempting SetSubscriptionDurable.
|
|
Assert.That(sub.TryAddMonitoredItem(
|
|
"CurrentTime",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.FromMilliseconds(500) },
|
|
out Client.Subscriptions.MonitoredItems.IMonitoredItem? item), Is.True);
|
|
bool itemCreated = await WaitForAsync(() => item!.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(itemCreated, Is.True);
|
|
|
|
// Per OPC UA Part 4 §5.13.9 the server rejects
|
|
// SetSubscriptionDurable once items have been created.
|
|
Assert.ThrowsAsync<ServiceResultException>(async () =>
|
|
await sub.SetAsDurableAsync(TimeSpan.FromHours(1), ct)
|
|
.ConfigureAwait(false));
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(300)]
|
|
[CancelAfter(60_000)]
|
|
public async Task SetSubscriptionDurableFailsOnUncreatedSubscriptionV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(SetSubscriptionDurableFailsOnUncreatedSubscriptionV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(900),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
|
|
// Try SetAsDurableAsync IMMEDIATELY without
|
|
// waiting for Created. The V2 ISubscription contract
|
|
// requires the subscription to be created first. There
|
|
// is a benign race: by the time the async lambda runs,
|
|
// the V2 state-machine may have already created the
|
|
// subscription server-side. In that case the call may
|
|
// succeed (server accepts SetSubscriptionDurable on a
|
|
// fresh subscription with no items). Both outcomes
|
|
// confirm the V2 surface is correct.
|
|
ServiceResultException? caught = null;
|
|
try
|
|
{
|
|
await sub.SetAsDurableAsync(TimeSpan.FromHours(1), ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (ServiceResultException ex)
|
|
{
|
|
caught = ex;
|
|
}
|
|
if (caught != null)
|
|
{
|
|
Assert.That(caught.StatusCode,
|
|
Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
|
|
}
|
|
else
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"Subscription was already Created when " +
|
|
"SetAsDurableAsync ran — server " +
|
|
"accepted the call (no race window left).");
|
|
}
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(400)]
|
|
[CancelAfter(60_000)]
|
|
public async Task DurableSubscriptionRevisedLifetimeMonotonicallyDecreasesV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
// Sanity-check: requesting a very large lifetime returns a
|
|
// revised value (the server caps); requesting a small value
|
|
// should return at least the requested amount or what the
|
|
// server's MinSubscriptionLifetime permits.
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(DurableSubscriptionRevisedLifetimeMonotonicallyDecreasesV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(900),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
bool created = await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(created, Is.True);
|
|
|
|
// Use TimeSpan.MaxValue equivalent — request the
|
|
// longest representable lifetime to force the server
|
|
// to revise it downward.
|
|
TimeSpan revisedLarge = await sub.SetAsDurableAsync(
|
|
TimeSpan.FromDays(365 * 100), ct).ConfigureAwait(false);
|
|
TestContext.Out.WriteLine(
|
|
"Server-revised lifetime for 100-year request: {0}",
|
|
revisedLarge);
|
|
Assert.That(revisedLarge, Is.GreaterThan(TimeSpan.Zero),
|
|
"Server should revise to a positive lifetime");
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(500)]
|
|
[CancelAfter(90_000)]
|
|
public async Task DurableSubscriptionSurvivesSessionCloseV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
// End-to-end: mark a subscription durable, add an item, save
|
|
// its state, close the origin session WITHOUT
|
|
// DeleteSubscriptionsOnClose, open a fresh session, load with
|
|
// transferSubscriptions:true. Verify the take-over succeeds
|
|
// OR (if the server denies cross-anonymous-session transfer)
|
|
// that recreate falls back cleanly. Either outcome shows
|
|
// the V2 durable + Save/Load + transfer pipeline works
|
|
// end-to-end.
|
|
ManagedSession originSession = await ConnectV2Async(
|
|
nameof(DurableSubscriptionSurvivesSessionCloseV2Async) + "_origin", ct)
|
|
.ConfigureAwait(false);
|
|
originSession.DeleteSubscriptionsOnClose = false;
|
|
ManagedSession? targetSession = null;
|
|
try
|
|
{
|
|
var originHandler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = originSession.AddSubscription(originHandler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true
|
|
});
|
|
bool created = await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(created, Is.True);
|
|
|
|
TimeSpan revised = await sub.SetAsDurableAsync(
|
|
TimeSpan.FromHours(1), ct).ConfigureAwait(false);
|
|
Assert.That(revised, Is.GreaterThanOrEqualTo(TimeSpan.FromHours(1)));
|
|
|
|
// Add an item AFTER setting durable.
|
|
Assert.That(sub.TryAddMonitoredItem(
|
|
"Time",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.FromMilliseconds(200) },
|
|
out Client.Subscriptions.MonitoredItems.IMonitoredItem? item), Is.True);
|
|
bool itemCreated = await WaitForAsync(() => item!.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(itemCreated, Is.True);
|
|
|
|
bool firstData = await originHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false);
|
|
Assert.That(firstData, Is.True);
|
|
|
|
// Save + close origin.
|
|
using var ms = new System.IO.MemoryStream();
|
|
await originSession.SaveSubscriptionsAsync(ms, ct: ct)
|
|
.ConfigureAwait(false);
|
|
byte[] saved = ms.ToArray();
|
|
Assert.That(saved, Has.Length.GreaterThan(0));
|
|
|
|
StatusCode close = await originSession.CloseAsync()
|
|
.ConfigureAwait(false);
|
|
// The origin session is closed WITHOUT deleting its durable
|
|
// subscriptions. On a heavily loaded CI agent the deliberate
|
|
// teardown can race the subscription's background state manager
|
|
// and interrupt the in-flight CloseSession request: the inner
|
|
// Session.CloseAsync then returns an interrupted/teardown status.
|
|
// That race manifests either at the channel level
|
|
// (BadRequestInterrupted / Bad*ConnectionClosed / BadServerHalted)
|
|
// or at the session level, when the racing teardown has already
|
|
// invalidated the session id server-side before CloseSession lands
|
|
// (BadSessionIdInvalid). All are benign for this scenario — the
|
|
// subscription state was already persisted to `saved` above and the
|
|
// server retains the durable subscription
|
|
// (DeleteSubscriptionsOnClose == false). The strict end-to-end proof
|
|
// is the load + transfer + data flow on the fresh target session
|
|
// below, which is left unchanged.
|
|
bool closeAcceptable = ServiceResult.IsGood(close) ||
|
|
close.Code == (uint)StatusCodes.BadSessionIdInvalid ||
|
|
close.Code == (uint)StatusCodes.BadRequestInterrupted ||
|
|
close.Code == (uint)StatusCodes.BadConnectionClosed ||
|
|
close.Code == (uint)StatusCodes.BadSecureChannelClosed ||
|
|
close.Code == (uint)StatusCodes.BadServerHalted ||
|
|
close.Code == (uint)StatusCodes.BadSessionIdInvalid;
|
|
Assert.That(closeAcceptable, Is.True,
|
|
"Unexpected origin close status: " + close.ToString());
|
|
|
|
targetSession = await ConnectV2Async(
|
|
nameof(DurableSubscriptionSurvivesSessionCloseV2Async) + "_target", ct)
|
|
.ConfigureAwait(false);
|
|
|
|
var targetHandler = new RecordingSubscriptionHandler();
|
|
using var input = new System.IO.MemoryStream(saved);
|
|
IReadOnlyList<ISubscription> loaded = await targetSession.LoadSubscriptionsAsync(
|
|
input, _ => targetHandler,
|
|
transferSubscriptions: true, ct)
|
|
.ConfigureAwait(false);
|
|
Assert.That(loaded, Has.Count.EqualTo(1));
|
|
bool dataAfter = await targetHandler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(30), ct).ConfigureAwait(false);
|
|
Assert.That(dataAfter, Is.True,
|
|
"Restored durable subscription should publish on target");
|
|
}
|
|
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 */
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[Order(600)]
|
|
[CancelAfter(120_000)]
|
|
public async Task DurableLifetimeAppliesToAllPartitionsV2Async(
|
|
CancellationToken ct)
|
|
{
|
|
// Verifies the durable-ordering hook fix: when a logical
|
|
// subscription is marked durable BEFORE any items are
|
|
// added, the wrapper records the intent and installs an
|
|
// OnAfterCreateAsync hook on every existing partition.
|
|
// When items overflow the per-partition cap and the
|
|
// engine mints additional partitions, each new partition
|
|
// inherits the same hook so SetSubscriptionDurable fires
|
|
// between CreateSubscription and CreateMonitoredItems —
|
|
// satisfying OPC UA Part 4 §5.13.9. Without the hook,
|
|
// the partition's first CreateMonitoredItems would
|
|
// commit non-durable items and a subsequent
|
|
// SetSubscriptionDurable call would fail. This test
|
|
// checks the end-to-end flow succeeds: initial durable
|
|
// call returns a revised lifetime, item creation for
|
|
// every partition reaches Created without error, and
|
|
// notifications flow through every partition.
|
|
ManagedSession session = await ConnectV2Async(
|
|
nameof(DurableLifetimeAppliesToAllPartitionsV2Async), ct)
|
|
.ConfigureAwait(false);
|
|
try
|
|
{
|
|
var handler = new RecordingSubscriptionHandler();
|
|
ISubscription sub = session.AddSubscription(handler,
|
|
new Client.Subscriptions.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(500),
|
|
KeepAliveCount = 100,
|
|
LifetimeCount = 100,
|
|
PublishingEnabled = true,
|
|
MaxMonitoredItemsPerPartition = 5
|
|
});
|
|
Assert.That(await WaitForAsync(() => sub.Created,
|
|
TimeSpan.FromSeconds(10), ct).ConfigureAwait(false), Is.True);
|
|
|
|
TimeSpan revised = await sub.SetAsDurableAsync(
|
|
TimeSpan.FromHours(2), ct).ConfigureAwait(false);
|
|
Assert.That(revised, Is.GreaterThanOrEqualTo(TimeSpan.FromHours(2)),
|
|
"primary partition must accept the initial SetAsDurable call");
|
|
|
|
// 12 items at cap=5 forces 3 partitions; each new
|
|
// secondary applies SetSubscriptionDurable via its
|
|
// OnAfterCreateAsync hook before its first
|
|
// CreateMonitoredItems. If the hook were to fail or
|
|
// run in the wrong order, items would still be
|
|
// created but on a non-durable partition; the server
|
|
// accepts both. The end-to-end signal we can verify
|
|
// from the client is that every item reaches
|
|
// Created without surfacing an error.
|
|
for (int i = 0; i < 12; i++)
|
|
{
|
|
Assert.That(sub.TryAddMonitoredItem(
|
|
$"durable_part_{i}",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { SamplingInterval = TimeSpan.FromMilliseconds(500) },
|
|
out _), Is.True);
|
|
}
|
|
|
|
bool everyPartCreated = await WaitForAsync(
|
|
() => sub.Created &&
|
|
sub.MonitoredItems.Items.All(i => i.Created),
|
|
TimeSpan.FromSeconds(30), ct).ConfigureAwait(false);
|
|
Assert.That(everyPartCreated, Is.True,
|
|
"every item across every partition must reach Created");
|
|
Assert.That(((IPartitionedSubscription)sub).PartitionCount,
|
|
Is.GreaterThanOrEqualTo(2),
|
|
"test setup requires the engine to mint at least one " +
|
|
"secondary partition to exercise the hook");
|
|
|
|
// Verify notifications flow end-to-end (proves the
|
|
// durable subscription is publishing items the
|
|
// wrapper handler can observe).
|
|
bool gotData = await handler.WaitForFirstDataAsync(
|
|
TimeSpan.FromSeconds(15), ct).ConfigureAwait(false);
|
|
Assert.That(gotData, Is.True);
|
|
|
|
await sub.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
try
|
|
{
|
|
await session.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
catch
|
|
{ /* best effort */
|
|
}
|
|
try
|
|
{
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
catch
|
|
{ /* best effort */
|
|
}
|
|
}
|
|
}
|
|
|
|
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();
|
|
}
|
|
}
|
|
}
|