mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-24 02:58:36 +08:00
Marc Schier
9bfc780e14
# Description Update 55 NuGet dependencies to the latest stable releases available on 2026-10-19, preserving documented compatibility pins and framework-specific version bands. Existing OpenUSD prereleases remain unchanged because no newer releases are available. Migrate existing tests to NUnit 5: await asynchronous assertions, propagate async through helpers, handle nullable predicates, and replace obsolete `Order` attributes with sequencing dependencies. Sequencing dependencies use `AllowFailure = true`; filtering a dependent test also runs prerequisites. Adjust fixture-level parallelism where NUnit dependency scheduling requires it. Keep Roslyn API/runtime pins intact; select newer Immutable/Metadata runtime packages only for test and profiling hosts requiring them. Update the approved SDK pin to 10.0.402 across local, CI, and container builds. Document NUnit 5 conventions.
643 lines
28 KiB
C#
643 lines
28 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/
|
|
* ======================================================================*/
|
|
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Reflection;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client;
|
|
using Opc.Ua.Client.TestFramework;
|
|
using Opc.Ua.Tests;
|
|
|
|
namespace Opc.Ua.Subscriptions.Classic.Tests
|
|
{
|
|
/// <summary>
|
|
/// Tests for subscription transfer scenarios.
|
|
/// Each fixture combines one <see cref="TransferType"/> with acknowledged or retained
|
|
/// origin notifications, initial values, and sequential publishing.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("TransferSubscription")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
[TestFixtureSource(nameof(TransferTypeArgs))]
|
|
[Parallelizable(ParallelScope.Fixtures)]
|
|
public class TransferSubscriptionTest : ClientTestFramework
|
|
{
|
|
/// <summary>
|
|
/// Fixture argument list: one fixture instance per <see cref="TransferType"/> value.
|
|
/// </summary>
|
|
public static readonly object[] TransferTypeArgs =
|
|
[
|
|
new object[] { TransferType.KeepOpen },
|
|
new object[] { TransferType.CloseSession },
|
|
new object[] { TransferType.DisconnectedAck },
|
|
new object[] { TransferType.DisconnectedRepublish },
|
|
new object[] { TransferType.DisconnectedRepublishDelayedAck }
|
|
];
|
|
|
|
/// <summary>
|
|
/// Describes how a subscription transfer scenario is set up.
|
|
/// </summary>
|
|
public enum TransferType
|
|
{
|
|
/// <summary>
|
|
/// The origin session remains open and gives up the subscription.
|
|
/// </summary>
|
|
KeepOpen,
|
|
|
|
/// <summary>
|
|
/// The origin session is gracefully closed with
|
|
/// DeleteSubscriptionsOnClose set to false.
|
|
/// </summary>
|
|
CloseSession,
|
|
|
|
/// <summary>
|
|
/// The origin session gets network disconnected; available sequence
|
|
/// numbers are just acknowledged.
|
|
/// </summary>
|
|
DisconnectedAck,
|
|
|
|
/// <summary>
|
|
/// The origin session gets network disconnected; available sequence
|
|
/// numbers are republished.
|
|
/// </summary>
|
|
DisconnectedRepublish,
|
|
|
|
/// <summary>
|
|
/// The origin session gets network disconnected; available sequence
|
|
/// numbers are republished but the target delays the acknowledgement.
|
|
/// </summary>
|
|
DisconnectedRepublishDelayedAck
|
|
}
|
|
|
|
[Test]
|
|
public void TransferredSequencesRemainAvailableForRepublish()
|
|
{
|
|
using var subscription = new Subscription(Session.DefaultSubscription) { RepublishAfterTransfer = true };
|
|
ArrayOf<uint> available = new uint[] { 9, 10, 11 }.ToArrayOf();
|
|
MethodInfo processTransfer = typeof(Subscription).GetMethod(
|
|
"ProcessTransferredSequenceNumbers", BindingFlags.Instance | BindingFlags.NonPublic)!;
|
|
|
|
processTransfer.Invoke(subscription, [available]);
|
|
|
|
Assert.That(subscription.AvailableSequenceNumbers.ToArray(), Is.EquivalentTo(available.ToArray()!));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Initializes a new fixture instance for the given <paramref name="transferType"/>.
|
|
/// </summary>
|
|
public TransferSubscriptionTest(TransferType transferType)
|
|
: base(Utils.UriSchemeOpcTcp)
|
|
{
|
|
m_transferType = transferType;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Start one server per fixture instance and establish a single shared session.
|
|
/// Tests create their own sessions independently.
|
|
/// </summary>
|
|
[OneTimeSetUp]
|
|
public override Task OneTimeSetUpAsync()
|
|
{
|
|
// Use a shared session for namespace URI lookups; tests create their own sessions.
|
|
SingleSession = true;
|
|
MaxChannelCount = 1000;
|
|
ClientFixtureSubscriptionEngineFactory = ClassicSubscriptionEngineFactory.Instance;
|
|
return OneTimeSetUpCoreAsync(securityNone: true);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
[OneTimeTearDown]
|
|
public override Task OneTimeTearDownAsync()
|
|
{
|
|
return base.OneTimeTearDownAsync();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
[SetUp]
|
|
public override Task SetUpAsync()
|
|
{
|
|
return base.SetUpAsync();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
[TearDown]
|
|
public override Task TearDownAsync()
|
|
{
|
|
return base.TearDownAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Transfer subscriptions for all combinations of
|
|
/// <paramref name="sendInitialValues"/>, <paramref name="sequentialPublishing"/>,
|
|
/// and <paramref name="retainOriginNotifications"/>.
|
|
/// </summary>
|
|
[Test]
|
|
[Combinatorial]
|
|
[Parallelizable]
|
|
public Task TransferSubscriptionAsync(
|
|
[Values] bool sendInitialValues,
|
|
[Values] bool sequentialPublishing,
|
|
[Values] bool retainOriginNotifications)
|
|
{
|
|
return InternalTransferSubscriptionAsync(
|
|
m_transferType,
|
|
sendInitialValues,
|
|
sequentialPublishing,
|
|
retainOriginNotifications);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Debug variant that repeats the transfer test 30 times (explicit, not for normal CI runs).
|
|
/// </summary>
|
|
[Test]
|
|
[Combinatorial]
|
|
[Explicit]
|
|
public async Task TransferSubscriptionDebugAsync(
|
|
[Values] bool sendInitialValues,
|
|
[Values] bool sequentialPublishing,
|
|
[Values] bool retainOriginNotifications)
|
|
{
|
|
const int loopCount = 30;
|
|
for (int i = 0; i < loopCount; i++)
|
|
{
|
|
await InternalTransferSubscriptionAsync(
|
|
m_transferType,
|
|
sendInitialValues,
|
|
sequentialPublishing,
|
|
retainOriginNotifications).ConfigureAwait(false);
|
|
|
|
TestContext.Out.WriteLine("===========================================");
|
|
TestContext.Out.WriteLine("===========================================");
|
|
TestContext.Out.WriteLine($"Completed iteration {i + 1} of {loopCount}.");
|
|
TestContext.Out.WriteLine("===========================================");
|
|
TestContext.Out.WriteLine("===========================================");
|
|
}
|
|
}
|
|
|
|
private async Task InternalTransferSubscriptionAsync(
|
|
TransferType transferType,
|
|
bool sendInitialValues,
|
|
bool sequentialPublishing,
|
|
bool retainOriginNotifications)
|
|
{
|
|
using ISession originSession = await ClientFixture
|
|
.ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256)
|
|
.ConfigureAwait(false);
|
|
string filePath = Path.GetTempFileName();
|
|
try
|
|
{
|
|
await InternalTransferSubscriptionAsync(
|
|
originSession,
|
|
transferType,
|
|
sendInitialValues,
|
|
sequentialPublishing,
|
|
retainOriginNotifications,
|
|
filePath).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
File.Delete(filePath);
|
|
}
|
|
}
|
|
|
|
private async Task InternalTransferSubscriptionAsync(
|
|
ISession originSession,
|
|
TransferType transferType,
|
|
bool sendInitialValues,
|
|
bool sequentialPublishing,
|
|
bool retainOriginNotifications,
|
|
string filePath)
|
|
{
|
|
const int kTestSubscriptions = 5;
|
|
const int kDelay = 2_000;
|
|
const int kQueueSize = 10;
|
|
|
|
if (retainOriginNotifications)
|
|
{
|
|
originSession.PublishSequenceNumbersToAcknowledge += DeferSubscriptionAcknowledge;
|
|
}
|
|
|
|
bool originSessionOpen = transferType == TransferType.KeepOpen;
|
|
|
|
// create subscriptions
|
|
var originSubscriptions = new SubscriptionCollection(kTestSubscriptions);
|
|
int[] originSubscriptionCounters = new int[kTestSubscriptions];
|
|
int[] originSubscriptionFastDataCounters = new int[kTestSubscriptions];
|
|
int[] targetSubscriptionCounters = new int[kTestSubscriptions];
|
|
int[] targetSubscriptionFastDataCounters = new int[kTestSubscriptions];
|
|
int[] originSubscriptionTransferred = new int[kTestSubscriptions];
|
|
var staticTargetNotifications = new ConcurrentQueue<DataChangeNotification>();
|
|
int targetKeepAliveCount = 0;
|
|
bool expectRepublish = retainOriginNotifications &&
|
|
transferType >= TransferType.DisconnectedRepublish;
|
|
int expectedStaticBatchCount = (expectRepublish ? 1 : 0) + (sendInitialValues ? 1 : 0);
|
|
using var subscriptionTemplate = new TestableSubscription(originSession.DefaultSubscription)
|
|
{
|
|
PublishingInterval = 1_000,
|
|
LifetimeCount = 30,
|
|
KeepAliveCount = 1,
|
|
PublishingEnabled = false,
|
|
RepublishAfterTransfer = transferType >= TransferType.DisconnectedRepublish,
|
|
SequentialPublishing = sequentialPublishing
|
|
};
|
|
|
|
await CreateSubscriptionsAsync(
|
|
originSession,
|
|
subscriptionTemplate,
|
|
originSubscriptions,
|
|
originSubscriptionCounters,
|
|
originSubscriptionFastDataCounters,
|
|
kTestSubscriptions,
|
|
kQueueSize).ConfigureAwait(false);
|
|
|
|
if (TransferType.KeepOpen == transferType)
|
|
{
|
|
foreach (Subscription subscription in originSubscriptions)
|
|
{
|
|
subscription.PublishStatusChanged += (s, e) =>
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
$"PublishStatusChanged: {s.Session!.SessionId}-{s.Id}-{e.Status}");
|
|
if ((e.Status & PublishStateChangedMask.Transferred) != 0)
|
|
{
|
|
// subscription transferred
|
|
Interlocked.Increment(ref originSubscriptionTransferred[(int)s.Handle!]);
|
|
}
|
|
};
|
|
}
|
|
}
|
|
|
|
foreach (Subscription subscription in originSubscriptions)
|
|
{
|
|
await subscription.SetPublishingModeAsync(true).ConfigureAwait(false);
|
|
}
|
|
await TestPolling.WaitUntilAsync(
|
|
() => Enumerable.Range(0, kTestSubscriptions).All(index =>
|
|
Volatile.Read(ref originSubscriptionCounters[index]) >=
|
|
originSubscriptions[index].MonitoredItemCount &&
|
|
Volatile.Read(ref originSubscriptionFastDataCounters[index]) > 0),
|
|
timeoutMessage: "The origin did not deliver every initial monitored-item value.")
|
|
.ConfigureAwait(false);
|
|
|
|
Subscription staticOrigin = originSubscriptions[0];
|
|
if (!retainOriginNotifications)
|
|
{
|
|
await TestPolling.WaitUntilAsync(
|
|
() => staticOrigin.AvailableSequenceNumbers.IsEmpty,
|
|
timeoutMessage: "The origin's initial notification was not acknowledged by the server.")
|
|
.ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
Assert.That(
|
|
staticOrigin.AvailableSequenceNumbers.ToArray(),
|
|
Is.EqualTo([staticOrigin.SequenceNumber]));
|
|
}
|
|
NotificationMessage originStaticMessage = staticOrigin.Notifications.Single();
|
|
Assert.That(originStaticMessage.NotificationData.Count, Is.EqualTo(1));
|
|
Assert.That(
|
|
originStaticMessage.NotificationData[0].TryGetValue(out DataChangeNotification? originStaticData),
|
|
Is.True);
|
|
Assert.That(originStaticData!.MonitoredItems.Count, Is.EqualTo(staticOrigin.MonitoredItemCount));
|
|
|
|
// close session, do not delete subscription
|
|
if (transferType != TransferType.KeepOpen)
|
|
{
|
|
originSession.DeleteSubscriptionsOnClose = false;
|
|
|
|
// save with custom Subscription subclass information
|
|
originSession.Save(filePath);
|
|
|
|
if (transferType == TransferType.CloseSession)
|
|
{
|
|
// graceful close
|
|
StatusCode close = await originSession.CloseAsync().ConfigureAwait(false);
|
|
Assert.That(ServiceResult.IsGood(close), Is.True);
|
|
}
|
|
else
|
|
{
|
|
// force a socket dispose, to emulate network disconnect
|
|
// without closing session on server
|
|
originSession.TransportChannel.Dispose();
|
|
}
|
|
}
|
|
|
|
// wait
|
|
await Task.Delay(kDelay).ConfigureAwait(false);
|
|
|
|
// close session, do not delete subscription
|
|
if (transferType > TransferType.CloseSession)
|
|
{
|
|
StatusCode closeResult2 = await originSession
|
|
.CloseAsync()
|
|
.ConfigureAwait(false);
|
|
}
|
|
|
|
// create target session
|
|
using ISession targetSession = await ClientFixture
|
|
.ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256)
|
|
.ConfigureAwait(false);
|
|
if (transferType == TransferType.DisconnectedRepublishDelayedAck)
|
|
{
|
|
targetSession.PublishSequenceNumbersToAcknowledge += DeferSubscriptionAcknowledge;
|
|
}
|
|
|
|
// restore client state
|
|
var transferSubscriptions = new SubscriptionCollection();
|
|
if (transferType != TransferType.KeepOpen)
|
|
{
|
|
// load subscriptions for transfer
|
|
transferSubscriptions.AddRange(targetSession.Load(filePath, true));
|
|
}
|
|
else
|
|
{
|
|
transferSubscriptions.AddRange((SubscriptionCollection)originSubscriptions.Clone());
|
|
}
|
|
|
|
for (int ii = 0; ii < transferSubscriptions.Count; ii++)
|
|
{
|
|
Subscription subscription = transferSubscriptions[ii];
|
|
if (originSessionOpen)
|
|
{
|
|
Assert.That(targetSession.AddSubscription(subscription), Is.True);
|
|
}
|
|
subscription.Handle = ii;
|
|
subscription.FastDataChangeCallback = (s, notification, _) =>
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"FastDataChangeHandlerTarget: {0}-{1}-{2}",
|
|
s.Id,
|
|
notification.SequenceNumber,
|
|
notification.MonitoredItems.Count);
|
|
int index = (int)s.Handle!;
|
|
Interlocked.Increment(ref targetSubscriptionFastDataCounters[index]);
|
|
if (index == 0)
|
|
{
|
|
staticTargetNotifications.Enqueue(notification);
|
|
}
|
|
};
|
|
if (ii == 0)
|
|
{
|
|
subscription.FastKeepAliveCallback = (_, _) => Interlocked.Increment(ref targetKeepAliveCount);
|
|
}
|
|
foreach (MonitoredItem monitoredItem in subscription.MonitoredItems)
|
|
{
|
|
monitoredItem.Notification += (item, _) =>
|
|
{
|
|
Interlocked.Increment(ref targetSubscriptionCounters[(int)subscription.Handle]);
|
|
foreach (DataValue value in item.DequeueValues())
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"Tra:{0}: {1:20}, {2}, {3}, {4}",
|
|
subscription.Id,
|
|
item.DisplayName,
|
|
value.WrappedValue,
|
|
value.SourceTimestamp,
|
|
value.StatusCode);
|
|
}
|
|
};
|
|
}
|
|
subscription.StateChanged += (s, e) =>
|
|
TestContext.Out.WriteLine($"StateChanged: {s.Session!.SessionId}-{s.Id}-{e.Status}");
|
|
subscription.PublishStatusChanged += (s, e) =>
|
|
TestContext.Out.WriteLine($"PublishStatusChanged: {s.Session!.SessionId}-{s.Id}-{e.Status}");
|
|
}
|
|
|
|
// transfer restored subscriptions
|
|
bool result = await targetSession
|
|
.TransferSubscriptionsAsync(transferSubscriptions, sendInitialValues)
|
|
.ConfigureAwait(false);
|
|
Assert.That(result, Is.True);
|
|
|
|
// validate results
|
|
for (int ii = 0; ii < transferSubscriptions.Count; ii++)
|
|
{
|
|
Assert.That(transferSubscriptions[ii].Created, Is.True);
|
|
}
|
|
|
|
TestContext.Out
|
|
.WriteLine("TargetSession is now SessionId={0}", targetSession.SessionId);
|
|
|
|
await TestPolling.WaitUntilAsync(
|
|
() => staticTargetNotifications.Count >= expectedStaticBatchCount &&
|
|
Volatile.Read(ref targetKeepAliveCount) > 0 &&
|
|
Enumerable.Range(1, kTestSubscriptions - 1).All(index =>
|
|
Volatile.Read(ref targetSubscriptionCounters[index]) > 0 &&
|
|
Volatile.Read(ref targetSubscriptionFastDataCounters[index]) > 0),
|
|
timeoutMessage: "The target did not deliver expected notifications and a keepalive.")
|
|
.ConfigureAwait(false);
|
|
|
|
if (TransferType.KeepOpen == transferType)
|
|
{
|
|
await TestPolling.WaitUntilAsync(
|
|
() => Enumerable.Range(0, kTestSubscriptions).All(index =>
|
|
Volatile.Read(ref originSubscriptionTransferred[index]) > 0),
|
|
timeoutMessage: "The origin did not receive every subscription-transferred notification.")
|
|
.ConfigureAwait(false);
|
|
foreach (Subscription subscription in originSubscriptions)
|
|
{
|
|
Assert.That(originSubscriptionTransferred[(int)subscription.Handle!], Is.EqualTo(1));
|
|
}
|
|
}
|
|
|
|
// stop publishing
|
|
foreach (Subscription subscription in transferSubscriptions)
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"SetPublishingMode(false) for SessionId={0}, SubscriptionId={1}",
|
|
subscription.Session!.SessionId,
|
|
subscription.Id);
|
|
await subscription.SetPublishingModeAsync(false).ConfigureAwait(false);
|
|
}
|
|
|
|
// validate expected counts
|
|
for (int jj = 0; jj < kTestSubscriptions; jj++)
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"-- Subscription {0}: OriginCounts {1}, TargetCounts {2} ",
|
|
jj,
|
|
originSubscriptionCounters[jj],
|
|
targetSubscriptionCounters[jj]);
|
|
TestContext.Out.WriteLine(
|
|
"-- Subscription {0}: OriginFastDataCounts {1}, TargetFastDataCounts {2} ",
|
|
jj,
|
|
originSubscriptionFastDataCounters[jj],
|
|
targetSubscriptionFastDataCounters[jj]);
|
|
uint monitoredItemCount = transferSubscriptions[jj].MonitoredItemCount;
|
|
uint originExpectedCount = monitoredItemCount;
|
|
uint targetExpectedCount = sendInitialValues ? monitoredItemCount : 0;
|
|
if (jj == 0)
|
|
{
|
|
targetExpectedCount = (uint)expectedStaticBatchCount * monitoredItemCount;
|
|
Assert.That(originSubscriptionCounters[jj], Is.EqualTo(originExpectedCount));
|
|
Assert.That(originSubscriptionFastDataCounters[jj], Is.EqualTo(1));
|
|
Assert.That(targetSubscriptionCounters[jj], Is.EqualTo(targetExpectedCount));
|
|
Assert.That(targetSubscriptionFastDataCounters[jj], Is.EqualTo(expectedStaticBatchCount));
|
|
AssertStaticNotifications(
|
|
staticTargetNotifications, originStaticData, expectRepublish, sendInitialValues);
|
|
}
|
|
else
|
|
{
|
|
// dynamic nodes, expect only one set of changes, another one if send initial values was set
|
|
Assert.That(originSubscriptionCounters[jj], Is.GreaterThanOrEqualTo(originExpectedCount));
|
|
Assert.That(targetSubscriptionCounters[jj], Is.GreaterThanOrEqualTo(targetExpectedCount));
|
|
}
|
|
}
|
|
|
|
int[] countsBeforeResume =
|
|
[
|
|
.. Enumerable.Range(0, kTestSubscriptions)
|
|
.Select(index => Volatile.Read(ref targetSubscriptionCounters[index]))
|
|
];
|
|
int[] batchesBeforeResume =
|
|
[
|
|
.. Enumerable.Range(0, kTestSubscriptions)
|
|
.Select(index => Volatile.Read(ref targetSubscriptionFastDataCounters[index]))
|
|
];
|
|
int keepAlivesBeforeResume = Volatile.Read(ref targetKeepAliveCount);
|
|
|
|
// restart publishing
|
|
foreach (Subscription subscription in transferSubscriptions)
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"SetPublishingMode(true) for SessionId={0}, SubscriptionId={1}",
|
|
subscription.Session!.SessionId,
|
|
subscription.Id);
|
|
await subscription.SetPublishingModeAsync(true).ConfigureAwait(false);
|
|
}
|
|
|
|
await TestPolling.WaitUntilAsync(
|
|
() => Volatile.Read(ref targetKeepAliveCount) > keepAlivesBeforeResume &&
|
|
Enumerable.Range(1, kTestSubscriptions - 1).All(index =>
|
|
Volatile.Read(ref targetSubscriptionCounters[index]) > countsBeforeResume[index] &&
|
|
Volatile.Read(ref targetSubscriptionFastDataCounters[index]) > batchesBeforeResume[index]),
|
|
timeoutMessage: "Transferred subscriptions did not resume dynamic values and static keepalives.")
|
|
.ConfigureAwait(false);
|
|
|
|
// validate expected counts
|
|
for (int jj = 0; jj < kTestSubscriptions; jj++)
|
|
{
|
|
TestContext.Out.WriteLine(
|
|
"-- Subscription {0}: OriginCounts {1}, TargetCounts {2} ",
|
|
jj,
|
|
originSubscriptionCounters[jj],
|
|
targetSubscriptionCounters[jj]);
|
|
TestContext.Out.WriteLine(
|
|
"-- Subscription {0}: OriginFastDataCounts {1}, TargetFastDataCounts {2} ",
|
|
jj,
|
|
originSubscriptionFastDataCounters[jj],
|
|
targetSubscriptionFastDataCounters[jj]);
|
|
|
|
if (jj == 0)
|
|
{
|
|
Assert.That(targetSubscriptionCounters[jj], Is.EqualTo(countsBeforeResume[jj]));
|
|
Assert.That(targetSubscriptionFastDataCounters[jj], Is.EqualTo(batchesBeforeResume[jj]));
|
|
AssertStaticNotifications(
|
|
staticTargetNotifications, originStaticData, expectRepublish, sendInitialValues);
|
|
}
|
|
else
|
|
{
|
|
Assert.That(targetSubscriptionCounters[jj], Is.GreaterThan(countsBeforeResume[jj]));
|
|
Assert.That(targetSubscriptionFastDataCounters[jj], Is.GreaterThan(batchesBeforeResume[jj]));
|
|
}
|
|
}
|
|
|
|
targetSession.DeleteSubscriptionsOnClose = true;
|
|
|
|
// close sessions
|
|
StatusCode closeResult = await targetSession.CloseAsync().ConfigureAwait(false);
|
|
Assert.That(ServiceResult.IsGood(closeResult), Is.True);
|
|
|
|
if (originSessionOpen)
|
|
{
|
|
closeResult = await originSession.CloseAsync().ConfigureAwait(false);
|
|
Assert.That(ServiceResult.IsGood(closeResult), Is.True);
|
|
}
|
|
}
|
|
|
|
private static void AssertStaticNotifications(
|
|
ConcurrentQueue<DataChangeNotification> received,
|
|
DataChangeNotification origin,
|
|
bool expectRepublish,
|
|
bool sendInitialValues)
|
|
{
|
|
DataChangeNotification[] notifications = [.. received];
|
|
Assert.That(
|
|
notifications.Count(notification => notification.SequenceNumber == origin.SequenceNumber),
|
|
Is.EqualTo(expectRepublish ? 1 : 0),
|
|
"Only a retained origin message requested for republish should reuse its sequence number.");
|
|
Assert.That(
|
|
notifications.Count(notification => notification.SequenceNumber != origin.SequenceNumber),
|
|
Is.EqualTo(sendInitialValues ? 1 : 0),
|
|
"Requested initial values must arrive in one new notification message.");
|
|
Assert.That(notifications.Select(notification => notification.SequenceNumber), Is.Unique);
|
|
|
|
Dictionary<uint, MonitoredItemNotification> originalItems =
|
|
origin.MonitoredItems.ToArray()!.ToDictionary(item => item.ClientHandle);
|
|
foreach (DataChangeNotification notification in notifications)
|
|
{
|
|
Assert.That(notification.SequenceNumber, Is.GreaterThanOrEqualTo(origin.SequenceNumber));
|
|
Assert.That(
|
|
notification.MonitoredItems.ToArray()!.Select(item => item.ClientHandle),
|
|
Is.EquivalentTo(originalItems.Keys));
|
|
foreach (MonitoredItemNotification item in notification.MonitoredItems)
|
|
{
|
|
Assert.That(
|
|
originalItems.TryGetValue(item.ClientHandle, out MonitoredItemNotification? original),
|
|
Is.True);
|
|
Assert.That(item.Value.WrappedValue, Is.EqualTo(original!.Value.WrappedValue));
|
|
Assert.That(item.Value.StatusCode, Is.EqualTo(original.Value.StatusCode));
|
|
Assert.That(item.Value.SourceTimestamp, Is.EqualTo(original.Value.SourceTimestamp));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Event handler to defer publish response sequence number acknowledge.
|
|
/// </summary>
|
|
private static void DeferSubscriptionAcknowledge(
|
|
ISession session,
|
|
PublishSequenceNumbersToAcknowledgeEventArgs e)
|
|
{
|
|
// for testing do not ack any sequence numbers
|
|
e.DeferredAcknowledgementsToSend.Clear();
|
|
e.AcknowledgementsToSend.Clear();
|
|
}
|
|
|
|
private readonly TransferType m_transferType;
|
|
}
|
|
}
|