mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 15:11:53 +08:00
# Description Fixes client subscription acknowledgement defects and replaces the mutable event-based acknowledgement policy with an injectable, asynchronous provider model. The latest changes in `7be912614` add opt-in **whole-message delivery**, harden lifecycle/concurrency behavior, and simplify acknowledgement state and wire-array construction. Existing features and all deprecations remain in this PR. The acknowledgement unit is one OPC UA `NotificationMessage`, not one data change, event, downstream message, or group of Publish responses indicated by `MoreNotifications`. ### Protocol and lifecycle fixes - Ensure acknowledgements make progress when fewer acknowledgements are queued than there are Publish workers. - Honor legacy deferral on V2 and preserve deferred/in-flight acknowledgements across retries. A throwing legacy callback restores the original lists; a throwing provider defers the batch. - Validate ownership before sending and after provider decisions, including subscription removal, recreation, changed IDs and server ID reuse. Reject detected responses from the previous owner before they advance the replacement subscription's sequence cursor. - Prune acknowledgements no longer retained by the server, including servers without a retransmission queue. - Retire one subscription's acknowledgement state without quiescing publishing for the whole session. - Reserve the classic engine's provider-decision slot before invoking application code. Concurrent Publish calls cannot overwrite pending decisions; asynchronous failures are observed and logged. - Preserve the original outstanding lifecycle when a cancelled queue attempt can still be queued. Retries do not create an additional terminal outcome. - Retire internal confirmation state synchronously, independently of application observer progress. Observer callbacks remain asynchronous and outside SDK locks. ### Public API and whole-message delivery | API | Purpose | |---|---| | `AcknowledgementHandle` | Process-local acknowledgement identity scoped to a subscription lifetime; avoids correlating messages solely by subscription ID and sequence number. | | `ISubscriptionAcknowledgementProvider` | Asynchronous, fail-closed send/defer decisions over `ArrayOf<AcknowledgementHandle>`. | | `ISubscriptionAcknowledgementObserver` | Terminal outcomes: `Acknowledged`, `NotRetained`, `OwnerRetired`, or `Abandoned`, subject to the cross-session limitation below. | | `AcknowledgementMode.AfterConfirmation` / `ISessionAcknowledgements.Confirm` | Do not acknowledge until the application explicitly confirms processing of the complete message. | | `ISubscriptionNotificationMessageHandler` / `SubscriptionNotificationMessage` | Opt-in delivery of all typed payloads in one envelope, preserving payload order, physical-partition identity, publish time and diagnostic string table. | - Provider configuration is available through `ManagedSessionOptions`, fluent builder methods and DI registration. - Whole-message handlers are registered through the V2 subscription APIs or `Subscription.NotificationMessageHandler` on the classic engine. Existing load/restore factories can return handlers implementing both notification interfaces. - Whole-message handling replaces data/event callbacks only for the opted-in subscription; keep-alive and lifecycle handling remain available. Returning from the handler does not implicitly confirm delivery. - Projection arrays do not borrow pooled wire notification containers. Monitored-item references remain live SDK objects. - Whole-message handlers receive shutdown cancellation before subscription retirement waits for their dispatch to finish. - Managed model-change tracking owns a dedicated whole-message subscription instead of sharing application streaming acknowledgement ownership. It confirms only after all events succeed; namespace-refresh, cache-invalidation and subscriber failures remain logged but do not produce a success confirmation. ### Compatibility and deprecations The following deprecations remain, with compatibility behavior preserved: - `ISession.PublishSequenceNumbersToAcknowledge`: honored when no provider is configured; not raised when a provider owns the policy. - Classic `FastDataChangeCallback`, `FastEventCallback`, `FastKeepAliveCallback` and their delegate types. - `ISession.DefaultSubscription` and its implementations. The lifetime proposal from the closed PR #4901 is incorporated through the provider/handle model rather than additional lifecycle events. ### Simplification and performance scope - Pending acknowledgements derive owner and generation from their scope rather than storing redundant copies. - Wire acknowledgements are projected directly into the final array, eliminating the intermediate converted list and its extra copy. - Whole-message envelopes are constructed only for opted-in handlers. The existing payload-callback path remains available. - Handles reuse cached subscription-lifetime scopes. Provider callbacks execute outside SDK locks. This is not a benchmarked throughput or allocation-parity claim. Broader lock/contention work, provider-path allocations and benchmark evidence remain tracked by #4912. ### Known limitations and non-goals - **#4929 remains open:** classic subscriptions served by the default V2 compatibility bridge do not yet participate in normal acknowledgement tracking. Supporting the classic engine is not the same as fixing that bridge. - **#4930 remains open:** transferring a classic subscription between different `Session` objects does not yet move tracked acknowledgements/confirmations between engines. Duplicate terminal reports or lost confirmations remain possible on that path. In-place managed-session recreation is a different, supported path. - Delaying protocol acknowledgement is not itself downstream retry or durable storage. Applications still own downstream retry/buffering; server retransmission queues are finite. - External observer delivery is buffered. Slow observers can grow their notification backlog, although they no longer retain internal confirmation state. Disposal waits up to five seconds for callbacks; an unresponsive observer can outlive disposal. - HA takeover/session-continuation redesign and server-side acknowledgement defects are not addressed here. Documentation: [Subscriptions](docs/Subscriptions.md), [Migration guide](docs/MigrationGuide.md), and [Model-change tracking](docs/ModelChangeTracking.md). ### Validation Local validation of the changes included in `7be912614`: | Check | Result | |---|---| | Expanded net10 client regression selection | **389 passed, 0 failed, 0 skipped** | | Final API-registration and whole-message-routing selection | **3 passed**; two overlap the regression selection, one is additional (**390 distinct client cases across the runs**) | | Acknowledgement integration tests | **4 passed**: classic/V2, each with legacy payload callbacks and whole-message handling | | `Opc.Ua.Client` net48 build | **0 warnings, 0 errors** | | Scoped whitespace, style and analyzer verification | Passed, including informational severity for style/analyzers | | Negative/revert controls | Tests reproduced the provider race, missing whole-message routing, observer-dependent cleanup, cancelled-queue lifecycle loss, shutdown deadlock and false model-change confirmation; corrections were restored and positive cases passed | <details> <summary>Exact local test/build commands and results</summary> Run from the repository root. Framework-specific assets must be restored before commands using `--no-restore`; perform the net48 build last because changing `CustomTestTarget` replaces restored assets. ```powershell dotnet test tests\Opc.Ua.Client.Tests\Opc.Ua.Client.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~AcknowledgementProviderTests|FullyQualifiedName~SubscriptionManagerTests|FullyQualifiedName~ClassicSubscriptionEngineTests|FullyQualifiedName~ManagedSessionAcknowledgementForwardingTests|FullyQualifiedName~AcknowledgementOptionsTests|FullyQualifiedName~SubscriptionTests|FullyQualifiedName~ModelChangeTrackerTests|FullyQualifiedName~PartitionForwardingHandlerTests|FullyQualifiedName~SessionEngineContextTests|FullyQualifiedName~DefaultSubscriptionEngineTests|FullyQualifiedName~MonitoredItemManagerTests|FullyQualifiedName~MessageProcessorTests" --logger "console;verbosity=minimal" -v quiet -nologo # Passed: 389; failed: 0; skipped: 0. dotnet test tests\Opc.Ua.Client.Tests\Opc.Ua.Client.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~WholeMessageRegistrationRejectsMissingInputs|FullyQualifiedName~WholeMessageDeliveryPreservesMixedPayloadsAsync" --logger "console;verbosity=minimal" -v quiet -nologo # Passed: 3; failed: 0; skipped: 0. Two cases overlap the previous selection. dotnet test tests\Opc.Ua.Subscriptions.Tests\Opc.Ua.Subscriptions.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~AcknowledgementProviderIntegrationTests" --logger "console;verbosity=minimal" -v quiet -nologo # Passed: 4; failed: 0; skipped: 0. dotnet build src\Opc.Ua.Client\Opc.Ua.Client.csproj -p:CustomTestTarget=net48 -p:RestoreSources=https://www.nuget.org/api/v2/ -p:NuGetAudit=false -v quiet -nologo # Build succeeded: 0 warnings, 0 errors. ``` The v2 feed and disabled NuGet audit above were local restore workarounds only; repository settings were not changed. CI dependency auditing remains required. </details> The full `UA.slnx` net48/net10 test matrix, NativeAOT publishing/runtime validation, fresh coverage measurements and performance benchmarks were **not run locally for this revision**. CI and maintainer API review remain outstanding. Historical results on `a28c12252` are not presented as validation of the latest head. ## Related Issues _If this is a relatively large or complex change, a design must have been discussed in the related tracking issue and signed off (which becomes the Architectural Decision Record (ADR))._ - Fixes #4894: acknowledgement progress, deferral/retry behavior and ownership validation. - Fixes #4898: injectable fail-closed decision API, with legacy precedence preserved. - Fixes #4913: admission of detected old-owner responses after subscription ID reuse. Non-closing references: | Issue | Title / scope | Disposition | |---|---|---| | [#4897](https://github.com/OPCFoundation/UA-.NETStandard/issues/4897) | Stable notification identity and terminal acknowledgement state | Core functionality is implemented via handles/providers, with whole-message delivery added. Keep this link non-closing until the cross-session exception in #4930 is resolved or the issue's accepted scope is explicitly narrowed. | | [#4912](https://github.com/OPCFoundation/UA-.NETStandard/issues/4912) | Reduce acknowledgement-path lock contention and allocations | Partially advanced by state/projection simplification; broader locking changes and benchmarks remain. | | [#4929](https://github.com/OPCFoundation/UA-.NETStandard/issues/4929) | Classic subscriptions on the V2 compatibility bridge | Not fixed; separate follow-up. | | [#4930](https://github.com/OPCFoundation/UA-.NETStandard/issues/4930) | Cross-session classic acknowledgement handover | Not fixed; separate follow-up and explicit limitation on terminal-outcome guarantees. | Design rationale: [maintainer request for injectable providers and session facets](https://github.com/OPCFoundation/UA-.NETStandard/pull/4899#discussion_r4366515418). The new whole-message API is submitted for maintainer review; this description does not claim that its final API shape or an ADR has been signed off. ## Checklist - [x] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [x] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [x] I have added all necessary documentation. - [x] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] 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. Regression tests were added and passed; the combined tests/coverage checkbox is left open because a fresh coverage increase has not yet been verified. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Marc Schier <marcschier@users.noreply.github.com> Co-authored-by: Marc <marcschier@hotmail.com>
824 lines
34 KiB
C#
824 lines
34 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/
|
|
* ======================================================================*/
|
|
|
|
// CA2000: test code; the hand-rolled managed-subscription stubs are
|
|
// IAsyncDisposable and their ownership is transferred to the manager under
|
|
// test (it disposes them on DisposeAsync). CA2000 cannot see through that
|
|
// transfer and would be noisy without a real leak risk. Disabled file-level.
|
|
#pragma warning disable CA2000
|
|
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client.Subscriptions;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
|
|
namespace Opc.Ua.Subscriptions.Tests
|
|
{
|
|
/// <summary>
|
|
/// Deterministic, fully offline unit tests for the internal
|
|
/// <see cref="SubscriptionManager"/>. The background publish
|
|
/// controller launched in the constructor is kept quiescent: the
|
|
/// stub context never marks a subscription as <c>Created</c>, so
|
|
/// <c>GetDesiredPublishWorkerCount</c> returns 0 and no publish
|
|
/// worker is ever spawned. Every test asserts on concrete values
|
|
/// and never depends on wall-clock timing.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[Category("SubscriptionManager")]
|
|
[Category("SubscriptionManagerUnit")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
public sealed class SubscriptionManagerCoverageTests
|
|
{
|
|
private static SubscriptionManager CreateManager(
|
|
StubSubscriptionManagerContext context,
|
|
DiagnosticsMasks diagnostics = DiagnosticsMasks.None)
|
|
{
|
|
return new SubscriptionManager(
|
|
context, NullLoggerFactory.Instance, diagnostics);
|
|
}
|
|
|
|
private static OptionsMonitor<SubscriptionOptions> SinglePartitionOptions()
|
|
{
|
|
// DisableUnboundedItemMode selects the inert single-partition
|
|
// wrapper path in Add (no placement policy, factory or idle
|
|
// timer), keeping the manager fully deterministic.
|
|
return new OptionsMonitor<SubscriptionOptions>(
|
|
new SubscriptionOptions { DisableUnboundedItemMode = true });
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConstructorInitializesQuiescentDefaultsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.Count, Is.Zero);
|
|
Assert.That(manager.Items, Is.Empty);
|
|
Assert.That(manager.PublishWorkerCount, Is.Zero);
|
|
Assert.That(manager.GoodPublishRequestCount, Is.Zero);
|
|
Assert.That(manager.BadPublishRequestCount, Is.Zero);
|
|
Assert.That(manager.MissingMessageCount, Is.Zero);
|
|
Assert.That(manager.RepublishMessageCount, Is.Zero);
|
|
Assert.That(manager.CreatedCount, Is.Zero);
|
|
Assert.That(manager.Created, Is.Empty);
|
|
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
|
|
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
|
|
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
|
|
Assert.That(manager.PoolNotifications, Is.False);
|
|
}
|
|
// The publish controller stayed parked the whole time.
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConstructorStoresReturnDiagnosticsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context, DiagnosticsMasks.All);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.All));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task MinPublishWorkerCountHonoursChangedAndEqualBranchesAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
|
|
|
|
// Equal-value early-return branch: value must be unchanged.
|
|
manager.MinPublishWorkerCount = 2;
|
|
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
|
|
|
|
// Changed-value branch: value updated and controller signalled.
|
|
manager.MinPublishWorkerCount = 7;
|
|
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(7));
|
|
|
|
// Equal-value branch again on the new value.
|
|
manager.MinPublishWorkerCount = 7;
|
|
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(7));
|
|
}
|
|
// No subscription is Created, so signalling never spawns a worker.
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task MaxPublishWorkerCountHonoursChangedAndEqualBranchesAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
|
|
|
|
// Equal-value early-return branch.
|
|
manager.MaxPublishWorkerCount = 15;
|
|
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
|
|
|
|
// Changed-value branch.
|
|
manager.MaxPublishWorkerCount = 30;
|
|
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(30));
|
|
|
|
// Equal-value branch again.
|
|
manager.MaxPublishWorkerCount = 30;
|
|
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(30));
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task TransferSubscriptionsOnRecreateRoundTripsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
|
|
manager.TransferSubscriptionsOnRecreate = true;
|
|
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.True);
|
|
manager.TransferSubscriptionsOnRecreate = false;
|
|
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task PoolNotificationsRoundTripsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
// PoolNotifications is a single public auto-property that
|
|
// implicitly implements both ISubscriptionManager and
|
|
// IMessageAckQueue, so the manager surface is authoritative.
|
|
Assert.That(manager.PoolNotifications, Is.False);
|
|
|
|
manager.PoolNotifications = true;
|
|
Assert.That(manager.PoolNotifications, Is.True);
|
|
|
|
manager.PoolNotifications = false;
|
|
Assert.That(manager.PoolNotifications, Is.False);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReturnDiagnosticsRoundTripsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context, DiagnosticsMasks.None);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.None));
|
|
|
|
manager.ReturnDiagnostics = DiagnosticsMasks.All;
|
|
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.All));
|
|
|
|
manager.ReturnDiagnostics = DiagnosticsMasks.OperationAll;
|
|
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.OperationAll));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task QueueAsyncThenDropPendingIsIdempotentAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
var first = new StubManagedSubscription { Id = 1u };
|
|
var second = new StubManagedSubscription { Id = 2u };
|
|
await manager.QueueAsync(first, first.MessageGeneration, new SubscriptionAcknowledgement
|
|
{
|
|
SubscriptionId = 1u,
|
|
SequenceNumber = 10u
|
|
}, CancellationToken.None).ConfigureAwait(false);
|
|
await manager.QueueAsync(second, second.MessageGeneration, new SubscriptionAcknowledgement
|
|
{
|
|
SubscriptionId = 2u,
|
|
SequenceNumber = 11u
|
|
}, CancellationToken.None).ConfigureAwait(false);
|
|
|
|
// Queued entries are validated when sent; only deferred ones are removed here.
|
|
Assert.That(manager.DropPendingForSubscription(first, 1u), Is.Zero);
|
|
Assert.That(manager.DropPendingForSubscription(first, 1u), Is.Zero);
|
|
Assert.That(manager.QueuedAcknowledgementCount, Is.EqualTo(2));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DropPendingForSubscriptionOnEmptyQueueReturnsZeroAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Assert.That(manager.DropPendingForSubscription(new StubManagedSubscription { Id = 7u }, 7u),
|
|
Is.Zero);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task CompleteAsyncUnknownSubscriptionReturnsWithoutChangeAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
await manager.CompleteAsync(new StubManagedSubscription { Id = 999u }, 999u,
|
|
CancellationToken.None).ConfigureAwait(false);
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task UpdateSignalsControllerWithoutThrowingAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
manager.Update();
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DrainAsyncCompletesImmediatelyWhenNoPublishActiveAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
Task drain = manager.DrainAsync(CancellationToken.None);
|
|
Assert.That(drain.IsCompleted, Is.True);
|
|
await drain.ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ResumeWithoutSubscriptionsDoesNotThrowAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
manager.Resume();
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task PauseWithoutSubscriptionsDoesNotThrowAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
manager.Pause();
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateSubscriptionsAsyncWithNoSubscriptionsReturnsImmediatelyAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
await manager.RecreateSubscriptionsAsync(null, CancellationToken.None)
|
|
.ConfigureAwait(false);
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncIsIdempotentAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
// Second dispose must be a no-op and must not throw.
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(manager.Count, Is.Zero);
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task AddRegistersSubscriptionAndReturnsWrapperAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
ISubscription result = manager.Add(handler, SinglePartitionOptions());
|
|
|
|
Assert.That(result, Is.Not.Null);
|
|
Assert.That(manager.Count, Is.EqualTo(1));
|
|
Assert.That(manager.Items, Has.Exactly(1).Items);
|
|
Assert.That(context.CreateSubscriptionCallCount, Is.EqualTo(1));
|
|
Assert.That(context.LastCreateQueue, Is.SameAs(manager));
|
|
Assert.That(manager.PublishWorkerCount, Is.Zero);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task AddWithNullHandlerThrowsArgumentNullExceptionAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
ArgumentNullException ex = Assert.Throws<ArgumentNullException>(
|
|
() => manager.Add(null!, SinglePartitionOptions()));
|
|
Assert.That(ex.ParamName, Is.EqualTo("handler"));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task AddWithNullOptionsThrowsArgumentNullExceptionAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
ArgumentNullException ex = Assert.Throws<ArgumentNullException>(
|
|
() => manager.Add(handler, null!));
|
|
Assert.That(ex.ParamName, Is.EqualTo("options"));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task AddTwoSubscriptionsCountAndItemsReflectBothAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscriptions = new Queue<IManagedSubscription>(
|
|
[
|
|
new StubManagedSubscription { Id = 1u },
|
|
new StubManagedSubscription { Id = 2u }
|
|
]);
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
Assert.That(manager.Count, Is.EqualTo(2));
|
|
Assert.That(manager.Items, Has.Exactly(2).Items);
|
|
Assert.That(context.CreateSubscriptionCallCount, Is.EqualTo(2));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task MissingMessageCountAggregatesAcrossSubscriptionsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscriptions = new Queue<IManagedSubscription>(
|
|
[
|
|
new StubManagedSubscription { Id = 1u, MissingMessageCount = 3 },
|
|
new StubManagedSubscription { Id = 2u, MissingMessageCount = 4 }
|
|
]);
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
Assert.That(manager.MissingMessageCount, Is.EqualTo(7));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RepublishMessageCountAggregatesAcrossSubscriptionsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscriptions = new Queue<IManagedSubscription>(
|
|
[
|
|
new StubManagedSubscription { Id = 1u, RepublishMessageCount = 2 },
|
|
new StubManagedSubscription { Id = 2u, RepublishMessageCount = 5 }
|
|
]);
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
Assert.That(manager.RepublishMessageCount, Is.EqualTo(7));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task CreatedCountIsZeroWhenSubscriptionsAreNotCreatedAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u, Created = false };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
Assert.That(manager.Count, Is.EqualTo(1));
|
|
Assert.That(manager.CreatedCount, Is.Zero);
|
|
Assert.That(manager.Created, Is.Empty);
|
|
Assert.That(manager.PublishWorkerCount, Is.Zero);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task CompleteAsyncRemovesKnownSubscriptionAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
Assert.That(manager.Count, Is.EqualTo(1));
|
|
|
|
await manager.CompleteAsync(subscription, 1u, CancellationToken.None).ConfigureAwait(false);
|
|
|
|
Assert.That(manager.Count, Is.Zero);
|
|
Assert.That(manager.Items, Is.Empty);
|
|
// CompleteAsync only unregisters; it does not dispose the partition.
|
|
Assert.That(subscription.DisposeAsyncCalls, Is.Zero);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncDisposesRegisteredSubscriptionsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
Assert.That(manager.Count, Is.EqualTo(1));
|
|
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(subscription.DisposeAsyncCalls, Is.EqualTo(1));
|
|
Assert.That(manager.Count, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task RecreateSubscriptionsAsyncInvokesRecreateOnEachSubscriptionAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u, Created = false };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
// previousSessionId == null and TransferSubscriptionsOnRecreate
|
|
// defaults to false, so no server transfer is attempted; every
|
|
// subscription is recreated directly.
|
|
await manager.RecreateSubscriptionsAsync(null, CancellationToken.None)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(subscription.RecreateAsyncCalls, Is.EqualTo(1));
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task ResumeAndPauseNotifyRegisteredSubscriptionsAsync()
|
|
{
|
|
var context = new StubSubscriptionManagerContext();
|
|
var handler = new NoopNotificationHandler();
|
|
var subscription = new StubManagedSubscription { Id = 1u };
|
|
SubscriptionManager manager = CreateManager(context);
|
|
await using (manager.ConfigureAwait(false))
|
|
{
|
|
context.CreateSubscriptionFactory = (_, _, _) => subscription;
|
|
|
|
manager.Add(handler, SinglePartitionOptions());
|
|
|
|
manager.Resume();
|
|
manager.Pause();
|
|
|
|
// Resume notifies with paused=false, then Pause with true.
|
|
Assert.That(subscription.PausedCalls, Has.Count.EqualTo(2));
|
|
Assert.That(subscription.PausedCalls[0], Is.False);
|
|
Assert.That(subscription.PausedCalls[1], Is.True);
|
|
}
|
|
Assert.That(context.PublishCallCount, Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Hand-rolled stub for <see cref="ISubscriptionManagerContext"/>.
|
|
/// Keeps the publish controller quiescent: it never returns publish
|
|
/// work and records that <see cref="PublishAsync"/> was not reached.
|
|
/// </summary>
|
|
private sealed class StubSubscriptionManagerContext : ISubscriptionManagerContext
|
|
{
|
|
public int SessionSubscriptionCount => 0;
|
|
|
|
public int CreateSubscriptionCallCount { get; private set; }
|
|
|
|
public IMessageAckQueue? LastCreateQueue { get; private set; }
|
|
|
|
public int PublishCallCount => Volatile.Read(ref m_publishCallCount);
|
|
|
|
public Func<ISubscriptionNotificationHandler,
|
|
IOptionsMonitor<SubscriptionOptions>, IMessageAckQueue,
|
|
IManagedSubscription>? CreateSubscriptionFactory
|
|
{ get; set; }
|
|
|
|
public IManagedSubscription CreateSubscription(
|
|
ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options,
|
|
IMessageAckQueue queue,
|
|
SubscriptionLoadState? loadState = null,
|
|
Action<IManagedSubscription>? initialize = null)
|
|
{
|
|
CreateSubscriptionCallCount++;
|
|
LastCreateQueue = queue;
|
|
if (CreateSubscriptionFactory == null)
|
|
{
|
|
throw new InvalidOperationException(
|
|
"CreateSubscriptionFactory was not configured for this test.");
|
|
}
|
|
IManagedSubscription subscription = CreateSubscriptionFactory(handler, options, queue);
|
|
initialize?.Invoke(subscription);
|
|
return subscription;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public Client.ISubscriptionAcknowledgementProvider? AcknowledgementProvider => null;
|
|
|
|
/// <inheritdoc/>
|
|
public Client.ISubscriptionAcknowledgementObserver? AcknowledgementObserver => null;
|
|
|
|
/// <inheritdoc/>
|
|
public bool HasAcknowledgementPolicy => false;
|
|
|
|
/// <inheritdoc/>
|
|
public AcknowledgementSelection PrepareAcknowledgementsToSend(
|
|
List<SubscriptionAcknowledgement> acknowledgements)
|
|
{
|
|
return new AcknowledgementSelection(acknowledgements, []);
|
|
}
|
|
|
|
public ValueTask<PublishResponse> PublishAsync(
|
|
RequestHeader? requestHeader,
|
|
ArrayOf<SubscriptionAcknowledgement> subscriptionAcknowledgements,
|
|
CancellationToken ct = default)
|
|
{
|
|
// Never expected to run: no subscription is marked Created, so
|
|
// the controller spawns zero workers. Record the call so tests
|
|
// can prove the loop stayed quiescent.
|
|
Interlocked.Increment(ref m_publishCallCount);
|
|
return new ValueTask<PublishResponse>(new PublishResponse());
|
|
}
|
|
|
|
public ValueTask<TransferSubscriptionsResponse> TransferSubscriptionsAsync(
|
|
RequestHeader? requestHeader,
|
|
ArrayOf<uint> subscriptionIds,
|
|
bool sendInitialValues,
|
|
CancellationToken ct = default)
|
|
{
|
|
return new ValueTask<TransferSubscriptionsResponse>(
|
|
new TransferSubscriptionsResponse());
|
|
}
|
|
|
|
public ValueTask<DeleteSubscriptionsResponse> DeleteSubscriptionsAsync(
|
|
RequestHeader? requestHeader,
|
|
ArrayOf<uint> subscriptionIds,
|
|
CancellationToken ct = default)
|
|
{
|
|
return new ValueTask<DeleteSubscriptionsResponse>(
|
|
new DeleteSubscriptionsResponse());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reports that no subscription in this test is owned by the session
|
|
/// outside the manager's registry, so every identifier falls through
|
|
/// to the manager's own handling.
|
|
/// </summary>
|
|
public bool TryDispatchToSessionSubscription(
|
|
uint subscriptionId,
|
|
NotificationMessage message,
|
|
ArrayOf<uint> availableSequenceNumbers,
|
|
ArrayOf<string> stringTable,
|
|
bool moreNotifications)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
private int m_publishCallCount;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Minimal, inert notification handler. None of its callbacks are
|
|
/// invoked because no publish response is ever dispatched.
|
|
/// </summary>
|
|
private sealed class NoopNotificationHandler : ISubscriptionNotificationHandler
|
|
{
|
|
public ValueTask OnDataChangeNotificationAsync(ISubscription subscription,
|
|
uint sequenceNumber, DateTime publishTime,
|
|
ReadOnlyMemory<DataValueChange> notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask OnEventDataNotificationAsync(ISubscription subscription,
|
|
uint sequenceNumber, DateTime publishTime,
|
|
ReadOnlyMemory<EventNotification> notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask OnKeepAliveNotificationAsync(ISubscription subscription,
|
|
uint sequenceNumber, DateTime publishTime,
|
|
PublishState publishStateMask)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask OnSubscriptionStateChangedAsync(ISubscription subscription,
|
|
SubscriptionState state, PublishState publishStateMask,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Hand-rolled stub for the internal <see cref="IManagedSubscription"/>.
|
|
/// Exposes settable identity/counter state and records the small set of
|
|
/// lifecycle callbacks the tests assert on. All dispatch-driven members
|
|
/// throw because they are never reached in these quiescent tests.
|
|
/// </summary>
|
|
private sealed class StubManagedSubscription : IManagedSubscription
|
|
{
|
|
public uint Id { get; init; }
|
|
|
|
public long MessageGeneration => 0;
|
|
|
|
public Client.AcknowledgementScope GetAcknowledgementScope(long generation)
|
|
{
|
|
return new Client.AcknowledgementScope(this, generation, Id);
|
|
}
|
|
|
|
public bool Created { get; init; }
|
|
|
|
public bool IsCreationInProgress { get; init; }
|
|
|
|
public bool IsIntentionallyDeleted { get; init; }
|
|
|
|
public long MissingMessageCount { get; init; }
|
|
|
|
public long RepublishMessageCount { get; init; }
|
|
|
|
public int DisposeAsyncCalls { get; private set; }
|
|
|
|
public int RecreateAsyncCalls { get; private set; }
|
|
|
|
public List<bool> PausedCalls { get; } = [];
|
|
|
|
public TimeSpan CurrentPublishingInterval => TimeSpan.Zero;
|
|
|
|
public byte CurrentPriority => 0;
|
|
|
|
public uint CurrentLifetimeCount => 0;
|
|
|
|
public uint CurrentKeepAliveCount => 0;
|
|
|
|
public bool CurrentPublishingEnabled => false;
|
|
|
|
public uint CurrentMaxNotificationsPerPublish => 0;
|
|
|
|
public IMonitoredItemCollection MonitoredItems { get; } = Mock.Of<IMonitoredItemCollection>();
|
|
|
|
public ValueTask ConditionRefreshAsync(CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<TimeSpan> SetAsDurableAsync(
|
|
TimeSpan lifetime, CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<TimeSpan> SetAsDurableAsync(
|
|
Func<TimeSpan?> lifetime, CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<SetTriggeringResult> SetTriggeringAsync(
|
|
IMonitoredItem triggeringItem,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToAdd = null,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToRemove = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask OnPublishReceivedAsync(NotificationMessage message,
|
|
IReadOnlyList<uint>? availableSequenceNumbers,
|
|
IReadOnlyList<string> stringTable)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask<bool> TryCompleteTransferAsync(
|
|
IReadOnlyList<uint> availableSequenceNumbers,
|
|
CancellationToken ct = default)
|
|
{
|
|
throw new NotSupportedException();
|
|
}
|
|
|
|
public ValueTask RecreateAsync(CancellationToken ct = default)
|
|
{
|
|
RecreateAsyncCalls++;
|
|
return default;
|
|
}
|
|
|
|
public void NotifySubscriptionManagerPaused(bool paused)
|
|
{
|
|
PausedCalls.Add(paused);
|
|
}
|
|
|
|
public ValueTask DisposeAsync()
|
|
{
|
|
DisposeAsyncCalls++;
|
|
return default;
|
|
}
|
|
}
|
|
}
|
|
}
|