Files
UA-.NETStandard/tests/Opc.Ua.Subscriptions.Tests/SubscriptionMinimumTests.cs
T
Marc SchierandCopilot fac30c45e6 Add MinSupportedSamplingInterval configuration and honor it when revising sampling intervals (#4243)
# Description

PR #4194 removed the unused
`ServerConfiguration.MinMetadataSamplingInterval` and left nothing in
its place. As a result there was no way to populate
`Server.ServerCapabilities.MinSupportedSampleRate` (`i=2272`) —
`ServerInternalData.CreateServerObjectAsync` hardcoded it to `0`.
Separately, the only clamp applied when revising a monitored item's
sampling interval was the per-node
`BaseVariableState.MinimumSamplingInterval`, which defaults to
`MinimumSamplingIntervals.Continuous` (`0`), so reference-server nodes
accepted any requested interval (e.g. 10 ms) unrevised — exactly what
was reported.

This adds a configuration entry **in the exact schema slot #4194
vacated** and enforces it as a lower bound.

## Configuration

- **`ServerConfiguration.MinSupportedSamplingInterval`** — `double`
(milliseconds), default `0`, `[DataTypeField(Order = 17)]`, i.e. between
`MaxNotificationsPerPublish` (16) and `AvailableSamplingRates` (18).
- `ApplicationConfiguration.xsd` — `<xs:element
name="MinSupportedSamplingInterval" type="xs:double" minOccurs="0" />`
in the same position the removed element occupied.
- Fluent API **`SetMinSupportedSamplingInterval(double)`** on
`IApplicationConfigurationBuilderServerOptions`, in the place
`SetMinMetadataSamplingInterval` was removed from.
- The element is added to the six config XML files #4194 touched
(reference / CTT / AOT / GDS use `0`; `test-server-full.xml` uses a
distinctive non-zero value for the round-trip test).

The address-space Property keeps its OPC UA spec name, so
`Server.ServerCapabilities.MinSupportedSampleRate` and
`VariableIds.Server_ServerCapabilities_MinSupportedSampleRate` are
unchanged.

```xml
<MaxNotificationsPerPublish>1000</MaxNotificationsPerPublish>
<MinSupportedSamplingInterval>2000</MinSupportedSamplingInterval>
<AvailableSamplingRates>
```

```csharp
application.Build(applicationUri, productUri)
    .AsServer([endpointUrl])
    .SetMinSupportedSamplingInterval(2000);
```

## Behavior

- `ServerInternalData` now publishes the configured value to
`Server.ServerCapabilities.MinSupportedSampleRate`.
- New `SubscriptionManager.CalculateRevisedSamplingInterval(...)` (two
overloads, alongside the existing `CalculateRevisedQueueSize`)
centralizes the revision rule and replaces the logic that was duplicated
across four call sites — the create and modify paths of
`AsyncCustomNodeManager` and `CustomNodeManager` — which now expose a
`MinSupportedSamplingInterval` property seeded from the configuration
(mirroring how `MaxQueueSize` is initialized).

The rule:

1. A requested interval below zero resolves to the *default sampling
interval*: the **publishing interval of the subscription** on create,
and the item's **current sampling interval** on modify. The modify case
preserves 1.5.378 behavior, so a `ModifyMonitoredItems` call that leaves
the sampling interval unspecified does not silently retune the item.
2. A node declaring `MinimumSamplingIntervals.Continuous` (`0`) for the
`Value` Attribute reports **by exception** and is **not** bound by the
server-wide minimum — the requested interval is returned unchanged.
3. Every other item is raised to `max(nodeMinimumSamplingInterval,
MinSupportedSamplingInterval)`. Attributes other than `Value`, nodes
declaring `MinimumSamplingIntervals.Indeterminate` (`-1`), and
non-Variable nodes are bound only by `MinSupportedSamplingInterval`.
4. `double.MaxValue` is capped to one year (pre-existing behavior, now
in one place).

Event monitored items do not sample and are unaffected.

**Backward compatible:** with the default of `0` the computation is
identical to the previous code for every input, including `Continuous`,
`Indeterminate`, positive node minimums, non-`Value` attributes and
`NaN`. No existing configuration or deployment changes behavior.

Node managers that implement `INodeManager` directly can call the new
public helper to apply the same rule.

## Related Issues

- Fixes #4191

The issue thread also asked whether
`Subscription.MaxNotificationsPerPublish` is revised by the server. It
is (`SubscriptionManager.CalculateMaxNotificationsPerPublish`), but it
was untested and is not observable from the `CreateSubscription`
response, which carries no revised field for that parameter. Four tests
were added that assert it through
`SubscriptionDiagnostics.MaxNotificationsPerPublish` on a privileged
session, covering create and modify, the above-limit, below-limit and
`0` (server default) cases.

## Tests

- **`SamplingIntervalRevisionTests`** (17 unit tests,
`Opc.Ua.Server.Tests`) — the full revision matrix: negative request
resolving to the default interval, the `Continuous` bypass,
`Indeterminate`, node minimum above *and* below the floor, a floor of
`0` being a no-op, non-`Value` attributes, non-Variable and null nodes,
and the `double.MaxValue` cap.
- **`MinSupportedSamplingIntervalTests`** (10 integration tests,
`Opc.Ua.Subscriptions.Tests`) — against a reference server configured
with a 500 ms floor: the capabilities node reflects the configuration; a
node declaring 100 ms is lifted to 500 ms;
`Scalar_Static_Arrays2D_LocalizedText` (1000 ms) wins over the floor;
`CurrentTime` (250 ms) is lifted; a `Continuous` node keeps its 10 ms; a
non-`Value` attribute is floored; `ModifyMonitoredItems` behaves the
same in both directions; event items are unaffected.
- Restored the configuration builder and round-trip test coverage that
#4194 deleted.
- `TestFixture` gained a `protected virtual void
ConfigureServer(ApplicationConfiguration)` hook so a derived fixture can
vary the server configuration before start. Default implementation is a
no-op, so no existing fixture changes behavior.

## Documentation

- `docs/Subscriptions.md` — new **Sampling interval revision** section
with the input table, the four-step rule (including both
negative-interval fallbacks), XML and fluent configuration examples, and
guidance for custom node managers.
- `docs/migrate/2.0.x/configuration.md` — new section documenting that
`MinMetadataSamplingInterval` was removed and
`MinSupportedSamplingInterval` added. Verified against `master378`: the
old setting defaulted to `1000` but was never read, and
`MinSupportedSampleRate` was hardcoded to `0` with no server-wide floor.
The guide therefore stresses that these are **two different settings,
not a rename** — consumers should drop the old element (a default of `0`
reproduces 1.5.378 exactly) and only opt into the new one deliberately,
since reusing the old value would silently introduce a sampling floor.

## Validation

- `dotnet build UA.slnx` — **0 warnings, 0 errors** across all TFMs.
- .NET 10: `Opc.Ua.Server.Tests` 4091 passed,
`Opc.Ua.Subscriptions.Tests` 619 passed (full suite including the new
fixture), `Opc.Ua.InformationModel.Tests` 1063 passed,
`Opc.Ua.Configuration.Tests` 221 passed, `Opc.Ua.Gds.Tests` 119 passed.
- .NET Framework 4.8: new unit tests (200 incl. sampling groups), new
integration tests (17) and configuration round-trip (135) all passed.
- `ConsoleReferenceServer` starts successfully with the updated
configuration file.

## Checklist

- [x] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [x] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [x] I have added all necessary documentation.
- [x] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [x] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [x] I have addressed **all** PR feedback received.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 79f8090c-b2a3-409c-a5bf-56b6d2a35e70
2026-08-12 17:41:29 +00:00

2202 lines
84 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;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using NUnit.Framework;
using Opc.Ua.Client;
using Opc.Ua.Client.TestFramework;
namespace Opc.Ua.Subscriptions.Tests
{
/// <summary>
/// compliance tests for Subscription Minimum 02 covering
/// minimum publishing interval, minimum lifetime count,
/// minimum keepalive count, and server revision behavior.
/// </summary>
[NonParallelizable]
[TestFixture]
[Category("Conformance")]
[Category("Subscription")]
[Category("SubscriptionMinimum")]
[Category("SubscriptionMinimum")]
public class SubscriptionMinimumTests : TestFixture
{
[Test]
public async Task MinimumPublishingIntervalZeroRevisedUpAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 0).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalNegativeRevisedUpAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: -1).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalVerySmallRevisedUpAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 0.001).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalOneMillisecondRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 1).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalTenMillisecondsAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 10).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalFiftyAcceptedOrRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 50).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalConsistentAcrossCreatesAsync()
{
CreateSubscriptionResponse r1 = await CreateSubAsync(
interval: 1).ConfigureAwait(false);
CreateSubscriptionResponse r2 = await CreateSubAsync(
interval: 1).ConfigureAwait(false);
Assert.That(r1.RevisedPublishingInterval,
Is.EqualTo(r2.RevisedPublishingInterval),
"Same requested interval should produce same revision.");
await DeleteSubAsync(r1.SubscriptionId).ConfigureAwait(false);
await DeleteSubAsync(r2.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumPublishingIntervalFromServerCapabilitiesAsync()
{
// Read MinSupportedSamplingInterval from ServerCapabilities
ReadResponse readResp = await Session.ReadAsync(
null, 0, TimestampsToReturn.Neither,
new ReadValueId[]
{
new() {
NodeId =
VariableIds.Server_ServerCapabilities_MinSupportedSampleRate,
AttributeId = Attributes.Value
}
}.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
if (!StatusCode.IsGood(readResp.Results[0].StatusCode))
{
Assert.Fail(
"MinSupportedSamplingInterval not available.");
}
double minRate = readResp.Results[0].WrappedValue.GetDouble();
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: 0).ConfigureAwait(false);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThanOrEqualTo(minRate));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountZeroRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: 0, keepAlive: 5).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount, Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountOneRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: 1, keepAlive: 5).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThanOrEqualTo(
3 * resp.RevisedMaxKeepAliveCount));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountTwoRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: 2, keepAlive: 5).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThanOrEqualTo(
3 * resp.RevisedMaxKeepAliveCount));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountExactlyThreeTimesKeepAliveAsync()
{
// Request lifetime = 3 * keepAlive exactly
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: 15, keepAlive: 5).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThanOrEqualTo(
3 * resp.RevisedMaxKeepAliveCount));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountLessThanThreeTimesRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: 10, keepAlive: 10).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThanOrEqualTo(
3 * resp.RevisedMaxKeepAliveCount));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumLifetimeCountMaxUint32RevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: uint.MaxValue).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedLifetimeCount, Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task LifetimeCountRevisedValueIsPositiveAsync()
{
foreach (uint lt in new uint[] { 0, 1, 5, 100, uint.MaxValue })
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: lt).ConfigureAwait(false);
Assert.That(resp.RevisedLifetimeCount, Is.GreaterThan(0u),
$"Lifetime {lt} should produce positive revised.");
await DeleteSubAsync(resp.SubscriptionId)
.ConfigureAwait(false);
}
}
[Test]
public async Task MinimumKeepAliveCountZeroRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
keepAlive: 0).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumKeepAliveCountOneAcceptedOrRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
keepAlive: 1).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task MinimumKeepAliveCountMaxUint32RevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
keepAlive: uint.MaxValue).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task KeepAliveCountRevisedValueIsPositiveAsync()
{
foreach (uint ka in new uint[] { 0, 1, 5, 100, uint.MaxValue })
{
CreateSubscriptionResponse resp = await CreateSubAsync(
keepAlive: ka).ConfigureAwait(false);
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u),
$"KeepAlive {ka} should produce positive revised.");
await DeleteSubAsync(resp.SubscriptionId)
.ConfigureAwait(false);
}
}
[Test]
public async Task KeepAliveCountRevisionConsistentAcrossCreatesAsync()
{
CreateSubscriptionResponse r1 = await CreateSubAsync(
keepAlive: 1).ConfigureAwait(false);
CreateSubscriptionResponse r2 = await CreateSubAsync(
keepAlive: 1).ConfigureAwait(false);
Assert.That(r1.RevisedMaxKeepAliveCount,
Is.EqualTo(r2.RevisedMaxKeepAliveCount));
await DeleteSubAsync(r1.SubscriptionId).ConfigureAwait(false);
await DeleteSubAsync(r2.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task ModifySubscriptionKeepAliveCountZeroRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync()
.ConfigureAwait(false);
uint id = resp.SubscriptionId;
ModifySubscriptionResponse mod =
await Session.ModifySubscriptionAsync(
null, id, 500, DefaultLifetime, 0, 0, 0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(mod.ResponseHeader.ServiceResult),
Is.True);
Assert.That(mod.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(id).ConfigureAwait(false);
}
[Test]
public async Task ModifySubscriptionKeepAliveCountRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync()
.ConfigureAwait(false);
uint id = resp.SubscriptionId;
ModifySubscriptionResponse mod =
await Session.ModifySubscriptionAsync(
null, id, 500, DefaultLifetime, 1, 0, 0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(mod.ResponseHeader.ServiceResult),
Is.True);
Assert.That(mod.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(id).ConfigureAwait(false);
}
[Test]
public async Task RevisedValuesConsistentWithinSameSessionAsync()
{
CreateSubscriptionResponse r1 = await CreateSubAsync(
interval: 50, lifetime: 10, keepAlive: 3)
.ConfigureAwait(false);
CreateSubscriptionResponse r2 = await CreateSubAsync(
interval: 50, lifetime: 10, keepAlive: 3)
.ConfigureAwait(false);
Assert.That(r1.RevisedPublishingInterval,
Is.EqualTo(r2.RevisedPublishingInterval));
Assert.That(r1.RevisedLifetimeCount,
Is.EqualTo(r2.RevisedLifetimeCount));
Assert.That(r1.RevisedMaxKeepAliveCount,
Is.EqualTo(r2.RevisedMaxKeepAliveCount));
await DeleteSubAsync(r1.SubscriptionId).ConfigureAwait(false);
await DeleteSubAsync(r2.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task RevisedPublishingIntervalGreaterThanZeroAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: -100).ConfigureAwait(false);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task RevisedLifetimeAlwaysThreeTimesKeepAliveAsync()
{
foreach (uint ka in new uint[] { 1, 5, 10, 50 })
{
CreateSubscriptionResponse resp = await CreateSubAsync(
lifetime: ka, keepAlive: ka)
.ConfigureAwait(false);
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThanOrEqualTo(
3 * resp.RevisedMaxKeepAliveCount),
$"With KeepAlive={ka}");
await DeleteSubAsync(resp.SubscriptionId)
.ConfigureAwait(false);
}
}
[Test]
public async Task AllRevisedValuesReturnedInResponseAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync()
.ConfigureAwait(false);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThan(0u));
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task ModifySubscriptionRevisesAllParametersAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync()
.ConfigureAwait(false);
uint id = resp.SubscriptionId;
ModifySubscriptionResponse mod =
await Session.ModifySubscriptionAsync(
null, id, 0.001, 1, 1, 0, 0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(mod.ResponseHeader.ServiceResult),
Is.True);
Assert.That(mod.RevisedPublishingInterval,
Is.GreaterThan(0));
Assert.That(mod.RevisedLifetimeCount, Is.GreaterThan(0u));
Assert.That(mod.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(id).ConfigureAwait(false);
}
[Test]
public async Task CreateSubscriptionAllBelowMinimumAllRevisedAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: -1, lifetime: 0, keepAlive: 0)
.ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
Assert.That(resp.RevisedPublishingInterval,
Is.GreaterThan(0));
Assert.That(resp.RevisedLifetimeCount,
Is.GreaterThan(0u));
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.GreaterThan(0u));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task RevisedValuesDoNotExceedServerMaximumsAsync()
{
CreateSubscriptionResponse resp = await CreateSubAsync(
interval: double.MaxValue,
lifetime: uint.MaxValue,
keepAlive: uint.MaxValue).ConfigureAwait(false);
Assert.That(resp.RevisedPublishingInterval,
Is.LessThan(double.MaxValue));
Assert.That(resp.RevisedLifetimeCount,
Is.LessThan(uint.MaxValue));
Assert.That(resp.RevisedMaxKeepAliveCount,
Is.LessThan(uint.MaxValue));
await DeleteSubAsync(resp.SubscriptionId).ConfigureAwait(false);
}
[Test]
public async Task CreateTwoEqualPrioritySubsPublishBothAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2,
"dd"u8.ToArray()).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var receivedSubs = new HashSet<uint>();
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
receivedSubs.Add(pub.SubscriptionId);
}
Assert.That(receivedSubs, Has.Count.EqualTo(2),
"Both equal-priority subscriptions should be serviced.");
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task CreateTwoSubsWithItemsPublishCallbacksAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"At least one subscription should have received data.");
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task CreateSubsMonitorWritePublishCleanupAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0));
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task CreateSubsWritePublishVerifyNotificationsAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
// Drain initial notifications
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"Expected data change after writing values.");
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task ModifySubRaisePriorityPublishVerifyAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2,
[1, 1]).ConfigureAwait(false);
// Raise priority of first subscription
ModifySubscriptionResponse mr = await Session.ModifySubscriptionAsync(
null, subs[0].SubId, DefaultInterval,
DefaultLifetime, DefaultKeepAlive, 0, 200,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(mr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task ModifySubLowerPriorityPublishVerifyAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2,
[200, 200]).ConfigureAwait(false);
// Lower priority of first subscription
ModifySubscriptionResponse mr = await Session.ModifySubscriptionAsync(
null, subs[0].SubId, DefaultInterval,
DefaultLifetime, DefaultKeepAlive, 0, 1,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(mr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task ModifySubRaiseThenLowerPriorityAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2,
[1, 1]).ConfigureAwait(false);
// Raise priority
await Session.ModifySubscriptionAsync(
null, subs[0].SubId, DefaultInterval,
DefaultLifetime, DefaultKeepAlive, 0, 200,
CancellationToken.None).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
await PublishAsync().ConfigureAwait(false);
// Lower priority to lowest
ModifySubscriptionResponse mr = await Session.ModifySubscriptionAsync(
null, subs[0].SubId, DefaultInterval,
DefaultLifetime, DefaultKeepAlive, 0, 0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(mr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task ModifySubSettingsWritePublishAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
ModifySubscriptionResponse mr = await Session.ModifySubscriptionAsync(
null, subs[0].SubId, 500,
DefaultLifetime, DefaultKeepAlive, 0, 100,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(mr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)mr.RevisedPublishingInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0));
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeToggleOnTwoSubsAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
// Drain initial
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable both
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach (StatusCode sc in sr.Results)
{
Assert.That(StatusCode.IsGood(sc), Is.True);
}
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pubDisabled = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pubDisabled))
{
Assert.Ignore("Timing-sensitive: received stale notification while disabled.");
}
// Re-enable both
sr = await Session.SetPublishingModeAsync(
null, true, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"Expected data after re-enabling publishing.");
foreach (uint id in ids)
{
await DeleteSubAsync(id).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeDisableOneOfTwoAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable second subscription only
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false,
new uint[] { subs[1].SubId }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
await WriteInt32ValueAsync(subs[0].NodeId, s_random.Next()).ConfigureAwait(false);
await WriteInt32ValueAsync(subs[1].NodeId, s_random.Next()).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
bool gotFirst = false;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub) && pub.SubscriptionId == subs[0].SubId)
{
gotFirst = true;
}
}
if (!gotFirst)
{
Assert.Ignore("Timing-sensitive: enabled subscription not serviced within publish window.");
}
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeEnableDisabledSubAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2,
enabled: false).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
// Enable both
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, true, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"Expected data after enabling previously disabled subscriptions.");
foreach (uint id in ids)
{
await DeleteSubAsync(id).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeDisableBothSubsAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
Assert.Ignore("Timing-sensitive: received stale notification on disabled subscriptions.");
}
foreach (uint id in ids)
{
await DeleteSubAsync(id).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeReEnableBothSubsAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable then re-enable
await Session.SetPublishingModeAsync(
null, false, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, true, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"Expected data after re-enabling both subscriptions.");
foreach (uint id in ids)
{
await DeleteSubAsync(id).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeDisableOneVerifyOtherContinuesAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
// Verify both initially deliver
var initialSubs = new HashSet<uint>();
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
initialSubs.Add(pub.SubscriptionId);
}
}
// Disable second
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false,
new uint[] { subs[1].SubId }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
await WriteInt32ValueAsync(subs[0].NodeId, s_random.Next()).ConfigureAwait(false);
await WriteInt32ValueAsync(subs[1].NodeId, s_random.Next()).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
bool enabledGotData = false;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub) && pub.SubscriptionId == subs[0].SubId)
{
enabledGotData = true;
}
}
if (!enabledGotData)
{
Assert.Ignore("Timing-sensitive: enabled subscription not serviced within publish window.");
}
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task SetPublishingModeToggleVerifyStopsReportingAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 4; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable first
await Session.SetPublishingModeAsync(
null, false,
new uint[] { subs[0].SubId }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
bool disabledGotData = false;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub) && pub.SubscriptionId == subs[0].SubId)
{
disabledGotData = true;
}
}
Assert.That(disabledGotData, Is.False,
"Disabled subscription should not report data.");
foreach (uint id in ids)
{
await DeleteSubAsync(id).ConfigureAwait(false);
}
}
[Test]
public async Task DeleteMultipleValidSubscriptionsAsync()
{
const int count = 5;
uint[] ids = new uint[count];
for (int i = 0; i < count; i++)
{
CreateSubscriptionResponse cr = await CreateSubAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(cr.ResponseHeader.ServiceResult), Is.True);
ids[i] = cr.SubscriptionId;
}
DeleteSubscriptionsResponse dr = await Session.DeleteSubscriptionsAsync(
null, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(dr.ResponseHeader.ServiceResult), Is.True);
Assert.That(dr.Results.Count, Is.EqualTo(count));
foreach (StatusCode sc in dr.Results)
{
Assert.That(StatusCode.IsGood(sc), Is.True);
}
}
[Test]
public async Task CreateSubsWithItemsPublishThenDeleteAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0));
uint[] ids = [.. subs.Select(s => s.SubId)];
DeleteSubscriptionsResponse dr = await Session.DeleteSubscriptionsAsync(
null, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(dr.ResponseHeader.ServiceResult), Is.True);
}
[Test]
public async Task PublishRepublishVerifyRetransmissionAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(1).ConfigureAwait(false);
uint id = subs[0].SubId;
NodeId nodeId = subs[0].NodeId;
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
Assert.That(HasDataChangeNotification(pub), Is.True);
uint seqNum = pub.NotificationMessage.SequenceNumber;
try
{
RepublishResponse rp = await Session.RepublishAsync(
null, id, seqNum,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(rp.ResponseHeader.ServiceResult), Is.True);
Assert.That(rp.NotificationMessage.SequenceNumber, Is.EqualTo(seqNum));
}
catch (ServiceResultException ex) when (
ex.StatusCode == StatusCodes.BadMessageNotAvailable)
{
Assert.Ignore("Republish not supported or message not available.");
}
await DeleteSubAsync(id).ConfigureAwait(false);
}
[Test]
public async Task CreateSubsLifecycleCleanupAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
}
uint[] ids = [.. subs.Select(s => s.SubId)];
DeleteSubscriptionsResponse dr = await Session.DeleteSubscriptionsAsync(
null, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(dr.ResponseHeader.ServiceResult), Is.True);
foreach (StatusCode sc in dr.Results)
{
Assert.That(StatusCode.IsGood(sc), Is.True);
}
}
[Test]
public async Task CreateSubsWriteVerifyNotificationDeliveryAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var receivedSubs = new HashSet<uint>();
for (int i = 0; i < 4; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
receivedSubs.Add(pub.SubscriptionId);
}
}
Assert.That(receivedSubs, Is.Not.Empty,
"Expected at least one subscription to deliver data.");
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task FiveSubsWithPrioritiesHighestDominatesAsync()
{
try
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5,
[1, 1, 1, 1, 200]).ConfigureAwait(false);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var firstResponders = new List<uint>();
for (int round = 0; round < 3; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
firstResponders.Add(pub.SubscriptionId);
}
}
// Drain remaining
for (int i = 0; i < 10; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Priority 200 sub should be first most often (warning-level check)
if (firstResponders.Count > 0)
{
int highPrioCount = firstResponders.Count(
id => id == subs[4].SubId);
Assert.That(highPrioCount, Is.GreaterThanOrEqualTo(0),
"Highest priority subscription should be serviced " +
"(server may not implement strict priority ordering).");
}
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRequestTimeout)
{
Assert.Ignore("Timing-sensitive: publish request timed out.");
}
}
[Test]
public async Task TwoSubsPublishBothReceiveCallbacksAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(2).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
// Drain any stale notifications first
for (int i = 0; i < 10; i++)
{
try
{
await PublishAsync().ConfigureAwait(false);
}
catch (ServiceResultException)
{
break;
}
}
// Write to trigger notifications on our subscriptions
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var receivedSubs = new HashSet<uint>();
for (int i = 0; i < 10; i++)
{
try
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
receivedSubs.Add(pub.SubscriptionId);
}
catch (ServiceResultException)
{
break;
}
}
foreach ((uint subId, _) in subs)
{
if (!receivedSubs.Contains(subId))
{
Assert.Fail($"Timing-sensitive: subscription {subId} not serviced within publish window.");
}
}
uint[] ids = [.. subs.Select(s => s.SubId)];
await DeleteSubsAsync(ids).ConfigureAwait(false);
}
[Test]
public async Task FiveSubsDisabledThenEnablePublishAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5,
enabled: false).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
// Publishing should not produce data
PublishResponse pubDisabled = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pubDisabled))
{
Assert.Fail("Timing-sensitive: received stale notification on disabled subscriptions.");
}
// Enable all
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, true, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 10; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"Expected notifications after enabling all subscriptions.");
await DeleteSubsAsync(ids).ConfigureAwait(false);
}
[Test]
public async Task FiveSubsDisableEvenNumberedVerifyOddContinueAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5).ConfigureAwait(false);
uint[] allIds = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 10; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable even-indexed (1, 3) subscriptions
uint[] evenIds = [subs[1].SubId, subs[3].SubId];
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false, evenIds.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var enabledSet = new HashSet<uint> { subs[0].SubId, subs[2].SubId, subs[4].SubId };
bool enabledGotData = false;
for (int i = 0; i < 10; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub) && enabledSet.Contains(pub.SubscriptionId))
{
enabledGotData = true;
}
}
if (!enabledGotData)
{
Assert.Fail("Timing-sensitive: enabled subscriptions not serviced within window.");
}
await DeleteSubsAsync(allIds).ConfigureAwait(false);
}
[Test]
public async Task ThreeSubsWithPriorities1And125And255Async()
{
try
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(3,
[1, 125, 255]).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
// Drain initial
for (int i = 0; i < 6; i++)
{
await PublishAsync().ConfigureAwait(false);
}
var firstResponders = new List<uint>();
for (int round = 0; round < 3; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
firstResponders.Add(pub.SubscriptionId);
}
}
// Drain remaining
for (int i = 0; i < 10; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Highest priority (255) should be serviced first (informational)
if (firstResponders.Count > 0)
{
int hiPrioCount = firstResponders.Count(
id => id == subs[2].SubId);
Assert.That(hiPrioCount, Is.GreaterThanOrEqualTo(0),
"Priority=255 subscription should tend to be serviced first.");
}
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRequestTimeout)
{
Assert.Ignore("Timing-sensitive: publish request timed out.");
}
}
[Test]
public async Task FiveSamePrioritySubsRoundRobinFairnessAsync()
{
const int subCount = 5;
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(subCount,
"22222"u8.ToArray()).ConfigureAwait(false);
// Publish initial keep-alives
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < subCount * 2; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Write values and publish multiple rounds
var servicedSubs = new HashSet<uint>();
for (int round = 0; round < 3; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
for (int i = 0; i < subCount; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
servicedSubs.Add(pub.SubscriptionId);
}
}
}
// All same-priority subscriptions should eventually be serviced
Assert.That(servicedSubs, Has.Count.GreaterThanOrEqualTo(2),
"Multiple equal-priority subscriptions should be serviced " +
"in a fair manner.");
uint[] allIds = [.. subs.Select(s => s.SubId)];
await DeleteSubsAsync(allIds).ConfigureAwait(false);
}
[Test]
public async Task FiveSubsPriority1And200HighestDominatesAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5,
[1, 1, 1, 1, 200]).ConfigureAwait(false);
var firstResponders = new List<uint>();
for (int round = 0; round < 5; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
firstResponders.Add(pub.SubscriptionId);
}
// Drain remaining
for (int i = 0; i < 5; i++)
{
await PublishAsync().ConfigureAwait(false);
}
}
if (firstResponders.Count > 0)
{
int highPrioCount = firstResponders.Count(
id => id == subs[4].SubId);
Assert.That(highPrioCount, Is.GreaterThanOrEqualTo(0),
"Priority=200 subscription should be serviced first " +
"(server may not implement strict ordering).");
}
uint[] allIds = [.. subs.Select(s => s.SubId)];
await DeleteSubsAsync(allIds).ConfigureAwait(false);
}
[Test]
public async Task DeleteFiveValidSubscriptionsAsync()
{
const int count = 5;
uint[] ids = new uint[count];
for (int i = 0; i < count; i++)
{
CreateSubscriptionResponse cr = await CreateSubAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(cr.ResponseHeader.ServiceResult), Is.True);
ids[i] = cr.SubscriptionId;
}
DeleteSubscriptionsResponse dr = await Session.DeleteSubscriptionsAsync(
null, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(dr.ResponseHeader.ServiceResult), Is.True);
Assert.That(dr.Results.Count, Is.EqualTo(count));
foreach (StatusCode sc in dr.Results)
{
Assert.That(StatusCode.IsGood(sc), Is.True);
}
}
[Test]
public async Task MultiSessionMultiSubPublishCallbacksAsync()
{
const int sessionCount = 2;
const int subsPerSession = 2;
var sessions = new List<ISession>();
var subIdsBySession = new List<List<uint>>();
try
{
for (int s = 0; s < sessionCount; s++)
{
ISession sess = await ClientFixture.ConnectAsync(
ServerUrl, SecurityPolicies.None).ConfigureAwait(false);
sessions.Add(sess);
var sessionSubs = new List<uint>();
for (int j = 0; j < subsPerSession; j++)
{
CreateSubscriptionResponse cr = await sess.CreateSubscriptionAsync(
null, DefaultInterval, DefaultLifetime, DefaultKeepAlive,
0, true, 0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(cr.ResponseHeader.ServiceResult), Is.True);
NodeId nodeId = ToNodeId(Constants.ScalarStaticInt32);
var item = new MonitoredItemCreateRequest
{
ItemToMonitor = new ReadValueId
{
NodeId = nodeId,
AttributeId = Attributes.Value
},
MonitoringMode = MonitoringMode.Reporting,
RequestedParameters = new MonitoringParameters
{
ClientHandle = (uint)((s * subsPerSession) + j + 1),
SamplingInterval = 250,
Filter = default,
DiscardOldest = true,
QueueSize = 10
}
};
await sess.CreateMonitoredItemsAsync(
null, cr.SubscriptionId, TimestampsToReturn.Both,
new MonitoredItemCreateRequest[] { item }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
sessionSubs.Add(cr.SubscriptionId);
}
subIdsBySession.Add(sessionSubs);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int totalDc = 0;
for (int s = 0; s < sessionCount; s++)
{
for (int j = 0; j < subsPerSession; j++)
{
PublishResponse pub = await sessions[s].PublishAsync(
null, default,
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
totalDc++;
}
}
}
Assert.That(totalDc, Is.GreaterThan(0),
"At least one subscription across sessions should receive data.");
}
finally
{
for (int s = 0; s < sessions.Count; s++)
{
try
{
if (subIdsBySession.Count > s)
{
await sessions[s].DeleteSubscriptionsAsync(
null, subIdsBySession[s].ToArray().ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
}
await sessions[s].CloseAsync(5000, true).ConfigureAwait(false);
}
catch
{
}
sessions[s].Dispose();
}
}
}
[Test]
public async Task FiveSubsEnableAfterDisabledReceiveDataAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5,
enabled: true).ConfigureAwait(false);
uint[] ids = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 10; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable all, then re-enable
await Session.SetPublishingModeAsync(
null, false, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, true, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
int dcCount = 0;
for (int i = 0; i < 10; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub))
{
dcCount++;
}
}
Assert.That(dcCount, Is.GreaterThan(0),
"All subscriptions should receive data after re-enabling.");
await DeleteSubsAsync(ids).ConfigureAwait(false);
}
[Test]
public async Task FiveSubsDisableSubsetVerifyOthersContinueAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(5).ConfigureAwait(false);
uint[] allIds = [.. subs.Select(s => s.SubId)];
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 10; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Disable subs at index 1 and 3
uint[] disableIds = [subs[1].SubId, subs[3].SubId];
SetPublishingModeResponse sr = await Session.SetPublishingModeAsync(
null, false, disableIds.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(sr.ResponseHeader.ServiceResult), Is.True);
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
var enabledIds = new HashSet<uint> { subs[0].SubId, subs[2].SubId, subs[4].SubId };
bool enabledGotData = false;
for (int i = 0; i < 10; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
if (HasDataChangeNotification(pub) && enabledIds.Contains(pub.SubscriptionId))
{
enabledGotData = true;
}
}
if (!enabledGotData)
{
Assert.Fail("Timing-sensitive: enabled subscriptions not serviced within window.");
}
await DeleteSubsAsync(allIds).ConfigureAwait(false);
}
[Test]
public async Task ThreeSubsPriorities1And125And255OrderingAsync()
{
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(3,
[1, 125, 255]).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < 6; i++)
{
await PublishAsync().ConfigureAwait(false);
}
var firstResponders = new List<uint>();
for (int round = 0; round < 5; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
firstResponders.Add(pub.SubscriptionId);
}
// Drain
for (int i = 0; i < 3; i++)
{
await PublishAsync().ConfigureAwait(false);
}
}
if (firstResponders.Count > 0)
{
int hiPrio = firstResponders.Count(id => id == subs[2].SubId);
Assert.That(hiPrio, Is.GreaterThanOrEqualTo(0),
"Priority=255 subscription should tend to be serviced first.");
}
foreach ((uint subId, _) in subs)
{
await DeleteSubAsync(subId).ConfigureAwait(false);
}
}
[Test]
public async Task SamePrioritySubsEachServicedOncePerLoopAsync()
{
const int subCount = 5;
List<(uint SubId, NodeId NodeId)> subs = await CreateSubsWithItemsAsync(subCount,
"22222"u8.ToArray()).ConfigureAwait(false);
await Task.Delay((int)DefaultInterval + 500).ConfigureAwait(false);
for (int i = 0; i < subCount * 2; i++)
{
await PublishAsync().ConfigureAwait(false);
}
// Write and publish: each loop iteration should service each sub once
var allServiced = new HashSet<uint>();
for (int round = 0; round < 3; round++)
{
foreach ((_, NodeId nodeId) in subs)
{
await WriteInt32ValueAsync(nodeId, s_random.Next()).ConfigureAwait(false);
}
await Task.Delay((int)DefaultInterval + 200).ConfigureAwait(false);
var roundServiced = new HashSet<uint>();
for (int i = 0; i < subCount; i++)
{
PublishResponse pub = await PublishAsync().ConfigureAwait(false);
Assert.That(StatusCode.IsGood(pub.ResponseHeader.ServiceResult), Is.True);
if (HasDataChangeNotification(pub))
{
allServiced.Add(pub.SubscriptionId);
// Each subscription should not be serviced twice in same round
Assert.That(roundServiced, Does.Not.Contain(pub.SubscriptionId),
"Subscription should not be serviced twice in same publish round.");
roundServiced.Add(pub.SubscriptionId);
}
}
}
Assert.That(allServiced, Has.Count.GreaterThanOrEqualTo(2),
"Multiple equal-priority subscriptions should be serviced.");
uint[] allIds = [.. subs.Select(s => s.SubId)];
await DeleteSubsAsync(allIds).ConfigureAwait(false);
}
[Test]
public async Task MaxNotificationsPerPublishAboveServerLimitIsRevisedAsync()
{
uint serverLimit = ServerMaxNotificationsPerPublish;
Assert.That(serverLimit, Is.GreaterThan(0u),
"The test requires a server that limits the notifications per publish.");
Assert.That(
await CreateSubscriptionAndReadMaxNotificationsPerPublishAsync(
serverLimit + 100_000).ConfigureAwait(false),
Is.EqualTo(serverLimit),
"A requested value above the server limit must be revised down.");
}
[Test]
public async Task MaxNotificationsPerPublishZeroUsesServerLimitAsync()
{
Assert.That(
await CreateSubscriptionAndReadMaxNotificationsPerPublishAsync(0)
.ConfigureAwait(false),
Is.EqualTo(ServerMaxNotificationsPerPublish),
"A requested value of zero means the server picks its own limit.");
}
[Test]
public async Task MaxNotificationsPerPublishBelowServerLimitIsKeptAsync()
{
uint requested = Math.Max(1, ServerMaxNotificationsPerPublish / 2);
Assert.That(
await CreateSubscriptionAndReadMaxNotificationsPerPublishAsync(requested)
.ConfigureAwait(false),
Is.EqualTo(requested),
"A requested value within the server limit must be kept.");
}
[Test]
public async Task ModifySubscriptionRevisesMaxNotificationsPerPublishAsync()
{
uint serverLimit = ServerMaxNotificationsPerPublish;
Assert.That(
await CreateSubscriptionAndReadMaxNotificationsPerPublishAsync(
1,
serverLimit + 100_000).ConfigureAwait(false),
Is.EqualTo(serverLimit),
"ModifySubscription must apply the same limit as CreateSubscription.");
}
private uint ServerMaxNotificationsPerPublish
=> (uint)ServerFixture.Config.ServerConfiguration.MaxNotificationsPerPublish;
/// <summary>
/// Creates a subscription with the requested MaxNotificationsPerPublish, optionally
/// modifies it, and returns the value the server actually applied. The
/// CreateSubscription response carries no revised value for this parameter, so the
/// subscription diagnostics are the only observable place. They require a
/// privileged session, hence the sysadmin connection.
/// </summary>
private async Task<uint> CreateSubscriptionAndReadMaxNotificationsPerPublishAsync(
uint requestedMaxNotificationsPerPublish,
uint? modifiedMaxNotificationsPerPublish = null)
{
ISession admin = await ConnectAsSysAdminAsync().ConfigureAwait(false);
if (admin == null)
{
Assert.Ignore("The server does not expose username authentication.");
}
try
{
CreateSubscriptionResponse resp = await admin.CreateSubscriptionAsync(
null,
DefaultInterval,
DefaultLifetime,
DefaultKeepAlive,
requestedMaxNotificationsPerPublish,
true,
0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(resp.ResponseHeader.ServiceResult),
Is.True);
if (modifiedMaxNotificationsPerPublish.HasValue)
{
ModifySubscriptionResponse mod = await admin.ModifySubscriptionAsync(
null,
resp.SubscriptionId,
DefaultInterval,
DefaultLifetime,
DefaultKeepAlive,
modifiedMaxNotificationsPerPublish.Value,
0,
CancellationToken.None).ConfigureAwait(false);
Assert.That(
StatusCode.IsGood(mod.ResponseHeader.ServiceResult),
Is.True);
}
ReadResponse read = await admin.ReadAsync(
null,
0,
TimestampsToReturn.Neither,
new ReadValueId[]
{
new()
{
NodeId = VariableIds
.Server_ServerDiagnostics_SubscriptionDiagnosticsArray,
AttributeId = Attributes.Value
}
}.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
if (StatusCode.IsBad(read.Results[0].StatusCode))
{
Assert.Ignore(
"SubscriptionDiagnosticsArray not accessible: " +
$"{read.Results[0].StatusCode}");
}
Assert.That(
read.Results[0].WrappedValue.TryGetValue(
out ArrayOf<ExtensionObject> diagnostics),
Is.True);
foreach (ExtensionObject entry in diagnostics)
{
if (entry.TryGetValue(
out SubscriptionDiagnosticsDataType diagnostic,
admin.MessageContext) &&
diagnostic.SubscriptionId == resp.SubscriptionId)
{
return diagnostic.MaxNotificationsPerPublish;
}
}
Assert.Fail(
$"No diagnostics found for subscription {resp.SubscriptionId}.");
return 0;
}
finally
{
if (admin != null)
{
await admin.CloseAsync(5000, true).ConfigureAwait(false);
admin.Dispose();
}
}
}
private async Task<CreateSubscriptionResponse> CreateSubAsync(
double interval = 500,
uint lifetime = DefaultLifetime,
uint keepAlive = DefaultKeepAlive,
uint maxNotif = 0,
bool enabled = true,
byte priority = 0)
{
return await Session.CreateSubscriptionAsync(
null, interval, lifetime, keepAlive, maxNotif,
enabled, priority,
CancellationToken.None).ConfigureAwait(false);
}
private async Task DeleteSubAsync(uint id)
{
await Session.DeleteSubscriptionsAsync(
null, new uint[] { id }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
}
private async Task DeleteSubsAsync(uint[] ids)
{
await Session.DeleteSubscriptionsAsync(
null, ids.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
}
private async Task<uint> AddMonitoredItemAsync(
uint subId, NodeId nodeId,
uint handle = 1, double sampling = 250,
uint queueSize = 10)
{
var item = new MonitoredItemCreateRequest
{
ItemToMonitor = new ReadValueId
{
NodeId = nodeId,
AttributeId = Attributes.Value
},
MonitoringMode = MonitoringMode.Reporting,
RequestedParameters = new MonitoringParameters
{
ClientHandle = handle,
SamplingInterval = sampling,
Filter = default,
DiscardOldest = true,
QueueSize = queueSize
}
};
CreateMonitoredItemsResponse resp = await Session.CreateMonitoredItemsAsync(
null, subId, TimestampsToReturn.Both,
new MonitoredItemCreateRequest[] { item }.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(resp.Results[0].StatusCode), Is.True);
return resp.Results[0].MonitoredItemId;
}
private async Task WriteInt32ValueAsync(NodeId nodeId, int value)
{
WriteResponse writeResp = await Session.WriteAsync(
null,
new WriteValue[]
{
new() {
NodeId = nodeId,
AttributeId = Attributes.Value,
Value = new DataValue(Variant.From(value))
}
}.ToArrayOf(),
CancellationToken.None).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(writeResp.Results[0]), Is.True);
}
private async Task<PublishResponse> PublishAsync()
{
return await Session.PublishWithTimeoutAsync().ConfigureAwait(false);
}
private static bool HasDataChangeNotification(PublishResponse pub)
{
if (pub.NotificationMessage?.NotificationData == null ||
pub.NotificationMessage.NotificationData.Count == 0)
{
return false;
}
foreach (ExtensionObject ext in pub.NotificationMessage.NotificationData)
{
var dcn = ExtensionObject.ToEncodeable(ext) as DataChangeNotification;
if (dcn != null && dcn.MonitoredItems != default && dcn.MonitoredItems.Count > 0)
{
return true;
}
}
return false;
}
/// <summary>
/// Creates N subscriptions with monitored items and returns (subId, nodeId) pairs.
/// </summary>
private async Task<List<(uint SubId, NodeId NodeId)>> CreateSubsWithItemsAsync(
int count, byte[] priorities = null, bool enabled = true)
{
var result = new List<(uint, NodeId)>();
NodeId int32Node = ToNodeId(Constants.ScalarStaticInt32);
NodeId[] nodes = [.. Enumerable.Repeat(int32Node, count)];
for (int i = 0; i < count; i++)
{
byte prio = priorities != null && i < priorities.Length
? priorities[i] : (byte)0;
CreateSubscriptionResponse cr = await CreateSubAsync(
interval: DefaultInterval,
enabled: enabled,
priority: prio).ConfigureAwait(false);
Assert.That(StatusCode.IsGood(cr.ResponseHeader.ServiceResult), Is.True);
NodeId nodeId = nodes[i % nodes.Length];
await AddMonitoredItemAsync(cr.SubscriptionId, nodeId,
handle: (uint)(i + 1)).ConfigureAwait(false);
result.Add((cr.SubscriptionId, nodeId));
}
return result;
}
private static readonly UnsecureRandom s_random = UnsecureRandom.Shared;
private const double DefaultInterval = 1000;
private const uint DefaultLifetime = 100;
private const uint DefaultKeepAlive = 10;
}
}