diff --git a/Stack/Opc.Ua.Bindings.Https/Stack/Https/HttpsTransportListener.cs b/Stack/Opc.Ua.Bindings.Https/Stack/Https/HttpsTransportListener.cs
index 1b97284911..4279a024b4 100644
--- a/Stack/Opc.Ua.Bindings.Https/Stack/Https/HttpsTransportListener.cs
+++ b/Stack/Opc.Ua.Bindings.Https/Stack/Https/HttpsTransportListener.cs
@@ -39,6 +39,7 @@
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Server.Kestrel.Https;
+using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Opc.Ua.Security.Certificates;
using Microsoft.Extensions.Logging;
@@ -99,16 +100,19 @@ public class Startup
{
private const string kHttpsContentType = "text/plain";
- ///
- /// Get the Https listener.
- ///
- public static HttpsTransportListener Listener { get; set; }
-
///
/// Configure the request pipeline for the listener.
+ /// The is resolved from the web host's DI container,
+ /// which is populated via IWebHostBuilder.ConfigureServices in
+ /// .
+ /// Using method injection here (rather than constructor injection) ensures the
+ /// instance is resolved from the web-host app container on all target frameworks,
+ /// including .NET 8+ where HostBuilder activates Startup from a
+ /// separate host-level container that does not include web-host app services.
///
/// The application builder.
- public void Configure(IApplicationBuilder appBuilder)
+ /// The transport listener that handles OPC UA requests.
+ public void Configure(IApplicationBuilder appBuilder, HttpsTransportListener listener)
{
appBuilder.Run(context =>
{
@@ -120,7 +124,7 @@ public void Configure(IApplicationBuilder appBuilder)
return context.Response.WriteAsync(string.Empty);
}
- return Listener.SendAsync(context);
+ return listener.SendAsync(context);
});
}
}
@@ -272,7 +276,6 @@ public void UpdateChannelLastActiveTime(string globalChannelId)
///
public void Start()
{
- Startup.Listener = this;
#if NET8_0_OR_GREATER
m_host = new HostBuilder()
.ConfigureWebHostDefaults(ConfigureWebHost)
@@ -285,6 +288,11 @@ public void Start()
#endif
}
+ // CA1859: The IWebHostBuilder interface cannot be narrowed to WebHostBuilder here because
+ // on NET8_0_OR_GREATER this method is called from HostBuilder.ConfigureWebHostDefaults()
+ // which passes an IWebHostBuilder, not WebHostBuilder.
+ [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1859:Use concrete types when possible for improved performance",
+ Justification = "IWebHostBuilder is required for cross-framework compatibility; on NET8_0_OR_GREATER ConfigureWebHostDefaults passes IWebHostBuilder.")]
private void ConfigureWebHost(IWebHostBuilder webHostBuilder)
{
// prepare the server TLS certificate
@@ -338,6 +346,9 @@ private void ConfigureWebHost(IWebHostBuilder webHostBuilder)
}
webHostBuilder.UseContentRoot(Directory.GetCurrentDirectory());
+ // Register this listener instance in the web host DI container so it can be
+ // injected into Startup.Configure as a method parameter — no static state needed.
+ webHostBuilder.ConfigureServices(services => services.AddSingleton(this));
webHostBuilder.UseStartup();
}
diff --git a/Tests/Opc.Ua.Client.Tests/ClientBatchTest.cs b/Tests/Opc.Ua.Client.Tests/ClientBatchTest.cs
index fb14ad61f3..3584e90bb1 100644
--- a/Tests/Opc.Ua.Client.Tests/ClientBatchTest.cs
+++ b/Tests/Opc.Ua.Client.Tests/ClientBatchTest.cs
@@ -49,6 +49,7 @@ namespace Opc.Ua.Client.Tests
[SetCulture("en-us")]
[SetUICulture("en-us")]
[TestFixtureSource(nameof(FixtureArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
[MemoryDiagnoser]
[DisassemblyDiagnoser]
public class ClientBatchTest : ClientTestFramework
diff --git a/Tests/Opc.Ua.Client.Tests/ClientFixture.cs b/Tests/Opc.Ua.Client.Tests/ClientFixture.cs
index 8adddb6e96..6b7649f7a6 100644
--- a/Tests/Opc.Ua.Client.Tests/ClientFixture.cs
+++ b/Tests/Opc.Ua.Client.Tests/ClientFixture.cs
@@ -174,6 +174,7 @@ public async Task StartReverseConnectHostAsync()
int serverStartRetries = 25;
do
{
+ retryStartServer = false;
try
{
var reverseConnectUri = new Uri("opc.tcp://localhost:" + testPort);
@@ -188,11 +189,13 @@ public async Task StartReverseConnectHostAsync()
{
throw;
}
+
testPort = UnsecureRandom.Shared.Next(
ServerFixtureUtils.MinTestPort,
ServerFixtureUtils.MaxTestPort);
retryStartServer = true;
}
+
await Task.Delay(UnsecureRandom.Shared.Next(100, 1000)).ConfigureAwait(false);
} while (retryStartServer);
}
@@ -219,7 +222,7 @@ public async Task ConnectAsync(string endpointUrl, CancellationToken c
nameof(endpointUrl));
}
- const int maxAttempts = 5;
+ const int maxAttempts = 25;
for (int attempt = 0; ; attempt++)
{
try
@@ -274,7 +277,7 @@ public async Task ConnectAsync(
getEndpointsUrl = CoreClientUtils.GetDiscoveryUrl(uri);
}
- const int maxAttempts = 5;
+ const int maxAttempts = 25;
for (int attempt = 0; ; attempt++)
{
try
@@ -297,6 +300,14 @@ public async Task ConnectAsync(
m_logger.LogError(e, "Failed to connect {Attempt}. Retrying in 1 second...", attempt + 1);
await Task.Delay(1000).ConfigureAwait(false);
}
+ catch (Exception e) when (
+ e is not IgnoreException &&
+ attempt < maxAttempts)
+ {
+ // Retry on transient errors (e.g. HttpRequestException when HTTPS server is not yet ready)
+ m_logger.LogError(e, "Failed to connect {Attempt}. Retrying in 1 second...", attempt + 1);
+ await Task.Delay(1000).ConfigureAwait(false);
+ }
}
throw new ServiceResultException(StatusCodes.BadNoCommunication);
diff --git a/Tests/Opc.Ua.Client.Tests/ClientTest.cs b/Tests/Opc.Ua.Client.Tests/ClientTest.cs
index 1c32624406..7eed44d534 100644
--- a/Tests/Opc.Ua.Client.Tests/ClientTest.cs
+++ b/Tests/Opc.Ua.Client.Tests/ClientTest.cs
@@ -58,6 +58,7 @@ namespace Opc.Ua.Client.Tests
[SetCulture("en-us")]
[SetUICulture("en-us")]
[TestFixtureSource(nameof(FixtureArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
[MemoryDiagnoser]
[DisassemblyDiagnoser]
public class ClientTest : ClientTestFramework
diff --git a/Tests/Opc.Ua.Client.Tests/ClientTestFramework.cs b/Tests/Opc.Ua.Client.Tests/ClientTestFramework.cs
index a857c131c7..3b202eadff 100644
--- a/Tests/Opc.Ua.Client.Tests/ClientTestFramework.cs
+++ b/Tests/Opc.Ua.Client.Tests/ClientTestFramework.cs
@@ -627,5 +627,101 @@ protected void SessionClosing(object sender, EventArgs e)
TestContext.Out.WriteLine("Session_Closing: {0}", session.SessionId);
}
}
+
+ ///
+ /// Create a set of subscriptions with monitored items on the given session.
+ ///
+ protected async Task CreateSubscriptionsAsync(
+ ISession session,
+ Subscription template,
+ SubscriptionCollection originSubscriptions,
+ int[] notificationCounters,
+ int[] fastDataCounters,
+ int subscriptionCount,
+ uint queueSize)
+ {
+ for (int ii = 0; ii < subscriptionCount; ii++)
+ {
+ // create subscription with static monitored items
+ var subscription = new TestableSubscription(template)
+ {
+ PublishingEnabled = true,
+ Handle = ii,
+ FastDataChangeCallback = (s, n, _) =>
+ {
+ TestContext.Out.WriteLine(
+ $"FastDataChangeHandlerOrigin: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
+ fastDataCounters[(int)s.Handle]++;
+ }
+ };
+
+ 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}");
+
+ originSubscriptions.Add(subscription);
+ session.AddSubscription(subscription);
+ await subscription.CreateAsync().ConfigureAwait(false);
+
+ // set defaults
+ subscription.DefaultItem.DiscardOldest = true;
+ subscription.DefaultItem.QueueSize = ii == 0 ? 0U : queueSize;
+ subscription.DefaultItem.MonitoringMode = MonitoringMode.Reporting;
+
+ // create test set using the passed session's namespace URIs
+ NamespaceTable namespaceUris = session.NamespaceUris;
+ var testSet = new List();
+ if (ii == 0)
+ {
+ testSet.AddRange(GetTestSetStatic(namespaceUris));
+ }
+ else
+ {
+ testSet.AddRange(GetTestSetSimulation(namespaceUris));
+ }
+
+ var list = CreateMonitoredItemTestSet(subscription, testSet).ToList();
+ list.ForEach(i =>
+ i.Notification += (item, _) =>
+ {
+ notificationCounters[(int)subscription.Handle]++;
+ foreach (DataValue value in item.DequeueValues())
+ {
+ TestContext.Out.WriteLine(
+ "Org:{0}: {1:20}, {2}, {3}, {4}",
+ subscription.Id,
+ item.DisplayName,
+ value.WrappedValue,
+ value.SourceTimestamp,
+ value.StatusCode);
+ }
+ });
+ subscription.AddItems(list);
+ await subscription.ApplyChangesAsync().ConfigureAwait(false);
+ }
+ }
+
+ ///
+ /// Create monitored items from a list of node ids on a subscription.
+ ///
+ protected static List CreateMonitoredItemTestSet(
+ Subscription subscription,
+ IList nodeIds)
+ {
+ var list = new List();
+ foreach (NodeId nodeId in nodeIds)
+ {
+ var item = new TestableMonitoredItem(subscription.DefaultItem)
+ {
+ StartNodeId = nodeId
+ };
+ list.Add(item);
+ }
+ return list;
+ }
}
}
diff --git a/Tests/Opc.Ua.Client.Tests/LoadTest.cs b/Tests/Opc.Ua.Client.Tests/LoadTest.cs
index b472be7bb8..97daa8acee 100644
--- a/Tests/Opc.Ua.Client.Tests/LoadTest.cs
+++ b/Tests/Opc.Ua.Client.Tests/LoadTest.cs
@@ -47,6 +47,7 @@ namespace Opc.Ua.Client.Tests
[SetCulture("en-us")]
[SetUICulture("en-us")]
[TestFixtureSource(nameof(FixtureArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
public class LoadTest : ClientTestFramework
{
public LoadTest(string uriScheme)
diff --git a/Tests/Opc.Ua.Client.Tests/NodeCacheAsyncTest.cs b/Tests/Opc.Ua.Client.Tests/NodeCacheAsyncTest.cs
index e824ab8685..7fc07156ea 100644
--- a/Tests/Opc.Ua.Client.Tests/NodeCacheAsyncTest.cs
+++ b/Tests/Opc.Ua.Client.Tests/NodeCacheAsyncTest.cs
@@ -46,6 +46,7 @@ namespace Opc.Ua.Client.Tests
[SetCulture("en-us")]
[SetUICulture("en-us")]
[TestFixtureSource(nameof(AsyncFixtureArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
[MemoryDiagnoser]
[DisassemblyDiagnoser]
public class NodeCacheAsyncTest : ClientTestFramework
diff --git a/Tests/Opc.Ua.Client.Tests/ReconnectWithSavedSessionSecretsTest.cs b/Tests/Opc.Ua.Client.Tests/ReconnectWithSavedSessionSecretsTest.cs
new file mode 100644
index 0000000000..3c1e27212f
--- /dev/null
+++ b/Tests/Opc.Ua.Client.Tests/ReconnectWithSavedSessionSecretsTest.cs
@@ -0,0 +1,405 @@
+/* ========================================================================
+ * 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.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace Opc.Ua.Client.Tests
+{
+ ///
+ /// Tests that reconnect a session on a new channel using saved session secrets.
+ /// Each fixture instance handles one security policy. The eight boolean combinations of
+ /// anonymous × sequentialPublishing × sendInitialValues run in parallel within each
+ /// fixture instance, while the fixture instances themselves also run in parallel.
+ ///
+ [TestFixture]
+ [Category("Client")]
+ [Category("ReconnectWithSavedSessionSecrets")]
+ [SetCulture("en-us")]
+ [SetUICulture("en-us")]
+ [TestFixtureSource(nameof(SecurityPolicyArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
+ public class ReconnectWithSavedSessionSecretsTest : ClientTestFramework
+ {
+ ///
+ /// Fixture argument list: one fixture instance per security policy.
+ ///
+ public static readonly object[] SecurityPolicyArgs =
+ [
+ new object[] { SecurityPolicies.None },
+ new object[] { SecurityPolicies.ECC_nistP256 },
+ new object[] { SecurityPolicies.Basic256Sha256 }
+ ];
+
+ private readonly string m_securityPolicy;
+
+ ///
+ /// Initializes a new fixture instance for the given .
+ ///
+ public ReconnectWithSavedSessionSecretsTest(string securityPolicy)
+ : base(Utils.UriSchemeOpcTcp)
+ {
+ m_securityPolicy = securityPolicy;
+ }
+
+ ///
+ /// Start one server per fixture instance and configure a TestableSessionFactory.
+ /// Tests create their own sessions independently.
+ ///
+ [OneTimeSetUp]
+ public override async Task OneTimeSetUpAsync()
+ {
+ // Use a shared session for namespace URI lookups; tests create their own sessions.
+ SingleSession = true;
+ MaxChannelCount = 1000;
+ await OneTimeSetUpCoreAsync(securityNone: true).ConfigureAwait(false);
+
+ // Set the session factory once; all parallel tests within this fixture use it.
+ ClientFixture.SessionFactory = new TestableSessionFactory(Telemetry);
+ }
+
+ ///
+ [OneTimeTearDown]
+ public override Task OneTimeTearDownAsync()
+ {
+ return base.OneTimeTearDownAsync();
+ }
+
+ ///
+ [SetUp]
+ public override Task SetUpAsync()
+ {
+ return base.SetUpAsync();
+ }
+
+ ///
+ [TearDown]
+ public override Task TearDownAsync()
+ {
+ return base.TearDownAsync();
+ }
+
+ ///
+ /// Reconnect a session using saved session secrets for all combinations of
+ /// , , and
+ /// . Each combination runs in parallel with the
+ /// others within this fixture and also across the three security-policy instances.
+ ///
+ [Test]
+ [Combinatorial]
+ [Order(352)]
+ [Parallelizable]
+ public Task ReconnectWithSavedSessionSecretsAsync(
+ [Values] bool anonymous,
+ [Values] bool sequentialPublishing,
+ [Values] bool sendInitialValues)
+ {
+ return ReconnectWithSavedSessionSecretsCoreAsync(
+ m_securityPolicy,
+ anonymous,
+ sequentialPublishing,
+ sendInitialValues);
+ }
+
+ ///
+ /// ECC variant of the reconnect test (explicit — not run in normal CI).
+ ///
+ [Test]
+ [Combinatorial]
+ [Order(351)]
+ [Explicit]
+ [Parallelizable]
+ public Task ReconnectWithSavedSessionSecretsECCAsync(
+ [Values(
+ SecurityPolicies.ECC_nistP256,
+ SecurityPolicies.ECC_nistP384,
+ SecurityPolicies.ECC_brainpoolP256r1,
+ SecurityPolicies.ECC_brainpoolP384r1
+ )]
+ string securityPolicy,
+ [Values] bool anonymous,
+ [Values] bool sequentialPublishing,
+ [Values] bool sendInitialValues)
+ {
+ return ReconnectWithSavedSessionSecretsCoreAsync(
+ securityPolicy,
+ anonymous,
+ sequentialPublishing,
+ sendInitialValues);
+ }
+
+ private async Task ReconnectWithSavedSessionSecretsCoreAsync(
+ string securityPolicy,
+ bool anonymous,
+ bool sequentialPublishing,
+ bool sendInitialValues)
+ {
+ const int kTestSubscriptions = 5;
+ const int kDelay = 2_000;
+ const int kQueueSize = 10;
+
+ ServiceResultException sre;
+ UserIdentity userIdentity = anonymous
+ ? new UserIdentity()
+ : new UserIdentity("user1", "password"u8);
+
+ // the first channel determines the endpoint
+ ConfiguredEndpoint endpoint = await ClientFixture
+ .GetEndpointAsync(ServerUrl, securityPolicy, Endpoints)
+ .ConfigureAwait(false);
+ Assert.That(endpoint, Is.Not.Null);
+
+ UserTokenPolicy identityPolicy = endpoint.Description.FindUserTokenPolicy(
+ userIdentity.TokenType,
+ userIdentity.IssuedTokenType,
+ endpoint.Description.SecurityPolicyUri);
+ if (identityPolicy == null)
+ {
+ Assert.Ignore(
+ $"No UserTokenPolicy found for {userIdentity.TokenType} / {userIdentity.IssuedTokenType}");
+ }
+
+ // the active channel
+ ISession session1 = await ClientFixture.ConnectAsync(endpoint, userIdentity)
+ .ConfigureAwait(false);
+ Assert.That(session1, Is.Not.Null);
+ NodeId sessionId1 = session1.SessionId;
+
+ int session1ConfigChanged = 0;
+ session1.SessionConfigurationChanged += (sender, e) => session1ConfigChanged++;
+
+ ServerStatusDataType value1 = await session1.ReadValueAsync(
+ VariableIds.Server_ServerStatus).ConfigureAwait(false);
+ Assert.That(value1, Is.Not.Null);
+
+ 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];
+ var subscriptionTemplate = new TestableSubscription(session1.DefaultSubscription)
+ {
+ PublishingInterval = 1_000,
+ KeepAliveCount = 5,
+ PublishingEnabled = true,
+ RepublishAfterTransfer = true,
+ SequentialPublishing = sequentialPublishing
+ };
+
+ await CreateSubscriptionsAsync(
+ session1,
+ subscriptionTemplate,
+ originSubscriptions,
+ originSubscriptionCounters,
+ originSubscriptionFastDataCounters,
+ kTestSubscriptions,
+ kQueueSize).ConfigureAwait(false);
+
+ // wait
+ await Task.Delay(kDelay).ConfigureAwait(false);
+
+ // save the session configuration
+ var configStream = new MemoryStream();
+ session1.SaveSessionConfiguration(configStream);
+
+ byte[] configStreamArray = configStream.ToArray();
+ TestContext.Out.WriteLine($"SessionSecrets: {configStream.Length} bytes");
+ TestContext.Out.WriteLine(Encoding.UTF8.GetString(configStreamArray));
+
+ var subscriptionStream = new MemoryStream();
+ session1.Save(
+ subscriptionStream,
+ session1.Subscriptions);
+
+ byte[] subscriptionStreamArray = subscriptionStream.ToArray();
+ TestContext.Out.WriteLine($"Subscriptions: {subscriptionStreamArray.Length} bytes");
+ TestContext.Out.WriteLine(Encoding.UTF8.GetString(subscriptionStreamArray));
+
+ // read the session configuration
+ var loadConfigurationStream = new MemoryStream(configStreamArray);
+ var sessionConfiguration = SessionConfiguration.Create(
+ loadConfigurationStream,
+ Telemetry);
+
+ // create the inactive channel
+ ITransportChannel channel2 = await ClientFixture
+ .CreateChannelAsync(sessionConfiguration.ConfiguredEndpoint, false)
+ .ConfigureAwait(false);
+ Assert.That(channel2, Is.Not.Null);
+
+ // prepare the inactive session with the new channel
+ ISession session2 = ClientFixture.CreateSession(
+ channel2,
+ sessionConfiguration.ConfiguredEndpoint);
+
+ int session2ConfigChanged = 0;
+ session2.SessionConfigurationChanged += (sender, e) => session2ConfigChanged++;
+
+ // apply the saved session configuration
+ bool success = session2.ApplySessionConfiguration(sessionConfiguration);
+
+ // restore the subscriptions
+ var loadSubscriptionStream = new MemoryStream(subscriptionStreamArray);
+ var restoredSubscriptions = new SubscriptionCollection(
+ session2.Load(loadSubscriptionStream, true, [typeof(SubscriptionState)]));
+
+ // hook notifications for log output
+ int ii = 0;
+ foreach (Subscription subscription in restoredSubscriptions)
+ {
+ subscription.Handle = ii;
+ subscription.FastDataChangeCallback = (s, n, _) =>
+ {
+ TestContext.Out.WriteLine(
+ $"FastDataChangeHandlerTarget: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
+ targetSubscriptionFastDataCounters[(int)subscription.Handle]++;
+ };
+ subscription
+ .MonitoredItems.ToList()
+ .ForEach(i =>
+ i.Notification += (item, _) =>
+ {
+ 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);
+ }
+ });
+ ii++;
+ }
+
+ // hook callback to renew the user identity
+ session2.RenewUserIdentity += (_, _) => userIdentity;
+
+ // activate the session from saved session secrets on the new channel
+ await session2.ReconnectAsync(channel2).ConfigureAwait(false);
+
+ // reactivate restored subscriptions
+ bool reactivateResult = await session2
+ .ReactivateSubscriptionsAsync(restoredSubscriptions, sendInitialValues)
+ .ConfigureAwait(false);
+ Assert.That(reactivateResult, Is.True);
+
+ await Task.Delay(2 * kDelay).ConfigureAwait(false);
+
+ try
+ {
+ Assert.That(session2.SessionId, Is.EqualTo(sessionId1));
+
+ DataValue value2 = await session2
+ .ReadValueAsync(VariableIds.Server_ServerStatus)
+ .ConfigureAwait(false);
+ Assert.That(value2, Is.Not.Null);
+
+ for (ii = 0; ii < kTestSubscriptions; ii++)
+ {
+ uint monitoredItemCount = restoredSubscriptions[ii].MonitoredItemCount;
+ string errorText = $"Error in test subscription {ii}";
+
+ // the static subscription doesn't resend data until there is a data change
+ if (ii == 0 && !sendInitialValues)
+ {
+ Assert.That(targetSubscriptionCounters[ii], Is.Zero, errorText);
+ Assert.That(targetSubscriptionFastDataCounters[ii], Is.Zero, errorText);
+ }
+ else if (ii == 0)
+ {
+ Assert.That(
+ targetSubscriptionCounters[ii],
+ Is.EqualTo(monitoredItemCount),
+ errorText);
+ Assert.That(targetSubscriptionFastDataCounters[ii], Is.EqualTo(1), errorText);
+ }
+ else
+ {
+ Assert.That(
+ targetSubscriptionCounters[ii],
+ Is.GreaterThanOrEqualTo(monitoredItemCount),
+ errorText);
+ Assert.That(targetSubscriptionFastDataCounters[ii], Is.GreaterThanOrEqualTo(1), errorText);
+ }
+ }
+
+ await Task.Delay(kDelay).ConfigureAwait(false);
+
+ // verify that reconnect created subclassed version of subscription and monitored item
+ foreach (Subscription s in session2.Subscriptions)
+ {
+ Assert.That(s.GetType(), Is.EqualTo(typeof(TestableSubscription)));
+ foreach (MonitoredItem m in s.MonitoredItems)
+ {
+ Assert.That(m.GetType(), Is.EqualTo(typeof(TestableMonitoredItem)));
+ }
+ }
+
+ // cannot read using a closed channel, validate the status code
+ if (endpoint.EndpointUrl.ToString()
+ .StartsWith(Utils.UriSchemeOpcTcp, StringComparison.Ordinal))
+ {
+ sre = Assert.ThrowsAsync(() =>
+ session1.ReadValueAsync(
+ VariableIds.Server_ServerStatus));
+ Assert.That(
+ sre.StatusCode,
+ Is.EqualTo(StatusCodes.BadSecureChannelIdInvalid),
+ sre.Message);
+ }
+ else
+ {
+ ServerStatusDataType result =
+ await session1.ReadValueAsync(
+ VariableIds.Server_ServerStatus).ConfigureAwait(false);
+ Assert.That(result, Is.Not.Null);
+ }
+ }
+ finally
+ {
+ session1.DeleteSubscriptionsOnClose = true;
+ session2.DeleteSubscriptionsOnClose = true;
+ await session1.CloseAsync(1000, true).ConfigureAwait(false);
+ await session2.CloseAsync(1000, true).ConfigureAwait(false);
+ Utils.SilentDispose(session1);
+ Utils.SilentDispose(session2);
+ }
+
+ Assert.That(session1ConfigChanged, Is.Zero);
+ Assert.That(session2ConfigChanged, Is.GreaterThanOrEqualTo(0));
+ }
+ }
+}
diff --git a/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs b/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs
index e47d13d5c1..b78f7fcddb 100644
--- a/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs
+++ b/Tests/Opc.Ua.Client.Tests/SubscriptionTest.cs
@@ -33,7 +33,6 @@
using System.Diagnostics;
using System.IO;
using System.Linq;
-using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NUnit.Framework;
@@ -438,304 +437,6 @@ await Session.SetPublishingModeAsync(
Assert.That(!failed, Is.EqualTo(enabled));
}
- ///
- /// Open a session on a channel, then reconnect (activate)
- /// the same session on a new channel with saved session secrets.
- /// Use only asnc methods. Test the ECC profiles.
- ///
- [Test]
- [Combinatorial]
- [Order(351)]
- [Explicit]
- public Task ReconnectWithSavedSessionSecretsOnlyECCAsync(
- [Values(
- SecurityPolicies.ECC_nistP256,
- SecurityPolicies.ECC_nistP384,
- SecurityPolicies.ECC_brainpoolP256r1,
- SecurityPolicies.ECC_brainpoolP384r1
- )]
- string securityPolicy,
- [Values] bool anonymous,
- [Values] bool sequentialPublishing,
- [Values] bool sendInitialValues)
- {
- return ReconnectWithSavedSessionSecretsAsync(
- securityPolicy,
- anonymous,
- sequentialPublishing,
- sendInitialValues);
- }
-
- ///
- /// Open a session on a channel, then reconnect (activate)
- /// the same session on a new channel with saved session secrets.
- /// Use only asnc methods.
- ///
- [Test]
- [Combinatorial]
- [Order(352)]
- public Task ReconnectWithSavedSessionSecretsOnlyAsync(
- [Values(SecurityPolicies.None,
- SecurityPolicies.ECC_nistP256,
- SecurityPolicies.Basic256Sha256)]
- string securityPolicy,
- [Values] bool anonymous,
- [Values] bool sequentialPublishing,
- [Values] bool sendInitialValues)
- {
- return ReconnectWithSavedSessionSecretsAsync(
- securityPolicy,
- anonymous,
- sequentialPublishing,
- sendInitialValues);
- }
-
- private async Task ReconnectWithSavedSessionSecretsAsync(
- string securityPolicy,
- bool anonymous,
- bool sequentialPublishing,
- bool sendInitialValues)
- {
- const int kTestSubscriptions = 5;
- const int kDelay = 2_000;
- const int kQueueSize = 10;
-
- ServiceResultException sre;
- UserIdentity userIdentity = anonymous
- ? new UserIdentity()
- : new UserIdentity("user1", "password"u8);
-
- ClientFixture.SessionFactory = new TestableSessionFactory(Telemetry);
-
- // the first channel determines the endpoint
- ConfiguredEndpoint endpoint = await ClientFixture
- .GetEndpointAsync(ServerUrl, securityPolicy, Endpoints)
- .ConfigureAwait(false);
- Assert.That(endpoint, Is.Not.Null);
-
- UserTokenPolicy identityPolicy = endpoint.Description.FindUserTokenPolicy(
- userIdentity.TokenType,
- userIdentity.IssuedTokenType,
- endpoint.Description.SecurityPolicyUri);
- if (identityPolicy == null)
- {
- Assert.Ignore(
- $"No UserTokenPolicy found for {userIdentity.TokenType} / {userIdentity.IssuedTokenType}");
- }
-
- // the active channel
- ISession session1 = await ClientFixture.ConnectAsync(endpoint, userIdentity)
- .ConfigureAwait(false);
- Assert.That(session1, Is.Not.Null);
- NodeId sessionId1 = session1.SessionId;
-
- int session1ConfigChanged = 0;
- session1.SessionConfigurationChanged += (sender, e) => session1ConfigChanged++;
-
- ServerStatusDataType value1 = await session1.ReadValueAsync(
- VariableIds.Server_ServerStatus).ConfigureAwait(false);
- Assert.That(value1, Is.Not.Null);
-
- 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];
- var subscriptionTemplate = new TestableSubscription(session1.DefaultSubscription)
- {
- PublishingInterval = 1_000,
- KeepAliveCount = 5,
- PublishingEnabled = true,
- RepublishAfterTransfer = true,
- SequentialPublishing = sequentialPublishing
- };
-
- await CreateSubscriptionsAsync(
- session1,
- subscriptionTemplate,
- originSubscriptions,
- originSubscriptionCounters,
- originSubscriptionFastDataCounters,
- kTestSubscriptions,
- kQueueSize).ConfigureAwait(false);
-
- // wait
- await Task.Delay(kDelay).ConfigureAwait(false);
-
- // save the session configuration
- var configStream = new MemoryStream();
- session1.SaveSessionConfiguration(configStream);
-
- byte[] configStreamArray = configStream.ToArray();
- TestContext.Out.WriteLine($"SessionSecrets: {configStream.Length} bytes");
- TestContext.Out.WriteLine(Encoding.UTF8.GetString(configStreamArray));
-
- var subscriptionStream = new MemoryStream();
- session1.Save(
- subscriptionStream,
- session1.Subscriptions);
-
- byte[] subscriptionStreamArray = subscriptionStream.ToArray();
- TestContext.Out.WriteLine($"Subscriptions: {subscriptionStreamArray.Length} bytes");
- TestContext.Out.WriteLine(Encoding.UTF8.GetString(subscriptionStreamArray));
-
- // read the session configuration
- var loadConfigurationStream = new MemoryStream(configStreamArray);
- var sessionConfiguration = SessionConfiguration.Create(
- loadConfigurationStream,
- Telemetry);
-
- // create the inactive channel
- ITransportChannel channel2 = await ClientFixture
- .CreateChannelAsync(sessionConfiguration.ConfiguredEndpoint, false)
- .ConfigureAwait(false);
- Assert.That(channel2, Is.Not.Null);
-
- // prepare the inactive session with the new channel
- ISession session2 = ClientFixture.CreateSession(
- channel2,
- sessionConfiguration.ConfiguredEndpoint);
-
- int session2ConfigChanged = 0;
- session2.SessionConfigurationChanged += (sender, e) => session2ConfigChanged++;
-
- // apply the saved session configuration
- bool success = session2.ApplySessionConfiguration(sessionConfiguration);
-
- // restore the subscriptions
- var loadSubscriptionStream = new MemoryStream(subscriptionStreamArray);
- var restoredSubscriptions = new SubscriptionCollection(
- session2.Load(loadSubscriptionStream, true, [typeof(SubscriptionState)]));
-
- // hook notifications for log output
- int ii = 0;
- foreach (Subscription subscription in restoredSubscriptions)
- {
- subscription.Handle = ii;
- subscription.FastDataChangeCallback = (s, n, _) =>
- {
- TestContext.Out.WriteLine(
- $"FastDataChangeHandlerTarget: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
- targetSubscriptionFastDataCounters[(int)subscription.Handle]++;
- };
- subscription
- .MonitoredItems.ToList()
- .ForEach(i =>
- i.Notification += (item, _) =>
- {
- 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);
- }
- });
- ii++;
- }
-
- // hook callback to renew the user identity
- session2.RenewUserIdentity += (_, _) => userIdentity;
-
- // activate the session from saved session secrets on the new channel
- await session2.ReconnectAsync(channel2).ConfigureAwait(false);
-
- // reactivate restored subscriptions
- bool reactivateResult = await session2
- .ReactivateSubscriptionsAsync(restoredSubscriptions, sendInitialValues)
- .ConfigureAwait(false);
- Assert.That(reactivateResult, Is.True);
-
- await Task.Delay(2 * kDelay).ConfigureAwait(false);
-
- try
- {
- Assert.That(session2.SessionId, Is.EqualTo(sessionId1));
-
- DataValue value2 = await session2
- .ReadValueAsync(VariableIds.Server_ServerStatus)
- .ConfigureAwait(false);
- Assert.That(value2, Is.Not.Null);
-
- for (ii = 0; ii < kTestSubscriptions; ii++)
- {
- uint monitoredItemCount = restoredSubscriptions[ii].MonitoredItemCount;
- string errorText = $"Error in test subscription {ii}";
-
- // the static subscription doesn't resend data until there is a data change
- if (ii == 0 && !sendInitialValues)
- {
- Assert.That(targetSubscriptionCounters[ii], Is.Zero, errorText);
- Assert.That(targetSubscriptionFastDataCounters[ii], Is.Zero, errorText);
- }
- else if (ii == 0)
- {
- Assert.That(
- targetSubscriptionCounters[ii],
- Is.EqualTo(monitoredItemCount),
- errorText);
- Assert.That(targetSubscriptionFastDataCounters[ii], Is.EqualTo(1), errorText);
- }
- else
- {
- Assert.That(
- targetSubscriptionCounters[ii],
- Is.GreaterThanOrEqualTo(monitoredItemCount),
- errorText);
- Assert.That(targetSubscriptionFastDataCounters[ii], Is.GreaterThanOrEqualTo(1), errorText);
- }
- }
-
- await Task.Delay(kDelay).ConfigureAwait(false);
-
- // verify that reconnect created subclassed version of subscription and monitored item
- foreach (Subscription s in session2.Subscriptions)
- {
- Assert.That(s.GetType(), Is.EqualTo(typeof(TestableSubscription)));
- foreach (MonitoredItem m in s.MonitoredItems)
- {
- Assert.That(m.GetType(), Is.EqualTo(typeof(TestableMonitoredItem)));
- }
- }
-
- // cannot read using a closed channel, validate the status code
- if (endpoint.EndpointUrl.ToString()
- .StartsWith(Utils.UriSchemeOpcTcp, StringComparison.Ordinal))
- {
- sre = Assert.ThrowsAsync(() =>
- session1.ReadValueAsync(
- VariableIds.Server_ServerStatus));
- Assert.That(
- sre.StatusCode,
- Is.EqualTo(StatusCodes.BadSecureChannelIdInvalid),
- sre.Message);
- }
- else
- {
- ServerStatusDataType result =
- await session1.ReadValueAsync(
- VariableIds.Server_ServerStatus).ConfigureAwait(false);
- Assert.That(result, Is.Not.Null);
- }
- }
- finally
- {
- session1.DeleteSubscriptionsOnClose = true;
- session2.DeleteSubscriptionsOnClose = true;
- await session1.CloseAsync(1000, true).ConfigureAwait(false);
- await session2.CloseAsync(1000, true).ConfigureAwait(false);
- Utils.SilentDispose(session1);
- Utils.SilentDispose(session2);
- }
-
- Assert.That(session1ConfigChanged, Is.Zero);
- Assert.That(session2ConfigChanged, Is.GreaterThanOrEqualTo(0));
- }
-
[Test]
[Order(400)]
[CancelAfter(30_000)]
@@ -810,470 +511,6 @@ public async Task PublishRequestCountAsync()
}
}
- public enum TransferType
- {
- ///
- /// The origin session remains open and
- /// gives up the subscription.
- ///
- KeepOpen,
-
- ///
- /// The origin session is gracefully closed with
- /// DeleteSubscriptionsOnClose set to false.
- ///
- CloseSession,
-
- ///
- /// The origin session gets network disconnected,
- /// after transfer available sequence numbers are
- /// just acknowledged.
- ///
- DisconnectedAck,
-
- ///
- /// The origin session gets network disconnected,
- /// after transfer available sequence numbers are
- /// republished.
- ///
- DisconnectedRepublish,
-
- ///
- /// The origin session gets network disconnected,
- /// after transfer available sequence numbers are
- /// republished. The ack was delayed by the client.
- ///
- DisconnectedRepublishDelayedAck
- }
-
- [Theory]
- [Order(811)]
- public Task TransferSubscriptionOnlyAsync(
- TransferType transferType,
- bool sendInitialValues,
- bool sequentialPublishing)
- {
- return InternalTransferSubscriptionAsync(
- transferType,
- sendInitialValues,
- sequentialPublishing);
- }
-
- [Theory]
- [Order(812)]
- [Explicit]
- public async Task TransferSubscriptionOnlyDebugAsync(
- TransferType transferType,
- bool sendInitialValues,
- bool sequentialPublishing)
- {
- const int loopCount = 30;
- for (int i = 0; i < loopCount; i++)
- {
- await InternalTransferSubscriptionAsync(
- transferType,
- sendInitialValues,
- sequentialPublishing).ConfigureAwait(false);
-
- TestContext.Out.WriteLine("===========================================");
- TestContext.Out.WriteLine("===========================================");
- TestContext.Out.WriteLine($"Completed {i}th iteration.");
- TestContext.Out.WriteLine("===========================================");
- TestContext.Out.WriteLine("===========================================");
- }
- }
-
- private async Task InternalTransferSubscriptionAsync(
- TransferType transferType,
- bool sendInitialValues,
- bool sequentialPublishing)
- {
- // create test session and subscription
- ISession originSession = await ClientFixture
- .ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256)
- .ConfigureAwait(false);
- ISession targetSession = null;
- try
- {
- targetSession = await InternalTransferSubscriptionAsync(
- originSession,
- transferType,
- sendInitialValues,
- sequentialPublishing).ConfigureAwait(false);
- }
- finally
- {
- Utils.SilentDispose(originSession);
- Utils.SilentDispose(targetSession);
- }
- }
-
- private async Task InternalTransferSubscriptionAsync(
- ISession originSession,
- TransferType transferType,
- bool sendInitialValues,
- bool sequentialPublishing)
- {
- const int kTestSubscriptions = 5;
- const int kDelay = 2_000;
- const int kQueueSize = 10;
-
- if (transferType == TransferType.DisconnectedRepublishDelayedAck)
- {
- 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 subscriptionTemplate = new TestableSubscription(originSession.DefaultSubscription)
- {
- PublishingInterval = 1_000,
- LifetimeCount = 30,
- KeepAliveCount = 5,
- PublishingEnabled = true,
- 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]);
- }
- };
- }
- }
-
- // settle
- await Task.Delay(kDelay).ConfigureAwait(false);
-
- // persist the subscription state
- string filePath = Path.GetTempFileName();
-
- // 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
- 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));
-
- // hook notifications for log output
- int ii = 0;
- foreach (Subscription subscription in transferSubscriptions)
- {
- subscription.Handle = ii;
- subscription.FastDataChangeCallback = (s, n, _) =>
- {
- TestContext.Out.WriteLine(
- $"FastDataChangeHandlerTarget: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
- targetSubscriptionFastDataCounters[(int)subscription.Handle]++;
- };
- subscription
- .MonitoredItems.ToList()
- .ForEach(i =>
- i.Notification += (item, _) =>
- {
- 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);
- }
- });
- ii++;
- }
-
- // wait
- await Task.Delay(kDelay).ConfigureAwait(false);
- }
- else
- {
- // wait
- await Task.Delay(kDelay).ConfigureAwait(false);
-
- transferSubscriptions.AddRange((SubscriptionCollection)originSubscriptions.Clone());
- int ii = 0;
- transferSubscriptions.ForEach(s =>
- {
- targetSession.AddSubscription(s);
- s.Handle = ii++;
- s.FastDataChangeCallback = (sub, n, _) =>
- {
- TestContext.Out.WriteLine(
- $"FastDataChangeHandlerTarget: {sub.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
- targetSubscriptionFastDataCounters[(int)s.Handle]++;
- };
- s.MonitoredItems.ToList()
- .ForEach(i =>
- i.Notification += (item, _) =>
- {
- targetSubscriptionCounters[(int)s.Handle]++;
- foreach (DataValue value in item.DequeueValues())
- {
- TestContext.Out.WriteLine(
- "Tra:{0}: {1:20}, {2}, {3}, {4}",
- s.Id,
- item.DisplayName,
- value.WrappedValue,
- value.SourceTimestamp,
- value.StatusCode);
- }
- });
- s.StateChanged += (su, e) =>
- TestContext.Out
- .WriteLine($"StateChanged: {su.Session.SessionId}-{su.Id}-{e.Status}");
- s.PublishStatusChanged += (su, e) =>
- TestContext.Out.WriteLine(
- $"PublishStatusChanged: {su.Session.SessionId}-{su.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);
-
- // wait for some events
- await Task.Delay(2 * kDelay).ConfigureAwait(false);
-
- if (TransferType.KeepOpen == transferType)
- {
- foreach (Subscription subscription in originSubscriptions)
- {
- // assert if originSubscriptionTransferred is incremented
- Assert.That(originSubscriptionTransferred[(int)subscription.Handle], Is.EqualTo(1));
- }
- }
-
- // For DisconnectedRepublishDelayedAck with sendInitialValues=true, the server
- // sends initial values immediately after transfer, and DeferSubscriptionAcknowledge
- // on the target session prevents those acks, causing the server to republish them
- // on the next publish cycle. Poll until subscription 0 reaches 2×monitoredItemCount
- // (initial values + one republish batch) or a generous timeout expires.
- // For sendInitialValues=false there is no reliable notification source for static
- // subscription 0: the server sends no initial values, and whether the origin
- // session's unacknowledged notifications are still available for republish depends
- // on the server's session-cleanup timing, so no polling is needed.
- if (transferType == TransferType.DisconnectedRepublishDelayedAck && sendInitialValues)
- {
- uint expectedCount0 = 2u * transferSubscriptions[0].MonitoredItemCount;
- var deadline = DateTime.UtcNow.AddSeconds(10);
- while ((uint)targetSubscriptionCounters[0] < expectedCount0 &&
- DateTime.UtcNow < deadline)
- {
- await Task.Delay(200).ConfigureAwait(false);
- }
- }
-
- // 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)
- {
- // correct for delayed ack and republish count:
- // when sendInitialValues=true, the target session's DeferSubscriptionAcknowledge
- // prevents acks for the initial-value notifications, causing the server to
- // republish them — adding monitoredItemCount to account for that republish batch.
- // When sendInitialValues=false, static subscription 0 may receive zero
- // notifications (server sends no initial values and origin-session republish
- // is not reliable), so no additional count is expected.
- if (transferType == TransferType.DisconnectedRepublishDelayedAck && sendInitialValues)
- {
- targetExpectedCount += monitoredItemCount;
- }
-
- // static nodes, expect only one set of changes, another one if send initial values was set
- Assert.That(originSubscriptionCounters[jj], Is.EqualTo(originExpectedCount));
- // For DisconnectedRepublishDelayedAck deferred acks cause continuous republishing,
- // so the counter may exceed the exact expected value.
- if (transferType == TransferType.DisconnectedRepublishDelayedAck)
- {
- Assert.That(targetSubscriptionCounters[jj], Is.GreaterThanOrEqualTo(targetExpectedCount));
- }
- else
- {
- Assert.That(targetSubscriptionCounters[jj], Is.EqualTo(targetExpectedCount));
- }
- }
- 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));
- }
- }
-
- // reset counters
- Array.Clear(originSubscriptionCounters, 0, kTestSubscriptions);
- Array.Clear(originSubscriptionFastDataCounters, 0, kTestSubscriptions);
- Array.Clear(targetSubscriptionCounters, 0, kTestSubscriptions);
- Array.Clear(targetSubscriptionFastDataCounters, 0, kTestSubscriptions);
-
- // 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);
- }
-
- // wait for some events
- await Task.Delay(2 * kDelay).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]);
-
- int[] testCounter = targetSubscriptionCounters;
- int[] testFastDataCounter = targetSubscriptionFastDataCounters;
-
- if (jj == 0)
- {
- // static nodes, expect no activity
- Assert.That(testCounter[jj], Is.Zero);
- Assert.That(testFastDataCounter[jj], Is.Zero);
- }
- else
- {
- // dynamic nodes, expect changes in target counters
- Assert.That(testCounter[jj], Is.GreaterThanOrEqualTo(0));
- Assert.That(testFastDataCounter[jj], Is.GreaterThanOrEqualTo(0));
- }
- }
-
- 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);
- }
-
- // cleanup
- File.Delete(filePath);
- return targetSession;
- }
-
[Test]
[Order(1000)]
public async Task FastKeepAliveCallbackAsync()
@@ -1395,108 +632,6 @@ public async Task FastKeepAliveCallbackAsync()
Assert.That(result, Is.True);
}
- private async Task CreateSubscriptionsAsync(
- ISession session,
- Subscription template,
- SubscriptionCollection originSubscriptions,
- int[] notificationCounters,
- int[] fastDataCounters,
- int subscriptionCount,
- uint queueSize)
- {
- for (int ii = 0; ii < subscriptionCount; ii++)
- {
- // create subscription with static monitored items
- var subscription = new TestableSubscription(template)
- {
- PublishingEnabled = true,
- Handle = ii,
- FastDataChangeCallback = (s, n, _) =>
- {
- TestContext.Out.WriteLine(
- $"FastDataChangeHandlerOrigin: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
- fastDataCounters[(int)s.Handle]++;
- }
- };
-
- 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}");
-
- originSubscriptions.Add(subscription);
- session.AddSubscription(subscription);
- await subscription.CreateAsync().ConfigureAwait(false);
-
- // set defaults
- subscription.DefaultItem.DiscardOldest = true;
- subscription.DefaultItem.QueueSize = ii == 0 ? 0U : queueSize;
- subscription.DefaultItem.MonitoringMode = MonitoringMode.Reporting;
-
- // create test set
- NamespaceTable namespaceUris = Session.NamespaceUris;
- var testSet = new List();
- if (ii == 0)
- {
- testSet.AddRange(GetTestSetStatic(namespaceUris));
- }
- else
- {
- testSet.AddRange(GetTestSetSimulation(namespaceUris));
- }
-
- var list = CreateMonitoredItemTestSet(subscription, testSet).ToList();
- list.ForEach(i =>
- i.Notification += (item, _) =>
- {
- notificationCounters[(int)subscription.Handle]++;
- foreach (DataValue value in item.DequeueValues())
- {
- TestContext.Out.WriteLine(
- "Org:{0}: {1:20}, {2}, {3}, {4}",
- subscription.Id,
- item.DisplayName,
- value.WrappedValue,
- value.SourceTimestamp,
- value.StatusCode);
- }
- });
- subscription.AddItems(list);
- await subscription.ApplyChangesAsync().ConfigureAwait(false);
- }
- }
-
- private static List CreateMonitoredItemTestSet(
- Subscription subscription,
- IList nodeIds)
- {
- var list = new List();
- foreach (NodeId nodeId in nodeIds)
- {
- var item = new TestableMonitoredItem(subscription.DefaultItem)
- {
- StartNodeId = nodeId
- };
- list.Add(item);
- }
- return list;
- }
-
- ///
- /// Event handler to defer publish response sequence number acknowledge.
- ///
- private void DeferSubscriptionAcknowledge(
- ISession session,
- PublishSequenceNumbersToAcknowledgeEventArgs e)
- {
- // for testing do not ack any sequence numbers
- e.DeferredAcknowledgementsToSend.Clear();
- e.AcknowledgementsToSend.Clear();
- }
-
[Test]
[Order(900)]
public async Task SetTriggeringTrackingAsync()
diff --git a/Tests/Opc.Ua.Client.Tests/TransferSubscriptionTest.cs b/Tests/Opc.Ua.Client.Tests/TransferSubscriptionTest.cs
new file mode 100644
index 0000000000..28224e3394
--- /dev/null
+++ b/Tests/Opc.Ua.Client.Tests/TransferSubscriptionTest.cs
@@ -0,0 +1,596 @@
+/* ========================================================================
+ * 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.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace Opc.Ua.Client.Tests
+{
+ ///
+ /// Tests for subscription transfer scenarios.
+ /// Each fixture instance handles one , and the four boolean
+ /// combinations of sendInitialValues × sequentialPublishing run in parallel within each
+ /// fixture, while the five fixture instances themselves also execute in parallel.
+ ///
+ [TestFixture]
+ [Category("Client")]
+ [Category("TransferSubscription")]
+ [SetCulture("en-us")]
+ [SetUICulture("en-us")]
+ [TestFixtureSource(nameof(TransferTypeArgs))]
+ [Parallelizable(ParallelScope.Fixtures)]
+ public class TransferSubscriptionTest : ClientTestFramework
+ {
+ ///
+ /// Fixture argument list: one fixture instance per value.
+ ///
+ 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 }
+ ];
+
+ ///
+ /// Describes how a subscription transfer scenario is set up.
+ ///
+ public enum TransferType
+ {
+ ///
+ /// The origin session remains open and gives up the subscription.
+ ///
+ KeepOpen,
+
+ ///
+ /// The origin session is gracefully closed with
+ /// DeleteSubscriptionsOnClose set to false.
+ ///
+ CloseSession,
+
+ ///
+ /// The origin session gets network disconnected; available sequence
+ /// numbers are just acknowledged.
+ ///
+ DisconnectedAck,
+
+ ///
+ /// The origin session gets network disconnected; available sequence
+ /// numbers are republished.
+ ///
+ DisconnectedRepublish,
+
+ ///
+ /// The origin session gets network disconnected; available sequence
+ /// numbers are republished but the client delays the acknowledgement.
+ ///
+ DisconnectedRepublishDelayedAck
+ }
+
+ private readonly TransferType m_transferType;
+
+ ///
+ /// Initializes a new fixture instance for the given .
+ ///
+ public TransferSubscriptionTest(TransferType transferType)
+ : base(Utils.UriSchemeOpcTcp)
+ {
+ m_transferType = transferType;
+ }
+
+ ///
+ /// Start one server per fixture instance and establish a single shared session.
+ /// Tests create their own sessions independently.
+ ///
+ [OneTimeSetUp]
+ public override Task OneTimeSetUpAsync()
+ {
+ // Use a shared session for namespace URI lookups; tests create their own sessions.
+ SingleSession = true;
+ MaxChannelCount = 1000;
+ 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();
+ }
+
+ ///
+ /// Transfer subscriptions for all combinations of
+ /// and .
+ /// Each combination runs in parallel with the others within this fixture instance
+ /// and also in parallel across the five fixture instances.
+ ///
+ [Test]
+ [Combinatorial]
+ [Order(811)]
+ [Parallelizable]
+ public Task TransferSubscriptionAsync(
+ [Values] bool sendInitialValues,
+ [Values] bool sequentialPublishing)
+ {
+ return InternalTransferSubscriptionAsync(
+ m_transferType,
+ sendInitialValues,
+ sequentialPublishing);
+ }
+
+ ///
+ /// Debug variant that repeats the transfer test 30 times (explicit, not for normal CI runs).
+ ///
+ [Test]
+ [Combinatorial]
+ [Order(812)]
+ [Explicit]
+ public async Task TransferSubscriptionDebugAsync(
+ [Values] bool sendInitialValues,
+ [Values] bool sequentialPublishing)
+ {
+ const int loopCount = 30;
+ for (int i = 0; i < loopCount; i++)
+ {
+ await InternalTransferSubscriptionAsync(
+ m_transferType,
+ sendInitialValues,
+ sequentialPublishing).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)
+ {
+ // create test session and subscription
+ ISession originSession = await ClientFixture
+ .ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256)
+ .ConfigureAwait(false);
+ ISession targetSession = null;
+ try
+ {
+ targetSession = await InternalTransferSubscriptionAsync(
+ originSession,
+ transferType,
+ sendInitialValues,
+ sequentialPublishing).ConfigureAwait(false);
+ }
+ finally
+ {
+ Utils.SilentDispose(originSession);
+ Utils.SilentDispose(targetSession);
+ }
+ }
+
+ private async Task InternalTransferSubscriptionAsync(
+ ISession originSession,
+ TransferType transferType,
+ bool sendInitialValues,
+ bool sequentialPublishing)
+ {
+ const int kTestSubscriptions = 5;
+ const int kDelay = 2_000;
+ const int kQueueSize = 10;
+
+ if (transferType == TransferType.DisconnectedRepublishDelayedAck)
+ {
+ 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 subscriptionTemplate = new TestableSubscription(originSession.DefaultSubscription)
+ {
+ PublishingInterval = 1_000,
+ LifetimeCount = 30,
+ KeepAliveCount = 5,
+ PublishingEnabled = true,
+ 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]);
+ }
+ };
+ }
+ }
+
+ // settle
+ await Task.Delay(kDelay).ConfigureAwait(false);
+
+ // persist the subscription state
+ string filePath = Path.GetTempFileName();
+
+ // 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
+ 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));
+
+ // hook notifications for log output
+ int ii = 0;
+ foreach (Subscription subscription in transferSubscriptions)
+ {
+ subscription.Handle = ii;
+ subscription.FastDataChangeCallback = (s, n, _) =>
+ {
+ TestContext.Out.WriteLine(
+ $"FastDataChangeHandlerTarget: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
+ targetSubscriptionFastDataCounters[(int)subscription.Handle]++;
+ };
+ subscription
+ .MonitoredItems.ToList()
+ .ForEach(i =>
+ i.Notification += (item, _) =>
+ {
+ 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);
+ }
+ });
+ ii++;
+ }
+
+ // wait
+ await Task.Delay(kDelay).ConfigureAwait(false);
+ }
+ else
+ {
+ // wait
+ await Task.Delay(kDelay).ConfigureAwait(false);
+
+ transferSubscriptions.AddRange((SubscriptionCollection)originSubscriptions.Clone());
+ int ii = 0;
+ transferSubscriptions.ForEach(s =>
+ {
+ targetSession.AddSubscription(s);
+ s.Handle = ii++;
+ s.FastDataChangeCallback = (sub, n, _) =>
+ {
+ TestContext.Out.WriteLine(
+ $"FastDataChangeHandlerTarget: {sub.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}");
+ targetSubscriptionFastDataCounters[(int)s.Handle]++;
+ };
+ s.MonitoredItems.ToList()
+ .ForEach(i =>
+ i.Notification += (item, _) =>
+ {
+ targetSubscriptionCounters[(int)s.Handle]++;
+ foreach (DataValue value in item.DequeueValues())
+ {
+ TestContext.Out.WriteLine(
+ "Tra:{0}: {1:20}, {2}, {3}, {4}",
+ s.Id,
+ item.DisplayName,
+ value.WrappedValue,
+ value.SourceTimestamp,
+ value.StatusCode);
+ }
+ });
+ s.StateChanged += (su, e) =>
+ TestContext.Out
+ .WriteLine($"StateChanged: {su.Session.SessionId}-{su.Id}-{e.Status}");
+ s.PublishStatusChanged += (su, e) =>
+ TestContext.Out.WriteLine(
+ $"PublishStatusChanged: {su.Session.SessionId}-{su.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);
+
+ // wait for some events
+ await Task.Delay(2 * kDelay).ConfigureAwait(false);
+
+ if (TransferType.KeepOpen == transferType)
+ {
+ foreach (Subscription subscription in originSubscriptions)
+ {
+ // assert if originSubscriptionTransferred is incremented
+ Assert.That(originSubscriptionTransferred[(int)subscription.Handle], Is.EqualTo(1));
+ }
+ }
+
+ // For DisconnectedRepublishDelayedAck with sendInitialValues=true, the server
+ // sends initial values immediately after transfer, and DeferSubscriptionAcknowledge
+ // on the target session prevents those acks, causing the server to republish them
+ // on the next publish cycle. Poll until subscription 0 reaches 2×monitoredItemCount
+ // (initial values + one republish batch) or a generous timeout expires.
+ // For sendInitialValues=false there is no reliable notification source for static
+ // subscription 0: the server sends no initial values, and whether the origin
+ // session's unacknowledged notifications are still available for republish depends
+ // on the server's session-cleanup timing, so no polling is needed.
+ if (transferType == TransferType.DisconnectedRepublishDelayedAck && sendInitialValues)
+ {
+ uint expectedCount0 = 2u * transferSubscriptions[0].MonitoredItemCount;
+ var deadline = DateTime.UtcNow.AddSeconds(10);
+ while ((uint)targetSubscriptionCounters[0] < expectedCount0 &&
+ DateTime.UtcNow < deadline)
+ {
+ await Task.Delay(200).ConfigureAwait(false);
+ }
+ }
+
+ // 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)
+ {
+ // correct for delayed ack and republish count:
+ // when sendInitialValues=true, the target session's DeferSubscriptionAcknowledge
+ // prevents acks for the initial-value notifications, causing the server to
+ // republish them — adding monitoredItemCount to account for that republish batch.
+ // When sendInitialValues=false, static subscription 0 may receive zero
+ // notifications (server sends no initial values and origin-session republish
+ // is not reliable), so no additional count is expected.
+ if (transferType == TransferType.DisconnectedRepublishDelayedAck && sendInitialValues)
+ {
+ targetExpectedCount += monitoredItemCount;
+ }
+
+ // static nodes, expect only one set of changes, another one if send initial values was set
+ Assert.That(originSubscriptionCounters[jj], Is.EqualTo(originExpectedCount));
+ // For DisconnectedRepublishDelayedAck deferred acks cause continuous republishing,
+ // so the counter may exceed the exact expected value.
+ if (transferType == TransferType.DisconnectedRepublishDelayedAck)
+ {
+ Assert.That(targetSubscriptionCounters[jj], Is.GreaterThanOrEqualTo(targetExpectedCount));
+ }
+ else
+ {
+ Assert.That(targetSubscriptionCounters[jj], Is.EqualTo(targetExpectedCount));
+ }
+ }
+ 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));
+ }
+ }
+
+ // reset counters
+ Array.Clear(originSubscriptionCounters, 0, kTestSubscriptions);
+ Array.Clear(originSubscriptionFastDataCounters, 0, kTestSubscriptions);
+ Array.Clear(targetSubscriptionCounters, 0, kTestSubscriptions);
+ Array.Clear(targetSubscriptionFastDataCounters, 0, kTestSubscriptions);
+
+ // 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);
+ }
+
+ // wait for some events
+ await Task.Delay(2 * kDelay).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]);
+
+ int[] testCounter = targetSubscriptionCounters;
+ int[] testFastDataCounter = targetSubscriptionFastDataCounters;
+
+ if (jj == 0)
+ {
+ // static nodes, expect no activity
+ Assert.That(testCounter[jj], Is.Zero);
+ Assert.That(testFastDataCounter[jj], Is.Zero);
+ }
+ else
+ {
+ // dynamic nodes, expect changes in target counters
+ Assert.That(testCounter[jj], Is.GreaterThanOrEqualTo(0));
+ Assert.That(testFastDataCounter[jj], Is.GreaterThanOrEqualTo(0));
+ }
+ }
+
+ 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);
+ }
+
+ // cleanup
+ File.Delete(filePath);
+ return targetSession;
+ }
+
+ ///
+ /// Event handler to defer publish response sequence number acknowledge.
+ ///
+ private static void DeferSubscriptionAcknowledge(
+ ISession session,
+ PublishSequenceNumbersToAcknowledgeEventArgs e)
+ {
+ // for testing do not ack any sequence numbers
+ e.DeferredAcknowledgementsToSend.Clear();
+ e.AcknowledgementsToSend.Clear();
+ }
+ }
+}
diff --git a/Tests/Opc.Ua.Gds.Tests/Common.cs b/Tests/Opc.Ua.Gds.Tests/Common.cs
index 2013aef50f..1bcdbed9d1 100644
--- a/Tests/Opc.Ua.Gds.Tests/Common.cs
+++ b/Tests/Opc.Ua.Gds.Tests/Common.cs
@@ -410,6 +410,7 @@ public static async Task StartGDSAsync(
int serverStartRetries = 25;
do
{
+ retryStartServer = false;
try
{
server = new GlobalDiscoveryTestServer(true, NUnitTelemetryContext.Create(true), maxTrustListSize);
@@ -418,15 +419,17 @@ public static async Task StartGDSAsync(
catch (ServiceResultException sre)
{
serverStartRetries--;
- testPort = UnsecureRandom.Shared.Next(
- ServerFixtureUtils.MinTestPort,
- ServerFixtureUtils.MaxTestPort);
if (serverStartRetries == 0 || sre.StatusCode != StatusCodes.BadNoCommunication)
{
throw;
}
+
+ testPort = UnsecureRandom.Shared.Next(
+ ServerFixtureUtils.MinTestPort,
+ ServerFixtureUtils.MaxTestPort);
retryStartServer = true;
}
+
await Task.Delay(UnsecureRandom.Shared.Next(100, 1000)).ConfigureAwait(false);
} while (retryStartServer);
diff --git a/Tests/Opc.Ua.Server.Tests/ServerFixture.cs b/Tests/Opc.Ua.Server.Tests/ServerFixture.cs
index 3554974699..0ff22b76a1 100644
--- a/Tests/Opc.Ua.Server.Tests/ServerFixture.cs
+++ b/Tests/Opc.Ua.Server.Tests/ServerFixture.cs
@@ -223,6 +223,7 @@ public async Task StartAsync(string pkiRoot, int port = 0)
do
{
+ retryStartServer = false;
try
{
await InternalStartServerAsync(testPort).ConfigureAwait(false);
@@ -237,6 +238,7 @@ public async Task StartAsync(string pkiRoot, int port = 0)
ServerFixtureUtils.MaxTestPort);
retryStartServer = true;
}
+
await Task.Delay(UnsecureRandom.Shared.Next(100, 1000)).ConfigureAwait(false);
} while (retryStartServer);
diff --git a/Tests/Opc.Ua.Server.Tests/ServerFixtureUtils.cs b/Tests/Opc.Ua.Server.Tests/ServerFixtureUtils.cs
index 1a985ca1d9..6fa614299b 100644
--- a/Tests/Opc.Ua.Server.Tests/ServerFixtureUtils.cs
+++ b/Tests/Opc.Ua.Server.Tests/ServerFixtureUtils.cs
@@ -355,6 +355,7 @@ public static int GetNextFreeIPPort()
{
return ep.Port;
}
+
return 0;
}
}