/* ======================================================================== * 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.Generic; using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua.Client; using Opc.Ua.Client.TestFramework; namespace Opc.Ua.Subscriptions.Classic.Tests { /// /// Integration tests that exercise the classic /// end-to-end against /// the in-process reference server. /// [TestFixture] [Category("Client")] [Category("SubscriptionEngine")] [Category("Classic")] [SetCulture("en-us")] [SetUICulture("en-us")] public class SubscriptionEngineIntegrationTests : ClientTestFramework { /// /// Set up a Server and a Client instance. /// [OneTimeSetUp] public override Task OneTimeSetUpAsync() { SupportsExternalServerUrl = true; SingleSession = false; // Pin the engine to classic so this test fixture continues to // exercise the classic dispatcher after the Session default // has been flipped to V2. ClientFixtureSubscriptionEngineFactory = ClassicSubscriptionEngineFactory.Instance; return OneTimeSetUpCoreAsync(securityNone: true); } /// /// Tear down the Server and the Client. /// [OneTimeTearDown] public override Task OneTimeTearDownAsync() { return base.OneTimeTearDownAsync(); } /// /// Test setup. /// [SetUp] public override Task SetUpAsync() { return base.SetUpAsync(); } /// /// Test teardown. /// [TearDown] public override Task TearDownAsync() { return base.TearDownAsync(); } [Test] [Order(100)] public void ClassicEngineSessionHasSubscriptionEngine() { var session = (Session)Session; ISubscriptionEngineFactory factory = session.SubscriptionEngineFactory; Assert.That(factory, Is.Not.Null); Assert.That( factory, Is.InstanceOf(), "Engine factory should be ClassicSubscriptionEngineFactory"); TestContext.Out.WriteLine( "SubscriptionEngineFactory type: {0}", factory.GetType().Name); } [Test] [Order(200)] [CancelAfter(30_000)] public async Task ClassicEngineCreateAndDeleteSubscription() { using var notificationReceived = new ManualResetEventSlim(false); int dataChangeCount = 0; using var subscription = new TestableSubscription(Session.DefaultSubscription) { PublishingInterval = 500, KeepAliveCount = 10, LifetimeCount = 100, MaxNotificationsPerPublish = 0, PublishingEnabled = true, Priority = 0 }; bool added = Session.AddSubscription(subscription); Assert.That(added, Is.True); await subscription.CreateAsync().ConfigureAwait(false); Assert.That(subscription.Created, Is.True); TestContext.Out.WriteLine( "Subscription Id: {0}, PublishingInterval: {1}", subscription.Id, subscription.CurrentPublishingInterval); var monitoredItem = new TestableMonitoredItem(subscription.DefaultItem) { DisplayName = "ServerStatusCurrentTime", StartNodeId = VariableIds.Server_ServerStatus_CurrentTime, SamplingInterval = 250 }; monitoredItem.Notification += (item, _) => { foreach (DataValue value in item.DequeueValues()) { Interlocked.Increment(ref dataChangeCount); TestContext.Out.WriteLine( "{0}: {1}, {2}, {3}", item.DisplayName, value.WrappedValue, value.SourceTimestamp, value.StatusCode); notificationReceived.Set(); } }; subscription.AddItem(monitoredItem); await subscription.ApplyChangesAsync().ConfigureAwait(false); Assert.That(monitoredItem.Status.Error, Is.Null.Or.Property("StatusCode").EqualTo(StatusCodes.Good)); bool received = notificationReceived.Wait(10_000); Assert.That(received, Is.True, "Should have received at least one data change notification"); Assert.That(dataChangeCount, Is.GreaterThanOrEqualTo(1)); OutputSubscriptionInfo(TestContext.Out, subscription); bool removed = await Session.RemoveSubscriptionAsync(subscription).ConfigureAwait(false); Assert.That(removed, Is.True); TestContext.Out.WriteLine("Total data changes received: {0}", dataChangeCount); } [Test] [Order(300)] [CancelAfter(30_000)] public async Task ClassicEnginePublishRequestCountScales() { const int subscriptionCount = 3; var subscriptions = new List(); try { int initialGoodCount = Session.GoodPublishRequestCount; TestContext.Out.WriteLine( "Initial GoodPublishRequestCount: {0}", initialGoodCount); for (int i = 0; i < subscriptionCount; i++) { var subscription = new TestableSubscription(Session.DefaultSubscription) { PublishingInterval = 1000, KeepAliveCount = 10, LifetimeCount = 100, PublishingEnabled = true }; bool added = Session.AddSubscription(subscription); Assert.That(added, Is.True); await subscription.CreateAsync().ConfigureAwait(false); Assert.That(subscription.Created, Is.True); subscriptions.Add(subscription); TestContext.Out.WriteLine( "Created subscription {0}, Id: {1}", i + 1, subscription.Id); } // allow publish requests to be sent await Task.Delay(3000).ConfigureAwait(false); int currentGoodCount = Session.GoodPublishRequestCount; TestContext.Out.WriteLine( "GoodPublishRequestCount after {0} subscriptions: {1}", subscriptionCount, currentGoodCount); Assert.That( currentGoodCount, Is.GreaterThan(0), "GoodPublishRequestCount should be positive with active subscriptions"); } finally { foreach (Subscription subscription in subscriptions) { bool removed = await Session.RemoveSubscriptionAsync(subscription) .ConfigureAwait(false); Assert.That(removed, Is.True); subscription.Dispose(); } } } [Test] [Order(400)] [CancelAfter(30_000)] public async Task ClassicEngineSubscriptionReceivesKeepAlive() { int keepAliveCount = 0; using var subscription = new TestableSubscription(Session.DefaultSubscription) { PublishingInterval = 500, KeepAliveCount = 1, LifetimeCount = 100, PublishingEnabled = true }; subscription.FastKeepAliveCallback = (_, notification) => { int n = Interlocked.Increment(ref keepAliveCount); TestContext.Out.WriteLine( "KeepAlive {0}, SequenceNumber: {1}, PublishTime: {2}", n, notification.SequenceNumber, notification.PublishTime); }; bool added = Session.AddSubscription(subscription); Assert.That(added, Is.True); await subscription.CreateAsync().ConfigureAwait(false); Assert.That(subscription.Created, Is.True); TestContext.Out.WriteLine( "Subscription Id: {0}, PublishingInterval: {1}, KeepAliveCount: {2}", subscription.Id, subscription.CurrentPublishingInterval, subscription.CurrentKeepAliveCount); // Wait for keep-alive. With publishing interval 500ms and // keep-alive count of 1, a keep-alive should arrive every ~500ms. await Task.Delay(5000).ConfigureAwait(false); TestContext.Out.WriteLine("Total keep-alives received: {0}", keepAliveCount); Assert.That( keepAliveCount, Is.GreaterThanOrEqualTo(1), "Should have received at least one keep-alive notification"); OutputSubscriptionInfo(TestContext.Out, subscription); bool removed = await Session.RemoveSubscriptionAsync(subscription).ConfigureAwait(false); Assert.That(removed, Is.True); } } }