mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-20 08:41:51 +08:00
romanett
9b034f8291
# Description Nullable reference types are now enabled once, in `common.props`, and every project in the repository builds warning-clean with them. **Configuration** - `<Nullable>enable</Nullable>` lives in `common.props`. All per-project `<Nullable>` settings are removed: 178 project files (enable, annotations, and the `disable` in the three Aot test projects). - 419 per-file `#nullable enable`/`restore` directives and the 4 `#nullable disable` directives are removed. - Kept on purpose: - The 5 `<auto-generated>` schema files (`UANodeSet.cs`, `UA Type Dictionary.cs`, `OPCBinarySchema.cs`, `SecuredApplication.cs`, `PInvokeHelper.cs`) keep `#nullable enable`. The compiler ignores the project setting in generated code, so the directive is required there. - `tests/Opc.Ua.Interop.LegacyPeer` keeps `disable`. It builds against the 1.5.378 packages and does not import the repository props. - `docs/DeveloperGuide.md` and `.github/copilot-instructions.md` now describe the central setting. **Migration** (the 43 unset and 13 annotations-only projects, roughly 10k diagnostics) - **Production code: annotated at the declarations.** Covers `tools/Opc.Ua.SourceGeneration.Core`, `src/Opc.Ua.Lds.Server` and the LDS sample. - Members, parameters and returns are nullable where null actually flows. - Null-check helpers carry `[NotNullWhen]` (`IsNull`, `IsValidXmlName`, `IsMethodTypeDesign`). - `CoreUtils.Format`, `Utils.Format`, `ITemplateWriter.WriteLine` take `params object?[]`. `Template.AddReplacement` takes `string?` (rendering already skips null replacements). - `IRegisteredServerStore.RegisterAsync` returns `Task<RegistrationEntry?>`: it returns null when a registration is removed. - Dereferences backed by an invariant (for example, the validated model) use `!`, so runtime behaviour is unchanged. - **Behavioural changes in production code** (small, deliberate): - `NodeSetToModelDesign` throws `InvalidOperationException` when the file is not a NodeSet. Before, it threw a NullReferenceException on the next line. - `ModelDesignValidator` starts its lookup tables empty instead of null. A redundant `if (m_nodes == null) return;` guard is gone. - **Test code: minimal annotations.** - `?` on values that can be null, `!` where the test asserts non-null. - `= null!` for fixture members assigned in SetUp. - `TryGetStructure<T>(out T? x)` where the inferred `T?` broke the `IEncodeable` constraint. - A few redundant `== null` checks on structs (`NodeId`, `ArrayOf<T>`) folded into the adjacent `IsNull`/`Count` checks. **Notes for reviewers** - The diff is large (~1,500 files) but almost entirely mechanical: `?`, `!` and directive removal. The non-annotation changes are listed above. - Known follow-up, not done here: `IModelDesign.Namespaces/TargetNamespace/NamespaceUris` stay non-nullable even though the validator returns null before it loads a dictionary. Making them nullable would ripple through every generator call site. ## Related Issues - Fixes #4439 ## Checklist - [ ] 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. - [ ] 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. - [ ] 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. - [ ] I have addressed **all** PR feedback received. The "no new warnings" box covers net10.0 and net48: full non-incremental builds of `UA.slnx`, `tools/SourceGeneration.slnx`, `tools/MigrationAnalyzer.slnx` and the Aot test projects finished with 0 warnings and 0 errors under the repository's warnings-as-errors settings. net9.0 and net8.0 also had no nullable diagnostics, but they were not rebuilt after the final fixes. Tests have not been run locally yet. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
3324 lines
158 KiB
C#
3324 lines
158 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.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using Microsoft.Extensions.Time.Testing;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client.Subscriptions.Fakes;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
using Opc.Ua.Tests;
|
|
|
|
namespace Opc.Ua.Client.Subscriptions
|
|
{
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("SubscriptionManager")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
public sealed class SubscriptionTests
|
|
{
|
|
[SetUp]
|
|
public void SetUp()
|
|
{
|
|
m_mockSubscriptionServices = new Mock<ISubscriptionServiceSetClientMethods>();
|
|
m_mockMonitoredItemServices = new Mock<IMonitoredItemServiceSetClientMethods>();
|
|
m_mockMethodServices = new Mock<IMethodServiceSetClientMethods>();
|
|
m_session = new FakeSubscriptionContext
|
|
{
|
|
SubscriptionServiceSet = m_mockSubscriptionServices.Object,
|
|
MethodServiceSet = m_mockMethodServices.Object,
|
|
MonitoredItemServiceSet = m_mockMonitoredItemServices.Object
|
|
};
|
|
m_completion = new FakeMessageAckQueue();
|
|
m_options = OptionsFactory.Create<SubscriptionOptions>();
|
|
|
|
m_telemetry = NUnitTelemetryContext.Create();
|
|
m_mockNotificationDataHandler = new Mock<ISubscriptionNotificationHandler>();
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task IntentionalDeletionDropsPublishDemandWithoutUnregisteringSubscriptionAsync(bool disable)
|
|
{
|
|
var clock = new FakeTimeProvider();
|
|
var publishEntered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var deleted = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var context = new FakeSubscriptionManagerContext
|
|
{
|
|
OnPublishAsync = async (_, _, ct) =>
|
|
{
|
|
publishEntered.TrySetResult(true);
|
|
await deleted.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
throw new ServiceResultException(StatusCodes.BadNoSubscription);
|
|
}
|
|
};
|
|
m_mockSubscriptionServices.Setup(value => value.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
TestSubscription subscription = null!;
|
|
context.CreateSubscriptionFactory = (handler, _, queue) =>
|
|
subscription = new TestSubscription(
|
|
m_session, handler, queue, m_options, m_telemetry, 10, timeProvider: clock);
|
|
await using var manager = new SubscriptionManager(
|
|
context, m_telemetry.LoggerFactory, DiagnosticsMasks.None, clock)
|
|
{
|
|
MinPublishWorkerCount = 1,
|
|
MaxPublishWorkerCount = 1
|
|
};
|
|
ISubscription logicalSubscription = manager.Add(m_mockNotificationDataHandler.Object, m_options);
|
|
manager.Resume();
|
|
await publishEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
Assert.That(manager.PublishWorkerCount, Is.EqualTo(1));
|
|
Assert.That(manager.CreatedCount, Is.EqualTo(1));
|
|
subscription.SubscriptionStateChanged.Reset();
|
|
if (disable)
|
|
{
|
|
m_options.Configure(_ => TestSubscription.SubscriptionOptions with { Disabled = true });
|
|
await subscription.SubscriptionStateChanged.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await subscription.DeleteAsync(CancellationToken.None).ConfigureAwait(false);
|
|
}
|
|
deleted.TrySetResult(true);
|
|
await WaitForAsync(() => manager.PublishWorkerCount == 0, TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
Assert.That(subscription.Created, Is.False);
|
|
Assert.That(subscription.Id, Is.Zero);
|
|
Assert.That(manager.Count, Is.EqualTo(1));
|
|
Assert.That(manager.Items, Is.EquivalentTo([logicalSubscription]));
|
|
Assert.That(manager.CreatedCount, Is.Zero);
|
|
Assert.That(manager.PublishWorkerCount, Is.Zero,
|
|
"Intentional deletion must remove retained demand, unlike a recovery identifier reset.");
|
|
int badRequests = manager.BadPublishRequestCount;
|
|
int attempts = context.PublishCalls.Count;
|
|
clock.Advance(TimeSpan.FromMinutes(1));
|
|
Assert.That(manager.BadPublishRequestCount, Is.EqualTo(badRequests));
|
|
Assert.That(manager.BadPublishRequestCount, Is.LessThanOrEqualTo(1));
|
|
Assert.That(context.PublishCalls, Has.Count.EqualTo(attempts));
|
|
m_mockSubscriptionServices.Verify(value => value.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(ids => ids.Count == 1 && ids[0] == 10),
|
|
It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
|
|
[Test]
|
|
public async Task AddMonitoredItemShouldAddItemToMonitoredItemsAsync()
|
|
{
|
|
// Arrange
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(sut.MonitoredItems.Items, Does.Contain(monitoredItem));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeSubscriptionOptionsShouldCallCreateAsyncIfNotCreatedAsync()
|
|
{
|
|
// Arrange
|
|
var publishingInterval = TimeSpan.FromSeconds(100);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
publishingInterval.TotalMilliseconds, 21, 7, 10, true,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
m_options.Configure(o => o with { Disabled = true });
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.Created, Is.False);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(TimeSpan.Zero));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.Zero);
|
|
Assert.That(sut.CurrentLifetimeCount, Is.Zero);
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.Zero);
|
|
Assert.That(sut.CurrentPriority, Is.Zero);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
m_options.Configure(o => o with
|
|
{
|
|
Disabled = false,
|
|
PublishingEnabled = true,
|
|
PublishingInterval = publishingInterval,
|
|
KeepAliveCount = 7,
|
|
LifetimeCount = 15,
|
|
Priority = 3,
|
|
MaxNotificationsPerPublish = 10
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(5));
|
|
Assert.That(sut.CurrentLifetimeCount, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentPriority, Is.EqualTo(3));
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Server-revised values at the edge of their wire types (an
|
|
/// hour-scale interval times a UInt32 keep-alive count exceeds
|
|
/// <see cref="TimeSpan.MaxValue"/>) or a NaN interval must not abort
|
|
/// the create after the server already created the subscription.
|
|
/// </summary>
|
|
[TestCase(3_600_000d, uint.MaxValue - 1)]
|
|
[TestCase(double.NaN, 10u)]
|
|
[TestCase(1e300, 10u)]
|
|
[CancelAfter(10_000)]
|
|
public async Task CreateToleratesExtremeRevisedValuesAsync(
|
|
double revisedPublishingInterval,
|
|
uint revisedKeepAliveCount,
|
|
CancellationToken testCt)
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = uint.MaxValue,
|
|
RevisedMaxKeepAliveCount = revisedKeepAliveCount,
|
|
RevisedPublishingInterval = revisedPublishingInterval
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
await sut.WaitForCreatedAsync(testCt).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(revisedKeepAliveCount));
|
|
m_mockSubscriptionServices.Verify(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public void SaturatingTimeSpanNeverThrows()
|
|
{
|
|
Assert.That(SaturatingTimeSpan.FromMilliseconds(double.NaN, TimeSpan.FromSeconds(3)),
|
|
Is.EqualTo(TimeSpan.FromSeconds(3)));
|
|
Assert.That(SaturatingTimeSpan.FromMilliseconds(double.PositiveInfinity, TimeSpan.Zero),
|
|
Is.EqualTo(TimeSpan.MaxValue));
|
|
Assert.That(SaturatingTimeSpan.FromMilliseconds(-5, TimeSpan.FromSeconds(1)),
|
|
Is.EqualTo(TimeSpan.Zero));
|
|
Assert.That(SaturatingTimeSpan.FromMilliseconds(1500, TimeSpan.Zero),
|
|
Is.EqualTo(TimeSpan.FromMilliseconds(1500)));
|
|
Assert.That(SaturatingTimeSpan.Multiply(TimeSpan.FromHours(1), uint.MaxValue),
|
|
Is.EqualTo(TimeSpan.MaxValue));
|
|
Assert.That(SaturatingTimeSpan.Multiply(TimeSpan.FromSeconds(2), 3),
|
|
Is.EqualTo(TimeSpan.FromSeconds(6)));
|
|
Assert.That(SaturatingTimeSpan.FromHours(uint.MaxValue), Is.EqualTo(TimeSpan.MaxValue));
|
|
Assert.That(SaturatingTimeSpan.FromHours(2), Is.EqualTo(TimeSpan.FromHours(2)));
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
[CancelAfter(10_000)]
|
|
public async Task AutomaticCreationWaitsForSessionRecoveryAsync(
|
|
bool disposeWhilePaused,
|
|
CancellationToken testCt)
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 30,
|
|
RevisedMaxKeepAliveCount = 10,
|
|
RevisedPublishingInterval = 1000
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
|
|
await using var manager = new SubscriptionManager(
|
|
new FakeSubscriptionManagerContext(), m_telemetry.LoggerFactory, DiagnosticsMasks.None);
|
|
manager.SetSessionRecoveryPaused(true);
|
|
await using var sut = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, manager, m_options, m_telemetry);
|
|
|
|
Assert.That(sut.Created, Is.False);
|
|
m_mockSubscriptionServices.Verify(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Never);
|
|
|
|
if (disposeWhilePaused)
|
|
{
|
|
await sut.DisposeAsync().AsTask().WaitAsync(testCt).ConfigureAwait(false);
|
|
Assert.That(sut.Created, Is.False);
|
|
m_mockSubscriptionServices.Verify(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Never);
|
|
}
|
|
else
|
|
{
|
|
manager.SetSessionRecoveryPaused(false);
|
|
await sut.SubscriptionStateChanged.WaitAsync(testCt).ConfigureAwait(false);
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
m_mockSubscriptionServices.Verify(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[CancelAfter(10_000)]
|
|
public async Task SessionRecoveryCancelsAutomaticCreationAndPreservesExplicitRestorationAsync(
|
|
CancellationToken testCt)
|
|
{
|
|
var started = new TaskCompletionSource<CancellationToken>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var blockedResponse = new TaskCompletionSource<CreateSubscriptionResponse>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var modified = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
int creates = 0;
|
|
int afterCreates = 0;
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
|
It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.Returns((RequestHeader _, double _, uint _, uint _, uint _, bool _, byte _,
|
|
CancellationToken token) =>
|
|
{
|
|
if (Interlocked.Increment(ref creates) == 1)
|
|
{
|
|
started.TrySetResult(token);
|
|
return new ValueTask<CreateSubscriptionResponse>(blockedResponse.Task.WaitAsync(token));
|
|
}
|
|
return new ValueTask<CreateSubscriptionResponse>(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 30,
|
|
RevisedMaxKeepAliveCount = 10,
|
|
RevisedPublishingInterval = 1000
|
|
});
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.ModifySubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), 22, It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.Returns(() =>
|
|
{
|
|
modified.TrySetResult(true);
|
|
return new ValueTask<ModifySubscriptionResponse>(new ModifySubscriptionResponse
|
|
{
|
|
RevisedLifetimeCount = 30,
|
|
RevisedMaxKeepAliveCount = 10,
|
|
RevisedPublishingInterval = 1000
|
|
});
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
|
|
await using var manager = new SubscriptionManager(
|
|
new FakeSubscriptionManagerContext(), m_telemetry.LoggerFactory, DiagnosticsMasks.None);
|
|
await using var sut = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, manager, m_options, m_telemetry)
|
|
{
|
|
OnAfterCreateAsync = _ =>
|
|
{
|
|
Interlocked.Increment(ref afterCreates);
|
|
return default;
|
|
}
|
|
};
|
|
CancellationToken automaticCreation = await started.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
|
|
manager.SetSessionRecoveryPaused(true);
|
|
Assert.That(automaticCreation.IsCancellationRequested, Is.True);
|
|
await sut.RecreateAsync(testCt).ConfigureAwait(false);
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
Assert.That(Volatile.Read(ref creates), Is.EqualTo(2));
|
|
Assert.That(Volatile.Read(ref afterCreates), Is.EqualTo(1));
|
|
|
|
m_options.Configure(options => options with { Priority = 1 });
|
|
manager.SetSessionRecoveryPaused(false);
|
|
await modified.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
Assert.That(Volatile.Read(ref creates), Is.EqualTo(2));
|
|
Assert.That(Volatile.Read(ref afterCreates), Is.EqualTo(1));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeSubscriptionOptionsShouldCallModifyAsyncIfCreatedAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var publishingInterval = TimeSpan.FromSeconds(100);
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.ModifySubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
22, publishingInterval.TotalMilliseconds, 30, 10, 0,
|
|
4, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new ModifySubscriptionResponse
|
|
{
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.SetPublishingModeAsync(It.IsAny<RequestHeader>(),
|
|
true, It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new SetPublishingModeResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingEnabled, Is.False);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(publishingInterval));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(7));
|
|
Assert.That(sut.CurrentLifetimeCount, Is.EqualTo(21));
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentPriority, Is.EqualTo(3));
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
m_options.Configure(o => o with
|
|
{
|
|
Disabled = false,
|
|
PublishingInterval = publishingInterval,
|
|
PublishingEnabled = true,
|
|
Priority = 4
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(5));
|
|
Assert.That(sut.CurrentLifetimeCount, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.Zero);
|
|
Assert.That(sut.CurrentPriority, Is.EqualTo(4));
|
|
Assert.That(sut.CurrentPublishingEnabled, Is.True);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeSubscriptionOptionsShouldCallSetPublishingModeIfItIsTheOnlyChangeAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var publishingInterval = TimeSpan.FromSeconds(100);
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.SetPublishingModeAsync(It.IsAny<RequestHeader>(),
|
|
true, It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new SetPublishingModeResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingEnabled, Is.False);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
m_options.Configure(o => o with
|
|
{
|
|
Disabled = false,
|
|
PublishingEnabled = true,
|
|
PublishingInterval = publishingInterval,
|
|
KeepAliveCount = 7,
|
|
LifetimeCount = 15,
|
|
Priority = 3,
|
|
MaxNotificationsPerPublish = 10
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.CurrentPublishingEnabled, Is.True);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task KeepAliveTimerShouldNotReportStoppedForNormalPublishGapAsync()
|
|
{
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.OnSubscriptionUpdateComplete(
|
|
created: true,
|
|
subscriptionId: 22,
|
|
revisedPublishingInterval: TimeSpan.FromMilliseconds(2800),
|
|
revisedKeepAliveCount: 30,
|
|
revisedLifetimeCount: 300,
|
|
priority: 3,
|
|
maxNotificationsPerPublish: 10,
|
|
publishingEnabled: true);
|
|
|
|
sut.ResetPublishStateForTest();
|
|
|
|
bool stoppedObserved = await WaitForSignalAsync(
|
|
sut.PublishStateChanged,
|
|
TimeSpan.FromSeconds(4)).ConfigureAwait(false);
|
|
|
|
Assert.That(stoppedObserved, Is.False);
|
|
Assert.That(sut.LastPublishState, Is.Default);
|
|
|
|
await sut.OnPublishReceivedAsync(BuildKeepAliveMessage(1), null, [])
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(sut.LastPublishState, Is.Default);
|
|
}
|
|
}
|
|
|
|
private static async Task<bool> WaitForSignalAsync(
|
|
AsyncManualResetEvent signal,
|
|
TimeSpan timeout)
|
|
{
|
|
Task waitTask = signal.WaitAsync();
|
|
Task completedTask = await Task.WhenAny(waitTask, Task.Delay(timeout))
|
|
.ConfigureAwait(false);
|
|
return completedTask == waitTask;
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeMonitoredItemOptionsShouldAddCreatedItemsAsync()
|
|
{
|
|
// Arrange
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 100,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.SubscriptionStateChanged.Reset();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", OptionsFactory.Create(new MonitoredItems.MonitoredItemOptions
|
|
{
|
|
StartNodeId = NodeId.Parse("ns=2;s=Demo")
|
|
}), out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
// Act
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(100));
|
|
Assert.That(monitoredItem.CurrentSamplingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(monitoredItem.CurrentQueueSize, Is.EqualTo(10));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Reporting));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeMonitoredItemOptionsShouldChangeSubscriptionAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(monitoredItem.ClientHandle));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Sampling));
|
|
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.ModifyMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemModifyRequest>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new ModifyMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
RevisedSamplingInterval = 100000,
|
|
RevisedQueueSize = 1000
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
options.Configure(o => o with
|
|
{
|
|
StartNodeId = NodeId.Parse("ns=2;s=Demo"),
|
|
MonitoringMode = MonitoringMode.Sampling,
|
|
SamplingInterval = TimeSpan.FromSeconds(555),
|
|
QueueSize = 3333,
|
|
DiscardOldest = true
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(monitoredItem.CurrentSamplingInterval, Is.EqualTo(TimeSpan.FromSeconds(100)));
|
|
Assert.That(monitoredItem.CurrentQueueSize, Is.EqualTo(1000));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Sampling));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ChangeMonitoredItemOptionsNameShouldDeleteAndRecreateMonitoredItemAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(monitoredItem.ClientHandle));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Sampling));
|
|
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == monitoredItem.ServerId), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
TimestampsToReturn.Both,
|
|
It.Is<ArrayOf<MonitoredItemCreateRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].ItemToMonitor.NodeId.ToString().Contains("NewDemo")),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 400,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
options.Configure(o => o with
|
|
{
|
|
StartNodeId = NodeId.Parse("ns=3;s=NewDemo"), // Changed
|
|
MonitoringMode = MonitoringMode.Reporting,
|
|
SamplingInterval = TimeSpan.FromSeconds(555),
|
|
QueueSize = 3333,
|
|
DiscardOldest = true
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(monitoredItem.CurrentSamplingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(monitoredItem.CurrentQueueSize, Is.EqualTo(10));
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(400));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Reporting));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task UpdatingMonitoringModeOnlyShouldCallSetMonitoringModeAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(monitoredItem.ClientHandle));
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Sampling));
|
|
// Now we have a monitored item in sampling mode
|
|
|
|
// Only set monitoring mode should be called for the item
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.SetMonitoringModeAsync(It.IsAny<RequestHeader>(),
|
|
sut.Id, MonitoringMode.Reporting,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == monitoredItem.ServerId),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new SetMonitoringModeResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
options.Configure(o => TestMonitoredItem.CreatedOptions with
|
|
{
|
|
MonitoringMode = MonitoringMode.Reporting
|
|
});
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(monitoredItem.CurrentMonitoringMode, Is.EqualTo(MonitoringMode.Reporting));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RemovingMonitoredItemShouldRemoveRemovedItemAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(sut.MonitoredItems.Count, Is.EqualTo(1));
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
// Now we got an item that is created
|
|
|
|
// Only delete monitored item should be called
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == monitoredItem.ServerId),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
|
|
success = sut.MonitoredItems.TryRemove(monitoredItem.ClientHandle);
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(sut.MonitoredItems.Count, Is.Zero);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RemovingMonitoredItemShouldTryAgainIfDeleteFailsAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(sut.MonitoredItems.Count, Is.EqualTo(1));
|
|
// Now we got an item that is created
|
|
|
|
// Only delete monitored item should be called
|
|
m_mockMonitoredItemServices
|
|
.SetupSequence(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(),
|
|
2, It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == monitoredItem.ServerId),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader
|
|
{
|
|
ServiceResult = StatusCodes.Bad
|
|
},
|
|
Results = []
|
|
})
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Bad]
|
|
})
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
;
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
success = sut.MonitoredItems.TryRemove(monitoredItem.ClientHandle);
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(sut.MonitoredItems.Count, Is.Zero);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConditionRefreshAsyncShouldCallSessionCallAsync()
|
|
{
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<ArrayOf<CallMethodRequest>>(), It.IsAny<CancellationToken>()))
|
|
.Callback<RequestHeader, ArrayOf<CallMethodRequest>, CancellationToken>((_, requests, _) =>
|
|
{
|
|
Assert.That(requests.Count, Is.EqualTo(1));
|
|
Assert.That(requests[0].ObjectId, Is.EqualTo(ObjectTypeIds.ConditionType));
|
|
Assert.That(requests[0].MethodId, Is.EqualTo(MethodIds.ConditionType_ConditionRefresh));
|
|
Assert.That(requests[0].InputArguments.Count, Is.EqualTo(1));
|
|
Assert.That(requests[0].InputArguments[0].TryGetValue(out uint subscriptionId), Is.True);
|
|
Assert.That(subscriptionId, Is.EqualTo(2u));
|
|
})
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
await sut.ConditionRefreshAsync(default).ConfigureAwait(false);
|
|
m_mockMethodServices.Verify();
|
|
}
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task ConditionRefreshPropagatesMethodFailureAsync(bool invalidMethod)
|
|
{
|
|
StatusCode statusCode = invalidMethod ? StatusCodes.BadMethodInvalid : StatusCodes.BadUserAccessDenied;
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(service => service.CallAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<CallMethodRequest>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results = [new CallMethodResult { StatusCode = statusCode }]
|
|
});
|
|
|
|
Assert.That(
|
|
async () => await sut.ConditionRefreshAsync(CancellationToken.None).ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(statusCode));
|
|
}
|
|
}
|
|
|
|
[TestCase(0)]
|
|
[TestCase(2)]
|
|
public async Task ConditionRefreshRejectsUnexpectedResultCountsAsync(int resultCount)
|
|
{
|
|
var results = new CallMethodResult[resultCount];
|
|
for (int index = 0; index < results.Length; index++)
|
|
{
|
|
results[index] = new CallMethodResult { StatusCode = StatusCodes.Good };
|
|
}
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(service => service.CallAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<CallMethodRequest>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse { Results = results });
|
|
|
|
Assert.That(
|
|
async () => await sut.ConditionRefreshAsync(CancellationToken.None).ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>());
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConditionRefreshAsyncThrowsIfNotYetCreatedAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act
|
|
async Task act() => await sut.ConditionRefreshAsync(CancellationToken.None).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
ServiceResultException ex = Assert.ThrowsAsync<ServiceResultException>(async () => await act().ConfigureAwait(false));
|
|
Assert.That(ex.StatusCode, Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DeleteAsyncShouldCallSessionDeleteSubscriptionsAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
}).
|
|
Verifiable(Times.Once);
|
|
|
|
// Act
|
|
await sut.DeleteAsync(default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncShouldCallSessionDeleteSubscriptionsAndCleanupMonitoredItemsAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
}).
|
|
Verifiable(Times.Once);
|
|
|
|
// Act
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(sut.MonitoredItems.Items, Is.Empty);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisableShouldCallSessionDeleteSubscriptionsButNotMonitoredItemsAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
}).
|
|
Verifiable(Times.Once);
|
|
|
|
// Act
|
|
sut.SubscriptionStateChanged.Reset();
|
|
m_options.Configure(o => o with { Disabled = true });
|
|
await sut.SubscriptionStateChanged.WaitAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(sut.MonitoredItems.Items, Is.Not.Empty);
|
|
}
|
|
}
|
|
|
|
[TestCase(true)]
|
|
[TestCase(false)]
|
|
public async Task DeleteAndResetRetireQueuedMessagesBeforeReplacementAsync(bool delete)
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedPublishingInterval = 1000,
|
|
RevisedLifetimeCount = 30,
|
|
RevisedMaxKeepAliveCount = 10
|
|
});
|
|
var firstEntered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseFirst = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var barrier = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var freshData = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var received = new ConcurrentQueue<uint>();
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.OnDataChangeAsync = async sequenceNumber =>
|
|
{
|
|
received.Enqueue(sequenceNumber);
|
|
if (sequenceNumber == 4999)
|
|
{
|
|
firstEntered.TrySetResult(true);
|
|
await releaseFirst.Task.ConfigureAwait(false);
|
|
}
|
|
if (sequenceNumber == 1)
|
|
{
|
|
freshData.TrySetResult(true);
|
|
}
|
|
};
|
|
sut.OnKeepAliveAsync = () =>
|
|
{
|
|
barrier.TrySetResult(true);
|
|
return default;
|
|
};
|
|
try
|
|
{
|
|
await sut.OnPublishReceivedAsync(BuildDataMessage(4999), null, []).ConfigureAwait(false);
|
|
await firstEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await sut.OnPublishReceivedAsync(BuildDataMessage(5000), null, []).ConfigureAwait(false);
|
|
|
|
Task reset = delete
|
|
? sut.DeleteAsync(CancellationToken.None).AsTask()
|
|
: sut.ResetToRecreateAsync(CancellationToken.None).AsTask();
|
|
releaseFirst.TrySetResult(true);
|
|
await reset.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
if (delete)
|
|
{
|
|
await sut.CreateAsync(TestSubscription.SubscriptionOptions, CancellationToken.None)
|
|
.ConfigureAwait(false);
|
|
}
|
|
await sut.WaitForCreatedAsync(CancellationToken.None)
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await sut.OnPublishReceivedAsync(BuildKeepAliveMessage(6000), null, []).ConfigureAwait(false);
|
|
await barrier.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
Assert.That(received, Is.EqualTo(new uint[] { 4999 }));
|
|
await sut.OnPublishReceivedAsync(BuildDataMessage(1), null, []).ConfigureAwait(false);
|
|
await freshData.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
Assert.That(received, Is.EqualTo(new uint[] { 4999, 1 }));
|
|
Assert.That(m_completion.QueuedAcks, Has.None.Matches<SubscriptionAcknowledgement>(
|
|
acknowledgement => acknowledgement.SequenceNumber == 5000));
|
|
}
|
|
finally
|
|
{
|
|
releaseFirst.TrySetResult(true);
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisabledCycleRecreatesExistingMonitoredItemsAsync()
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedPublishingInterval = 1000,
|
|
RevisedLifetimeCount = 30,
|
|
RevisedMaxKeepAliveCount = 10
|
|
});
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 22, TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results = [new MonitoredItemCreateResult { MonitoredItemId = 200, StatusCode = StatusCodes.Good }]
|
|
});
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.MonitoredItems.TryAdd("item",
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(), out IMonitoredItem? item), Is.True);
|
|
Assert.That(item!.Created, Is.True);
|
|
sut.SubscriptionStateChanged.Reset();
|
|
m_options.Configure(options => options with { Disabled = true });
|
|
await sut.SubscriptionStateChanged.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
Assert.That(sut.Created, Is.False);
|
|
Assert.That(item.Created, Is.False);
|
|
|
|
m_options.Configure(_ => TestSubscription.SubscriptionOptions);
|
|
await WaitForAsync(() => item.ServerId == 200, TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
Assert.That(sut.Id, Is.EqualTo(22u));
|
|
Assert.That(item.ServerId, Is.EqualTo(200u));
|
|
m_mockMonitoredItemServices.Verify(s => s.CreateMonitoredItemsAsync(
|
|
It.IsAny<RequestHeader>(), 22, TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
}
|
|
|
|
[TestCase("interval", 21u)]
|
|
[TestCase("keepalive", 24u)]
|
|
[TestCase("lifetime", 40u)]
|
|
[TestCase("priority", 21u)]
|
|
[TestCase("notifications", 21u)]
|
|
public async Task RequestedSettingsTrackCompleteRequestsNotServerRevisionsAsync(
|
|
string setting, uint expectedLifetime)
|
|
{
|
|
SubscriptionOptions original = TestSubscription.SubscriptionOptions;
|
|
SubscriptionOptions changed = setting switch
|
|
{
|
|
"interval" => original with { PublishingInterval = TimeSpan.FromSeconds(200) },
|
|
"keepalive" => original with { KeepAliveCount = 8 },
|
|
"lifetime" => original with { LifetimeCount = 40 },
|
|
"priority" => original with { Priority = 4 },
|
|
"notifications" => original with { MaxNotificationsPerPublish = 20 },
|
|
_ => throw new ArgumentOutOfRangeException(nameof(setting))
|
|
};
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedPublishingInterval = 1000,
|
|
RevisedLifetimeCount = 300,
|
|
RevisedMaxKeepAliveCount = 100
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.ModifySubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), 22, changed.PublishingInterval.TotalMilliseconds,
|
|
It.IsAny<uint>(), changed.KeepAliveCount, changed.MaxNotificationsPerPublish, changed.Priority,
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new ModifySubscriptionResponse
|
|
{
|
|
RevisedPublishingInterval = 2000,
|
|
RevisedLifetimeCount = 600,
|
|
RevisedMaxKeepAliveCount = 200
|
|
});
|
|
m_options.Configure(options => options with { Disabled = true });
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
await sut.CreateAsync(original, CancellationToken.None).ConfigureAwait(false);
|
|
await sut.ModifyAsync(original, CancellationToken.None).ConfigureAwait(false);
|
|
m_mockSubscriptionServices.Verify(s => s.ModifySubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<uint>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Never);
|
|
|
|
await sut.ModifyAsync(changed, CancellationToken.None).ConfigureAwait(false);
|
|
await sut.ModifyAsync(changed, CancellationToken.None).ConfigureAwait(false);
|
|
|
|
m_mockSubscriptionServices.Verify(s => s.ModifySubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), 22, changed.PublishingInterval.TotalMilliseconds, expectedLifetime,
|
|
changed.KeepAliveCount, changed.MaxNotificationsPerPublish, changed.Priority,
|
|
It.IsAny<CancellationToken>()), Times.Once);
|
|
m_mockSubscriptionServices.Verify(s => s.ModifySubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<uint>(), It.IsAny<double>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DeleteAsyncShouldCatchAllExceptionsAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Bad]
|
|
}).
|
|
Verifiable(Times.Once);
|
|
|
|
// Act
|
|
await sut.DeleteAsync(default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncShouldDisposeCleanlyAsync()
|
|
{
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
// Act & Assert - should not throw
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposalBoundsUnavailableServerDeletionAndStillReleasesLocalStateAsync()
|
|
{
|
|
var clock = new FakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
CancellationToken deletionToken = default;
|
|
m_mockSubscriptionServices.Setup(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.Returns((RequestHeader _, ArrayOf<uint> _, CancellationToken token) =>
|
|
{
|
|
deletionToken = token;
|
|
entered.TrySetResult(true);
|
|
return new ValueTask<DeleteSubscriptionsResponse>(WaitForServerAsync(token));
|
|
});
|
|
var subscription = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
|
subscriptionIdForAlreadyCreatedState: 22, timeProvider: clock);
|
|
Task disposal = subscription.DisposeAsync().AsTask();
|
|
try
|
|
{
|
|
await entered.Task.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
|
|
clock.Advance(TimeSpan.FromSeconds(4));
|
|
Assert.That(disposal.IsCompleted, Is.False);
|
|
Assert.That(deletionToken.IsCancellationRequested, Is.False);
|
|
clock.Advance(TimeSpan.FromSeconds(1));
|
|
await disposal.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
|
|
Assert.That(deletionToken.IsCancellationRequested, Is.True);
|
|
Assert.That(subscription.Disposed, Is.True);
|
|
Assert.That(subscription.Created, Is.False);
|
|
Assert.That(subscription.MonitoredItems.Items, Is.Empty);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await disposal.ConfigureAwait(false);
|
|
}
|
|
|
|
async Task<DeleteSubscriptionsResponse> WaitForServerAsync(CancellationToken token)
|
|
{
|
|
await release.Task.WaitAsync(token).ConfigureAwait(false);
|
|
return new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] };
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A notification callback that awaits SetTriggeringAsync holds the
|
|
/// dispatch gate while waiting for an apply pass, which needs the
|
|
/// state lock. Dispose holds the state lock while it waits for the
|
|
/// dispatch gate. The pending operation must be abandoned so both
|
|
/// sides progress instead of deadlocking.
|
|
/// </summary>
|
|
[Test]
|
|
[CancelAfter(15_000)]
|
|
public async Task DisposeDoesNotDeadlockWithCallbackAwaitingSetTriggeringAsync(
|
|
CancellationToken testCt)
|
|
{
|
|
m_mockSubscriptionServices.Setup(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
// Keep the apply pass from running so the queued triggering
|
|
// operation stays pending, as it would while the state lock is
|
|
// held by the delete.
|
|
m_completion.OnRunWithSessionAvailableAsync = (_, ct) =>
|
|
new ValueTask(Task.Delay(Timeout.Infinite, ct));
|
|
var subscription = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
|
subscriptionIdForAlreadyCreatedState: 22);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Trigger", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? trigger), Is.True);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Triggered", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? triggered), Is.True);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var result = new TaskCompletionSource<SetTriggeringResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
subscription.OnDataChangeAsync = async _ =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
try
|
|
{
|
|
result.TrySetResult(await subscription.SetTriggeringAsync(
|
|
trigger!, [triggered!]).ConfigureAwait(false));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
result.TrySetException(ex);
|
|
}
|
|
};
|
|
|
|
await subscription.OnPublishReceivedAsync(BuildDataMessage(1), null, []).ConfigureAwait(false);
|
|
await entered.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
|
|
await subscription.DisposeAsync().AsTask().WaitAsync(testCt).ConfigureAwait(false);
|
|
|
|
SetTriggeringResult setTriggering = await result.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
Assert.That(StatusCode.IsBad(setTriggering.ServiceResult), Is.True);
|
|
Assert.That(subscription.Disposed, Is.True);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A reset that gives up waiting for a running callback (its token
|
|
/// is cancelled) must withdraw its release request. Otherwise every
|
|
/// later SetTriggeringAsync from a callback is abandoned at once,
|
|
/// although no reset waits for the dispatch gate any more.
|
|
/// </summary>
|
|
[Test]
|
|
[CancelAfter(15_000)]
|
|
public async Task CancelledResetDoesNotAbandonLaterSetTriggeringFromCallbackAsync(
|
|
CancellationToken testCt)
|
|
{
|
|
m_completion.OnRunWithSessionAvailableAsync = (_, ct) =>
|
|
new ValueTask(Task.Delay(Timeout.Infinite, ct));
|
|
await using var subscription = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
|
subscriptionIdForAlreadyCreatedState: 22);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Trigger", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? trigger), Is.True);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Triggered", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? triggered), Is.True);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var hold = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var calling = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var result = new TaskCompletionSource<SetTriggeringResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
using var setTriggeringCts = new CancellationTokenSource();
|
|
subscription.OnDataChangeAsync = async _ =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
await hold.Task.ConfigureAwait(false);
|
|
calling.TrySetResult(true);
|
|
try
|
|
{
|
|
result.TrySetResult(await subscription.SetTriggeringAsync(
|
|
trigger!, [triggered!], ct: setTriggeringCts.Token).ConfigureAwait(false));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
result.TrySetException(ex);
|
|
}
|
|
};
|
|
|
|
await subscription.OnPublishReceivedAsync(BuildDataMessage(1), null, []).ConfigureAwait(false);
|
|
await entered.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
|
|
// The reset contends for the gate the callback holds, then gives up.
|
|
using (var resetCts = new CancellationTokenSource())
|
|
{
|
|
resetCts.CancelAfter(200);
|
|
Assert.That(async () => await subscription.ResetToRecreateAsync(resetCts.Token)
|
|
.ConfigureAwait(false), Throws.InstanceOf<OperationCanceledException>());
|
|
}
|
|
|
|
hold.TrySetResult(true);
|
|
await calling.Task.WaitAsync(testCt).ConfigureAwait(false);
|
|
await Task.Delay(200, testCt).ConfigureAwait(false);
|
|
Assert.That(result.Task.IsCompleted, Is.False,
|
|
"No reset waits for the gate, so the operation must stay pending.");
|
|
|
|
setTriggeringCts.Cancel();
|
|
Assert.That(() => result.Task.WaitAsync(testCt),
|
|
Throws.InstanceOf<OperationCanceledException>());
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposalDrainsCallbackPastCleanupBudgetAndDisposesOwnedItemsAsync()
|
|
{
|
|
var clock = new CleanupTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var received = new ConcurrentQueue<uint>();
|
|
CancellationToken deletionToken = default;
|
|
m_mockSubscriptionServices.Setup(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.Returns((RequestHeader _, ArrayOf<uint> _, CancellationToken token) =>
|
|
{
|
|
deletionToken = token;
|
|
token.ThrowIfCancellationRequested();
|
|
return new ValueTask<DeleteSubscriptionsResponse>(
|
|
new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
});
|
|
await using var subscription = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
|
subscriptionIdForAlreadyCreatedState: 22, timeProvider: clock);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Held", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(), out IMonitoredItem? owned),
|
|
Is.True);
|
|
var monitoredItem = (TestMonitoredItem)owned!;
|
|
subscription.OnDataChangeAsync = async sequenceNumber =>
|
|
{
|
|
received.Enqueue(sequenceNumber);
|
|
entered.TrySetResult(true);
|
|
await release.Task.ConfigureAwait(false);
|
|
};
|
|
try
|
|
{
|
|
await subscription.OnPublishReceivedAsync(BuildDataMessage(1), null, []).ConfigureAwait(false);
|
|
await entered.Task.ConfigureAwait(false);
|
|
|
|
Task disposal = subscription.DisposeAsync().AsTask();
|
|
TimeSpan cleanupBudget = await clock.CleanupScheduled.ConfigureAwait(false);
|
|
Assert.That(cleanupBudget, Is.EqualTo(TimeSpan.FromSeconds(5)));
|
|
clock.Advance(TimeSpan.FromSeconds(5));
|
|
|
|
Assert.That(disposal.IsCompleted, Is.False);
|
|
Assert.That(monitoredItem!.DisposeCalls, Is.Zero);
|
|
Assert.That(subscription.MonitoredItems.Items, Does.Contain(monitoredItem));
|
|
release.TrySetResult(true);
|
|
await disposal.ConfigureAwait(false);
|
|
|
|
Assert.That(deletionToken.IsCancellationRequested, Is.True);
|
|
Assert.That(subscription.Disposed, Is.True);
|
|
Assert.That(subscription.Created, Is.False);
|
|
Assert.That(subscription.MonitoredItems.Items, Is.Empty);
|
|
Assert.That(monitoredItem.DisposeCalls, Is.EqualTo(1));
|
|
Assert.That(received, Is.EqualTo(new uint[] { 1 }));
|
|
Assert.That(subscription.LastSequenceNumberForTest, Is.Zero);
|
|
Assert.That(subscription.LastDataSequenceNumberForTest, Is.Zero);
|
|
Assert.That(m_completion.CompletedProcessors, Is.EqualTo(new IMessageProcessor[] { subscription }));
|
|
Assert.That(m_completion.CompletedSubscriptions, Is.EqualTo(new uint[] { 22 }));
|
|
Assert.That(m_completion.QueuedAcks, Has.Count.EqualTo(1));
|
|
Assert.That(m_completion.QueuedAcks[0].SubscriptionId, Is.EqualTo(22u));
|
|
Assert.That(m_completion.QueuedAcks[0].SequenceNumber, Is.EqualTo(1u));
|
|
m_mockSubscriptionServices.Verify(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.Is<ArrayOf<uint>>(ids => ids.Count == 1 && ids[0] == 22),
|
|
It.IsAny<CancellationToken>()), Times.Once);
|
|
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(monitoredItem.DisposeCalls, Is.EqualTo(1));
|
|
Assert.That(m_completion.CompletedProcessors, Has.Count.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
try
|
|
{
|
|
await subscription.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await monitoredItem!.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task StateManagerFaultStillDisposesRemainingOwnedItemsAsync()
|
|
{
|
|
m_mockSubscriptionServices.Setup(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
await using var subscription = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
|
subscriptionIdForAlreadyCreatedState: 22);
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Disposed", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(), out IMonitoredItem? disposed),
|
|
Is.True);
|
|
await using var disposedItem = (TestMonitoredItem)disposed!;
|
|
Assert.That(subscription.MonitoredItems.TryAdd(
|
|
"Owned", OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(), out IMonitoredItem? owned),
|
|
Is.True);
|
|
await using var ownedItem = (TestMonitoredItem)owned!;
|
|
await disposedItem!.DisposeAsync().ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await subscription.DisposeAsync().ConfigureAwait(false),
|
|
Throws.InstanceOf<ObjectDisposedException>()).ConfigureAwait(false);
|
|
|
|
Assert.That(subscription.Disposed, Is.True);
|
|
Assert.That(ownedItem!.DisposeCalls, Is.EqualTo(1));
|
|
Assert.That(subscription.MonitoredItems.Items, Is.Empty);
|
|
Assert.That(m_completion.CompletedProcessors, Is.EqualTo(new IMessageProcessor[] { subscription }));
|
|
Assert.That(m_completion.CompletedSubscriptions, Is.EqualTo(new uint[] { 22 }));
|
|
}
|
|
|
|
[Test]
|
|
public async Task FindItemByClientHandleShouldReturnMonitoredItemAsync()
|
|
{
|
|
// Arrange
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
|
|
// Act
|
|
success = sut.MonitoredItems.TryGetMonitoredItemByClientHandle(
|
|
monitoredItem.ClientHandle, out IMonitoredItem? result);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(result, Is.EqualTo(monitoredItem));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FindItemByClientHandleShouldReturnNullIfNotFoundAsync()
|
|
{
|
|
// Arrange
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out _);
|
|
Assert.That(success, Is.True);
|
|
|
|
// Act
|
|
success = sut.MonitoredItems.TryGetMonitoredItemByClientHandle(55,
|
|
out IMonitoredItem? result);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.False);
|
|
Assert.That(result, Is.Null);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task OnPublishReceivedAsyncShouldProcessNotificationAsync()
|
|
{
|
|
// Arrange
|
|
var message = new NotificationMessage();
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act & Assert - should not throw
|
|
await sut.OnPublishReceivedAsync(message, null, null!).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncFromNotificationCallbackShouldFailFastAsync()
|
|
{
|
|
var observed = new TaskCompletionSource<Exception>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.OnKeepAliveAsync = async () =>
|
|
{
|
|
try
|
|
{
|
|
await sut.RecreateAsync(default).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
observed.TrySetResult(ex);
|
|
}
|
|
};
|
|
|
|
await sut.OnPublishReceivedAsync(BuildKeepAliveMessage(1), null, [])
|
|
.ConfigureAwait(false);
|
|
Exception exception = await observed.Task
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
Assert.That(exception, Is.TypeOf<InvalidOperationException>());
|
|
Assert.That(m_completion.PublishingQuiescenceCalls, Is.Zero);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncFromNotificationCallbackShouldFailFastAsync()
|
|
{
|
|
var observed = new TaskCompletionSource<Exception>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.OnKeepAliveAsync = async () =>
|
|
{
|
|
try
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
observed.TrySetResult(ex);
|
|
}
|
|
};
|
|
|
|
await sut.OnPublishReceivedAsync(BuildKeepAliveMessage(1), null, [])
|
|
.ConfigureAwait(false);
|
|
Exception exception = await observed.Task
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
Assert.That(exception, Is.TypeOf<InvalidOperationException>());
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncShouldReCreateSubscriptionAndMonitoredItemsAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
|
|
// We have a running subscription with one monitored item.
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 10u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 22,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 200,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
Assert.That(sut.Created, Is.True);
|
|
sut.SetMessageStateForTest(17, 19, [11, 13]);
|
|
|
|
// Act
|
|
await sut.RecreateAsync(default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(5));
|
|
Assert.That(sut.CurrentLifetimeCount, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentPriority, Is.EqualTo(3));
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(200));
|
|
Assert.That(sut.LastSequenceNumberForTest, Is.Zero);
|
|
Assert.That(sut.LastDataSequenceNumberForTest, Is.Zero);
|
|
Assert.That(sut.AvailableSequenceNumbersForTest, Is.Empty);
|
|
Assert.That(m_completion.PublishingQuiescenceCalls, Is.EqualTo(1));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncShouldReCreateSubscriptionsAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 10u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 22,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 200,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
Assert.That(sut.Created, Is.True);
|
|
|
|
// Act
|
|
await sut.RecreateAsync(default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.CurrentPublishingInterval, Is.EqualTo(TimeSpan.FromSeconds(10)));
|
|
Assert.That(sut.CurrentKeepAliveCount, Is.EqualTo(5));
|
|
Assert.That(sut.CurrentLifetimeCount, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentMaxNotificationsPerPublish, Is.EqualTo(10));
|
|
Assert.That(sut.CurrentPriority, Is.EqualTo(3));
|
|
Assert.That(sut.Id, Is.EqualTo(22));
|
|
Assert.That(sut.MonitoredItems.Count, Is.EqualTo(1));
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(200));
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncShouldPreserveCurrentGenerationWhenDeleteFailsAsync()
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 10u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.BadUserAccessDenied]
|
|
});
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.SetMessageStateForTest(17, 19, [11, 13]);
|
|
|
|
ServiceResultException exception =
|
|
Assert.ThrowsAsync<ServiceResultException>(async () =>
|
|
await sut.RecreateAsync(default).ConfigureAwait(false));
|
|
|
|
Assert.That(exception.StatusCode,
|
|
Is.EqualTo(StatusCodes.BadUserAccessDenied));
|
|
Assert.That(sut.Id, Is.EqualTo(10u));
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.LastSequenceNumberForTest, Is.EqualTo(17u));
|
|
Assert.That(sut.LastDataSequenceNumberForTest, Is.EqualTo(19u));
|
|
Assert.That(sut.AvailableSequenceNumbersForTest,
|
|
Is.EqualTo(new uint[] { 11, 13 }));
|
|
Assert.That(m_completion.DroppedSubscriptions,
|
|
Does.Not.Contain(10u));
|
|
Assert.That(m_completion.PublishingQuiescenceCalls, Is.EqualTo(1));
|
|
m_mockSubscriptionServices.Verify(s =>
|
|
s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncShouldDiscardQueuedBurstFromRetiredGenerationAsync()
|
|
{
|
|
var deleteEntered = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var continueDelete = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 10u),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(async (RequestHeader _, ArrayOf<uint> _,
|
|
CancellationToken _) =>
|
|
{
|
|
deleteEntered.TrySetResult(true);
|
|
await continueDelete.Task.ConfigureAwait(false);
|
|
return new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
};
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
});
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Task recreate = sut.RecreateAsync(default).AsTask();
|
|
await deleteEntered.Task.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
for (uint sequenceNumber = 1; sequenceNumber <= 1024; sequenceNumber++)
|
|
{
|
|
await sut.OnPublishReceivedAsync(
|
|
BuildKeepAliveMessage(sequenceNumber), null, [])
|
|
.ConfigureAwait(false);
|
|
}
|
|
continueDelete.TrySetResult(true);
|
|
await recreate.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
for (uint sequenceNumber = 1; sequenceNumber <= 3; sequenceNumber++)
|
|
{
|
|
await sut.OnPublishReceivedAsync(
|
|
BuildKeepAliveMessage(sequenceNumber), null, [])
|
|
.ConfigureAwait(false);
|
|
}
|
|
await WaitForAsync(() => sut.KeepAliveNotificationCount == 3,
|
|
TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await Task.Delay(100).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.KeepAliveNotificationCount, Is.EqualTo(3));
|
|
Assert.That(sut.KeepAliveSequenceNumbers,
|
|
Is.EqualTo(new uint[] { 1, 2, 3 }));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncShouldWaitForActiveRecreateAsync()
|
|
{
|
|
var deleteEntered = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var continueDelete = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<ArrayOf<uint>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(async (RequestHeader _, ArrayOf<uint> ids,
|
|
CancellationToken _) =>
|
|
{
|
|
if (ids[0] == 10u)
|
|
{
|
|
deleteEntered.TrySetResult(true);
|
|
await continueDelete.Task.ConfigureAwait(false);
|
|
}
|
|
return new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
};
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
});
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
Task recreate = sut.RecreateAsync(default).AsTask();
|
|
await deleteEntered.Task.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Task dispose = sut.DisposeAsync().AsTask();
|
|
await Task.Delay(100).ConfigureAwait(false);
|
|
Assert.That(dispose.IsCompleted, Is.False);
|
|
|
|
continueDelete.TrySetResult(true);
|
|
await recreate.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await dispose.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateAsyncShouldRunAfterCreateHookBeforeMonitoredItemsOnEveryCreateAsync()
|
|
{
|
|
var order = new List<string>();
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<ArrayOf<uint>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.Callback(() => order.Add("subscription"))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 22,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
});
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 22,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => order.Add("items"))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 200,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
});
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 10);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options =
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
Assert.That(sut.MonitoredItems.TryAdd("Test", options, out _), Is.True);
|
|
sut.OnAfterCreateAsync = _ =>
|
|
{
|
|
order.Add("durable");
|
|
return default;
|
|
};
|
|
|
|
await sut.RecreateAsync(default).ConfigureAwait(false);
|
|
await sut.RecreateAsync(default).ConfigureAwait(false);
|
|
|
|
Assert.That(order,
|
|
Is.EqualTo(s_recreateOperationOrder));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RemoveItemShouldRemoveItemFromMonitoredItemsAsync()
|
|
{
|
|
// Arrange
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
|
|
// Act
|
|
success = sut.MonitoredItems.TryRemove(monitoredItem.ClientHandle);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(sut.MonitoredItems.Items, Does.Not.Contain(monitoredItem));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// An unusable GetMonitoredItems answer is treated like a server that
|
|
/// does not provide the optional method: the server ids the client
|
|
/// knows complete the transfer.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldUseKnownServerIdsWhenResponseWrong1Async()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([1990u, 2200u, 3300u]), // serverHandles
|
|
new Variant([22222u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], CancellationToken.None).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem!.Created, Is.True);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// An unusable GetMonitoredItems answer is treated like a server that
|
|
/// does not provide the optional method: the server ids the client
|
|
/// knows complete the transfer.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldUseKnownServerIdsWhenResponseWrong2Async()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant(["string"]), // serverHandles
|
|
new Variant([22u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], CancellationToken.None).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem!.Created, Is.True);
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldCallGetMonitoredItemsAsyncAndCreateServerItemAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([19900u]), // serverHandles
|
|
new Variant([monitoredItem.ClientHandle]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(19900));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldCallGetMonitoredItemsAsyncAndDoNothingIfStateIsCorrectAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
uint serverId = monitoredItem.ServerId;
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([monitoredItem.ServerId]), // serverHandles
|
|
new Variant([monitoredItem.ClientHandle]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(serverId));
|
|
Assert.That(success, Is.True);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A create whose response was lost and that was then issued again
|
|
/// leaves two server items with the same client handle (client
|
|
/// handles need not be unique, Part 4 §7.21). Synchronizing must keep
|
|
/// the bound item, delete the extra server item and not throw.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldTolerateDuplicateClientHandlesAsync()
|
|
{
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
uint clientId = monitoredItem!.ClientHandle;
|
|
uint serverId = monitoredItem.ServerId;
|
|
const uint duplicateServerId = 55555u;
|
|
Assert.That(serverId, Is.Not.EqualTo(duplicateServerId));
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([duplicateServerId, serverId]), // serverHandles
|
|
new Variant([clientId, clientId]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == duplicateServerId),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(serverId));
|
|
Assert.That(sut.MonitoredItems.TryGetMonitoredItemByClientHandle(
|
|
clientId, out IMonitoredItem? resolved), Is.True);
|
|
Assert.That(resolved, Is.SameAs(monitoredItem));
|
|
m_mockMonitoredItemServices.Verify();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldCallGetMonitoredItemsAsyncAndDeleteItemIfNotInSubscriptionAsync()
|
|
{
|
|
// Arrange
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([19900u]), // serverHandles
|
|
new Variant([300u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 19900u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
bool success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(success, Is.True);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldReturnFalseWhenDeletingExtraneousServerItemFailsAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([19900u]), // serverHandles
|
|
new Variant([300u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 19900u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader
|
|
{
|
|
ServiceResult = StatusCodes.Good
|
|
},
|
|
Results = [StatusCodes.BadUnexpectedError]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
bool success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.False);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// GetMonitoredItems is an optional method (OPC 10000-5, 9.1). When
|
|
/// the server does not provide it, the server ids the client already
|
|
/// knows complete the transfer instead of failing it.
|
|
/// </summary>
|
|
[TestCaseSource(nameof(UnavailableGetMonitoredItemsResults))]
|
|
public async Task TryCompleteTransferAsyncShouldUseKnownServerIdsIfGetMonitoredItemsIsUnavailableAsync(
|
|
StatusCode statusCode)
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.Created, Is.True);
|
|
uint clientId = monitoredItem.ClientHandle;
|
|
uint serverId = monitoredItem.ServerId;
|
|
|
|
SetupGetMonitoredItemsResult(statusCode);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.Created, Is.True);
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(serverId));
|
|
Assert.That(monitoredItem.ClientHandle, Is.EqualTo(clientId));
|
|
m_mockMonitoredItemServices.Verify(s => s.DeleteMonitoredItemsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<uint>(),
|
|
It.IsAny<ArrayOf<uint>>(),
|
|
It.IsAny<CancellationToken>()), Times.Never);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A transient GetMonitoredItems failure still fails the transfer, and
|
|
/// so does an unavailable method when no server id is known (the items
|
|
/// cannot be mapped): the caller then recreates the subscription.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldFailIfGetMonitoredItemsFailsTransientlyOrNoServerIdIsKnownAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem!.Created, Is.True);
|
|
|
|
// Act: a timeout does not prove anything, the items are reset
|
|
SetupGetMonitoredItemsResult(StatusCodes.BadTimeout);
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.False);
|
|
Assert.That(monitoredItem.Created, Is.False);
|
|
|
|
// Act: the method is unavailable and no server id is known
|
|
SetupGetMonitoredItemsResult(StatusCodes.BadMethodInvalid);
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.False);
|
|
Assert.That(monitoredItem.Created, Is.False);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// BadSubscriptionIdInvalid comes from a server that implements
|
|
/// GetMonitoredItems but no longer has the subscription, so the known
|
|
/// server ids must not be trusted.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldFailIfGetMonitoredItemsRejectsTheSubscriptionAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem!.Created, Is.True);
|
|
|
|
SetupGetMonitoredItemsResult(StatusCodes.BadSubscriptionIdInvalid);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.False);
|
|
Assert.That(monitoredItem.Created, Is.False);
|
|
}
|
|
}
|
|
|
|
private static IEnumerable<StatusCode> UnavailableGetMonitoredItemsResults()
|
|
{
|
|
yield return StatusCodes.Bad;
|
|
yield return StatusCodes.BadMethodInvalid;
|
|
yield return StatusCodes.BadNotSupported;
|
|
yield return StatusCodes.BadNothingToDo; // asyncua
|
|
yield return StatusCodes.BadInternalError; // open62541 without XML encoding
|
|
}
|
|
|
|
private void SetupGetMonitoredItemsResult(StatusCode statusCode)
|
|
{
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = statusCode,
|
|
OutputArguments = []
|
|
}
|
|
]
|
|
});
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldAssignServerIdToMonitoredItemWithClientIdAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.Created, Is.True);
|
|
uint clientId = monitoredItem.ClientHandle;
|
|
uint serverId = monitoredItem.ServerId;
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([serverId]), // serverHandles
|
|
new Variant([19900u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.ClientHandle, Is.EqualTo(19900));
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(serverId));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryCompleteTransferAsyncShouldCreateWhatIsMissingOnServerAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 2);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options = OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
bool success = sut.MonitoredItems.TryAdd("Test", options, out IMonitoredItem? monitoredItem);
|
|
Assert.That(monitoredItem, Is.Not.Null);
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.Created, Is.True);
|
|
uint clientId = monitoredItem.ClientHandle;
|
|
uint serverId = monitoredItem.ServerId;
|
|
|
|
m_mockMethodServices
|
|
.Setup(s => s.CallAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
|
r.Count == 1 &&
|
|
r[0].InputArguments.Count == 1 &&
|
|
r[0].InputArguments[0].AsBoxedObject()!.Equals(2u) &&
|
|
r[0].ObjectId == ObjectIds.Server &&
|
|
r[0].MethodId == MethodIds.Server_GetMonitoredItems), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CallResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
OutputArguments =
|
|
[
|
|
new Variant([30000u]), // serverHandles
|
|
new Variant([19900u]) // clientHandles
|
|
]
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Delete monitored item should be called
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.DeleteMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 30000u), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteMonitoredItemsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.CreateMonitoredItemsAsync(It.IsAny<RequestHeader>(), 2,
|
|
TimestampsToReturn.Both,
|
|
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateMonitoredItemsResponse
|
|
{
|
|
Results =
|
|
[
|
|
new ()
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = 44444,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}
|
|
]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
success = await sut.TryCompleteTransferAsync([], default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(success, Is.True);
|
|
Assert.That(monitoredItem.Created, Is.True);
|
|
Assert.That(monitoredItem.ClientHandle, Is.EqualTo(clientId));
|
|
Assert.That(monitoredItem.ServerId, Is.EqualTo(44444));
|
|
|
|
// m_mockSession.Verify() was no-op (no Verifiable setups on the context); inner-mock verifications retained.
|
|
m_mockMonitoredItemServices.Verify();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SetTriggeringAsyncShouldThrowOnNullTriggeringItemAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act + Assert
|
|
Assert.That(async () => await sut.SetTriggeringAsync(
|
|
null!, null, null, default).ConfigureAwait(false),
|
|
Throws.TypeOf<ArgumentNullException>());
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SetTriggeringAsyncShouldThrowWhenItemNotInSubscriptionAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
bool added = sut.MonitoredItems.TryAdd("trig",
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? trig);
|
|
Assert.That(added, Is.True);
|
|
|
|
// Construct a stray item that belongs to a different
|
|
// subscription's MonitoredItemManager.
|
|
var strayContext = new FakeMonitoredItemContext();
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> strayOpts =
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>();
|
|
var stray = new TestMonitoredItem(strayContext, "stray",
|
|
strayOpts, m_telemetry.CreateLogger("stray"));
|
|
|
|
// Act + Assert: validation by reference identity
|
|
// rejects the stray item.
|
|
Assert.That(async () => await sut.SetTriggeringAsync(
|
|
trig!, [stray], null, default)
|
|
.ConfigureAwait(false),
|
|
Throws.ArgumentException);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SetTriggeringAsyncShouldQueueAndApplyAsync()
|
|
{
|
|
// Arrange — set up the SetTriggering RPC mock first so
|
|
// the apply pass succeeds when the queued op fires.
|
|
m_mockMonitoredItemServices
|
|
.Setup(s => s.SetTriggeringAsync(
|
|
It.IsAny<RequestHeader?>(), 22u, It.IsAny<uint>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1),
|
|
It.Is<ArrayOf<uint>>(r => r.Count == 0),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new SetTriggeringResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader(),
|
|
AddResults = new[] { StatusCodes.Good }.ToArrayOf(),
|
|
RemoveResults = Array.Empty<StatusCode>().ToArrayOf()
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
bool addedTrig = sut.MonitoredItems.TryAdd("trig",
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? trig);
|
|
bool addedTgt = sut.MonitoredItems.TryAdd("tgt",
|
|
OptionsFactory.Create<MonitoredItems.MonitoredItemOptions>(),
|
|
out IMonitoredItem? tgt);
|
|
Assert.That(addedTrig && addedTgt, Is.True);
|
|
Assert.That(trig!.Created, Is.True);
|
|
Assert.That(tgt!.Created, Is.True);
|
|
|
|
// Act
|
|
SetTriggeringResult result = await sut.SetTriggeringAsync(
|
|
trig,
|
|
[tgt],
|
|
null,
|
|
default).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(result, Is.Not.Null);
|
|
Assert.That(result.TriggeringItem, Is.SameAs(trig));
|
|
Assert.That(result.AddResults, Has.Count.EqualTo(1));
|
|
Assert.That(StatusCode.IsGood(result.AddResults[0].Status), Is.True);
|
|
Assert.That(result.RemoveResults, Is.Empty);
|
|
Assert.That(tgt.TriggeringItems, Contains.Item(trig));
|
|
m_mockMonitoredItemServices.Verify();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Regression coverage for OPCFoundation/UA-.NETStandard#3540.
|
|
/// The V2 engine's <c>OnStatusChangeNotificationAsync</c> must
|
|
/// always surface the
|
|
/// <see cref="PublishState.Transferred"/> flag via
|
|
/// <c>OnPublishStateChanged</c> so listeners can react —
|
|
/// previously the override was a no-op TODO that swallowed the
|
|
/// flag.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task GoodSubscriptionTransferredRaisesPublishStateTransferredAsync()
|
|
{
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.Created, Is.True);
|
|
sut.PublishStateChanged.Reset();
|
|
|
|
await sut.OnPublishReceivedAsync(BuildStatusChangeMessage(
|
|
sequenceNumber: 1,
|
|
StatusCodes.GoodSubscriptionTransferred), null, []).ConfigureAwait(false);
|
|
|
|
await sut.PublishStateChanged.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(sut.LastPublishState & PublishState.Transferred,
|
|
Is.EqualTo(PublishState.Transferred));
|
|
Assert.That(sut.Id, Is.EqualTo(22u),
|
|
"ReportOnly default must not recreate the subscription.");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Regression coverage for OPCFoundation/UA-.NETStandard#3540.
|
|
/// With <see cref="SubscriptionRecoveryPolicy.RecreateOnUnsolicitedTransfer"/>
|
|
/// the V2 engine must recreate the subscription in place on the
|
|
/// same session (driven through <see cref="ISubscription.RecreateAsync"/>)
|
|
/// so
|
|
/// the application keeps receiving data after a server-side
|
|
/// quirk emits Good_SubscriptionTransferred against a freshly
|
|
/// created subscription.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task GoodSubscriptionTransferredRecreatesSubscriptionWhenPolicyEnabledAsync()
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 33,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22,
|
|
SubscriptionRecoveryPolicy.RecreateOnUnsolicitedTransfer);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
Assert.That(sut.Created, Is.True);
|
|
Assert.That(sut.Id, Is.EqualTo(22u));
|
|
|
|
// Queue a stale ack for the doomed id so we can assert
|
|
// the ack-pruning step ran after successful retirement.
|
|
await m_completion.QueueAsync(new SubscriptionAcknowledgement
|
|
{
|
|
SubscriptionId = 22u,
|
|
SequenceNumber = 7u
|
|
}).ConfigureAwait(false);
|
|
|
|
await sut.OnPublishReceivedAsync(BuildStatusChangeMessage(
|
|
sequenceNumber: 1,
|
|
StatusCodes.GoodSubscriptionTransferred), null, []).ConfigureAwait(false);
|
|
|
|
await WaitForAsync(() => sut.Id == 33u, TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(sut.Id, Is.EqualTo(33u),
|
|
"Recovery must assign the new server-side id.");
|
|
Assert.That(m_completion.DroppedSubscriptions,
|
|
Does.Contain(22u),
|
|
"Ack-pruning must run for the retired subscription id.");
|
|
m_mockSubscriptionServices.Verify();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A burst of unsolicited Good_SubscriptionTransferred messages
|
|
/// must collapse into a single recreate via the
|
|
/// <c>m_recreateAfterTransferInProgress</c> guard — the
|
|
/// service set must see exactly one CreateSubscriptionAsync
|
|
/// call.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task GoodSubscriptionTransferredCollapsesConcurrentRecreatesAsync()
|
|
{
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<ArrayOf<uint>>(a => a.Count == 1 && a[0] == 22u),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse
|
|
{
|
|
Results = [StatusCodes.Good]
|
|
})
|
|
.Verifiable(Times.Once);
|
|
m_mockSubscriptionServices
|
|
.Setup(s => s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
TimeSpan.FromSeconds(100).TotalMilliseconds, 21, 7, 10, false,
|
|
3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 44,
|
|
RevisedLifetimeCount = 10,
|
|
RevisedMaxKeepAliveCount = 5,
|
|
RevisedPublishingInterval = 10000
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
var sut = new TestSubscription(m_session, m_mockNotificationDataHandler.Object,
|
|
m_completion, m_options, m_telemetry, 22,
|
|
SubscriptionRecoveryPolicy.RecreateOnUnsolicitedTransfer);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
for (uint i = 1; i <= 3; i++)
|
|
{
|
|
await sut.OnPublishReceivedAsync(BuildStatusChangeMessage(
|
|
sequenceNumber: i,
|
|
StatusCodes.GoodSubscriptionTransferred), null, [])
|
|
.ConfigureAwait(false);
|
|
}
|
|
|
|
await WaitForAsync(() => sut.Id == 44u, TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
await Task.Delay(300).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.Id, Is.EqualTo(44u));
|
|
m_mockSubscriptionServices.Verify(s =>
|
|
s.CreateSubscriptionAsync(It.IsAny<RequestHeader>(),
|
|
It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once,
|
|
"Idempotency guard must prevent duplicate recreates from the burst.");
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task BadTimeoutRecoveryRecreatesSubscriptionAndAcceptsRestartedSequenceAsync()
|
|
{
|
|
var created = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var delivered = new TaskCompletionSource<uint>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
m_mockSubscriptionServices
|
|
.Setup(service => service.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), TimeSpan.FromSeconds(100).TotalMilliseconds,
|
|
21, 7, 10, false, 3, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new CreateSubscriptionResponse
|
|
{
|
|
SubscriptionId = 33,
|
|
RevisedLifetimeCount = 21,
|
|
RevisedMaxKeepAliveCount = 7,
|
|
RevisedPublishingInterval = TimeSpan.FromSeconds(100).TotalMilliseconds
|
|
});
|
|
m_mockSubscriptionServices
|
|
.Setup(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new DeleteSubscriptionsResponse { Results = [StatusCodes.Good] });
|
|
|
|
var sut = new TestSubscription(
|
|
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry, 22);
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.SetMessageStateForTest(4999, 4999, []);
|
|
sut.OnAfterCreateAsync = _ =>
|
|
{
|
|
created.TrySetResult(true);
|
|
return default;
|
|
};
|
|
sut.OnDataChangeAsync = sequenceNumber =>
|
|
{
|
|
delivered.TrySetResult(sequenceNumber);
|
|
return default;
|
|
};
|
|
|
|
await sut.OnPublishReceivedAsync(
|
|
BuildStatusChangeMessage(5000, StatusCodes.BadTimeout), null, []).ConfigureAwait(false);
|
|
await created.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await m_completion.WaitForQueuedAckAsync(1).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.Id, Is.EqualTo(33u));
|
|
Assert.That(sut.LastSequenceNumberForTest, Is.Zero);
|
|
Assert.That(sut.LastDataSequenceNumberForTest, Is.Zero);
|
|
Assert.That(m_completion.QueuedAcks[0].SubscriptionId, Is.EqualTo(22u));
|
|
Assert.That(m_completion.QueuedAcks[0].SequenceNumber, Is.EqualTo(5000u));
|
|
|
|
await sut.OnPublishReceivedAsync(BuildDataMessage(1), null, []).ConfigureAwait(false);
|
|
Assert.That(await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false),
|
|
Is.EqualTo(1u));
|
|
await m_completion.WaitForQueuedAckAsync(2).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.LastDataSequenceNumberForTest, Is.EqualTo(1u));
|
|
Assert.That(m_completion.QueuedAcks, Has.Count.EqualTo(2));
|
|
Assert.That(m_completion.QueuedAcks[1].SubscriptionId, Is.EqualTo(33u));
|
|
Assert.That(m_completion.QueuedAcks[1].SequenceNumber, Is.EqualTo(1u));
|
|
m_mockSubscriptionServices.Verify(service => service.CreateSubscriptionAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(), It.IsAny<uint>(),
|
|
It.IsAny<uint>(), It.IsAny<bool>(), It.IsAny<byte>(), It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
m_mockSubscriptionServices.Verify(service => service.DeleteSubscriptionsAsync(
|
|
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
}
|
|
}
|
|
|
|
private static NotificationMessage BuildStatusChangeMessage(
|
|
uint sequenceNumber, StatusCode status)
|
|
{
|
|
return new NotificationMessage
|
|
{
|
|
SequenceNumber = sequenceNumber,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new StatusChangeNotification
|
|
{
|
|
Status = status
|
|
})
|
|
]
|
|
};
|
|
}
|
|
|
|
private static NotificationMessage BuildDataMessage(uint sequenceNumber)
|
|
{
|
|
return new NotificationMessage
|
|
{
|
|
SequenceNumber = sequenceNumber,
|
|
NotificationData = [new ExtensionObject(new DataChangeNotification())]
|
|
};
|
|
}
|
|
|
|
private static NotificationMessage BuildKeepAliveMessage(uint sequenceNumber)
|
|
{
|
|
return new NotificationMessage
|
|
{
|
|
SequenceNumber = sequenceNumber,
|
|
NotificationData = []
|
|
};
|
|
}
|
|
|
|
private static async Task WaitForAsync(
|
|
Func<bool> predicate, TimeSpan timeout)
|
|
{
|
|
DateTime deadline = DateTime.UtcNow + timeout;
|
|
while (!predicate() && DateTime.UtcNow < deadline)
|
|
{
|
|
await Task.Delay(25).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private sealed class CleanupTimeProvider : FakeTimeProvider
|
|
{
|
|
public Task<TimeSpan> CleanupScheduled => m_cleanupScheduled.Task;
|
|
|
|
public override ITimer CreateTimer(
|
|
TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period)
|
|
{
|
|
ITimer timer = base.CreateTimer(callback, state, dueTime, period);
|
|
if (dueTime != Timeout.InfiniteTimeSpan && period == Timeout.InfiniteTimeSpan)
|
|
{
|
|
m_cleanupScheduled.TrySetResult(dueTime);
|
|
}
|
|
return timer;
|
|
}
|
|
|
|
private readonly TaskCompletionSource<TimeSpan> m_cleanupScheduled =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
}
|
|
|
|
private sealed class TestMonitoredItem : MonitoredItems.MonitoredItem
|
|
{
|
|
public static MonitoredItems.MonitoredItemOptions CreatedOptions => new()
|
|
{
|
|
StartNodeId = NodeId.Parse("ns=2;s=Demo"),
|
|
MonitoringMode = MonitoringMode.Sampling,
|
|
SamplingInterval = TimeSpan.FromSeconds(1),
|
|
QueueSize = 5,
|
|
TimestampsToReturn = TimestampsToReturn.Both,
|
|
AttributeId = Attributes.Value,
|
|
DiscardOldest = true
|
|
};
|
|
|
|
public TestMonitoredItem(IMonitoredItemContext subscription, string name,
|
|
OptionsMonitor<MonitoredItems.MonitoredItemOptions> options, ILogger logger)
|
|
: base(subscription, name, options, logger)
|
|
{
|
|
if (options.CurrentValue.StartNodeId.IsNull)
|
|
{
|
|
// Auto create
|
|
options.Configure(o => CreatedOptions);
|
|
Assert.That(TryGetPendingChange(out Change? change), Is.True);
|
|
Assert.That(change, Is.Not.Null);
|
|
change.SetCreateResult(new MonitoredItemCreateRequest
|
|
{
|
|
ItemToMonitor = new ReadValueId
|
|
{
|
|
AttributeId = Options.CurrentValue.AttributeId,
|
|
NodeId = Options.CurrentValue.StartNodeId
|
|
},
|
|
MonitoringMode = MonitoringMode.Sampling,
|
|
RequestedParameters = new MonitoringParameters
|
|
{
|
|
ClientHandle = ClientHandle,
|
|
SamplingInterval = Options.CurrentValue.SamplingInterval.TotalMilliseconds,
|
|
QueueSize = Options.CurrentValue.QueueSize,
|
|
DiscardOldest = Options.CurrentValue.DiscardOldest
|
|
}
|
|
},
|
|
new MonitoredItemCreateResult
|
|
{
|
|
StatusCode = StatusCodes.Good,
|
|
MonitoredItemId = ClientHandle,
|
|
RevisedSamplingInterval = 10000,
|
|
RevisedQueueSize = 10
|
|
}, 0, [], new ResponseHeader());
|
|
}
|
|
}
|
|
|
|
public int DisposeCalls { get; private set; }
|
|
|
|
protected override async ValueTask DisposeAsync(bool disposing)
|
|
{
|
|
await base.DisposeAsync(disposing).ConfigureAwait(false);
|
|
if (disposing)
|
|
{
|
|
DisposeCalls++;
|
|
}
|
|
}
|
|
}
|
|
|
|
private sealed class TestSubscription : Subscription
|
|
{
|
|
public static SubscriptionOptions SubscriptionOptions => new()
|
|
{
|
|
Disabled = false,
|
|
PublishingEnabled = false,
|
|
PublishingInterval = TimeSpan.FromSeconds(100),
|
|
KeepAliveCount = 7,
|
|
LifetimeCount = 15,
|
|
Priority = 3,
|
|
MaxNotificationsPerPublish = 10
|
|
};
|
|
|
|
public TestSubscription(ISubscriptionContext session, ISubscriptionNotificationHandler handler,
|
|
IMessageAckQueue completion, OptionsMonitor<SubscriptionOptions> options,
|
|
ITelemetryContext telemetry, uint? subscriptionIdForAlreadyCreatedState = null,
|
|
SubscriptionRecoveryPolicy recoveryPolicy = SubscriptionRecoveryPolicy.ReportOnly,
|
|
TimeProvider? timeProvider = null)
|
|
: base(session, handler, completion, !subscriptionIdForAlreadyCreatedState.HasValue ?
|
|
options : options.Configure(o => o with { Disabled = true }), telemetry, timeProvider: timeProvider)
|
|
{
|
|
// Let the subscription create itself
|
|
if (subscriptionIdForAlreadyCreatedState.HasValue)
|
|
{
|
|
Options = SubscriptionOptions with { RecoveryPolicy = recoveryPolicy };
|
|
OnSubscriptionUpdateComplete(true, subscriptionIdForAlreadyCreatedState.Value,
|
|
TimeSpan.FromSeconds(100), 7, 21, 3, 10, false);
|
|
// Now we have a created subscription
|
|
}
|
|
}
|
|
|
|
public SemaphoreSlim Block { get; } = new(0, 1);
|
|
public AsyncManualResetEvent SubscriptionStateChanged { get; } = new();
|
|
public AsyncManualResetEvent PublishStateChanged { get; } = new();
|
|
public PublishState LastPublishState { get; private set; }
|
|
public uint LastSequenceNumberForTest => LastSequenceNumberProcessed;
|
|
public uint LastDataSequenceNumberForTest => LastDataSequenceNumberProcessed;
|
|
|
|
public IReadOnlyList<uint> AvailableSequenceNumbersForTest
|
|
=> AvailableInRetransmissionQueue;
|
|
|
|
public int KeepAliveNotificationCount
|
|
=> Volatile.Read(ref m_keepAliveNotificationCount);
|
|
|
|
public IReadOnlyList<uint> KeepAliveSequenceNumbers
|
|
=> [.. m_keepAliveSequenceNumbers];
|
|
|
|
public Func<ValueTask>? OnKeepAliveAsync { get; set; }
|
|
public Func<uint, ValueTask>? OnDataChangeAsync { get; set; }
|
|
|
|
public void SetMessageStateForTest(uint lastSequenceNumber,
|
|
uint lastDataSequenceNumber,
|
|
IReadOnlyList<uint> availableSequenceNumbers)
|
|
{
|
|
LastSequenceNumberProcessed = lastSequenceNumber;
|
|
LastDataSequenceNumberProcessed = lastDataSequenceNumber;
|
|
AvailableInRetransmissionQueue = availableSequenceNumbers;
|
|
}
|
|
|
|
public async ValueTask WaitAsync()
|
|
{
|
|
await Block.WaitAsync().ConfigureAwait(false);
|
|
Block.Release();
|
|
}
|
|
|
|
public void ResetPublishStateForTest()
|
|
{
|
|
LastPublishState = default;
|
|
PublishStateChanged.Reset();
|
|
}
|
|
|
|
protected override void OnSubscriptionStateChanged(SubscriptionState state)
|
|
{
|
|
if (state != SubscriptionState.Opened)
|
|
{
|
|
SubscriptionStateChanged.Set();
|
|
}
|
|
base.OnSubscriptionStateChanged(state);
|
|
}
|
|
|
|
protected override void OnPublishStateChanged(PublishState stateMask)
|
|
{
|
|
LastPublishState = stateMask;
|
|
PublishStateChanged.Set();
|
|
base.OnPublishStateChanged(stateMask);
|
|
}
|
|
|
|
protected override MonitoredItems.MonitoredItem CreateMonitoredItem(string name,
|
|
IOptionsMonitor<MonitoredItems.MonitoredItemOptions> options, IMonitoredItemContext context,
|
|
ITelemetryContext telemetry)
|
|
{
|
|
return new TestMonitoredItem(context, name,
|
|
(OptionsMonitor<MonitoredItems.MonitoredItemOptions>)options,
|
|
telemetry.CreateLogger("TestMonitoredItem"));
|
|
}
|
|
|
|
protected override ValueTask OnKeepAliveNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, PublishState publishStateMask)
|
|
{
|
|
m_keepAliveSequenceNumbers.Enqueue(sequenceNumber);
|
|
Interlocked.Increment(ref m_keepAliveNotificationCount);
|
|
return OnKeepAliveAsync?.Invoke() ?? default;
|
|
}
|
|
|
|
protected override ValueTask OnDataChangeNotificationAsync(
|
|
uint sequenceNumber,
|
|
DateTime publishTime,
|
|
DataChangeNotification notification,
|
|
PublishState publishStateMask,
|
|
IReadOnlyList<string> stringTable)
|
|
{
|
|
return OnDataChangeAsync != null
|
|
? OnDataChangeAsync(sequenceNumber)
|
|
: base.OnDataChangeNotificationAsync(
|
|
sequenceNumber, publishTime, notification, publishStateMask, stringTable);
|
|
}
|
|
|
|
private readonly ConcurrentQueue<uint> m_keepAliveSequenceNumbers = new();
|
|
private int m_keepAliveNotificationCount;
|
|
}
|
|
|
|
private static readonly string[] s_recreateOperationOrder =
|
|
[
|
|
"subscription", "durable", "items",
|
|
"subscription", "durable", "items"
|
|
];
|
|
|
|
private FakeMessageAckQueue m_completion;
|
|
private OptionsMonitor<SubscriptionOptions> m_options;
|
|
private ITelemetryContext m_telemetry;
|
|
private Mock<ISubscriptionServiceSetClientMethods> m_mockSubscriptionServices;
|
|
private Mock<IMonitoredItemServiceSetClientMethods> m_mockMonitoredItemServices;
|
|
private Mock<IMethodServiceSetClientMethods> m_mockMethodServices;
|
|
private FakeSubscriptionContext m_session;
|
|
private Mock<ISubscriptionNotificationHandler> m_mockNotificationDataHandler;
|
|
}
|
|
}
|