mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-21 17:16:16 +08:00
Marc Schier
416d619c41
# Description Two changes to how CI is routed and how it reports. ## 1. Split the CI matrix back across both systems Flips both halves of the CI backend switch from `ado` to `actions`, so the workload is shared instead of concentrated on the managed pool. The switch is checked into source in **two places that must always be flipped together**: | File | Setting | Before | After | | --- | --- | --- | --- | | `azure-pipelines.yml` | `parameters.ciBuildBackend` | `ado` | `actions` | | `.github/workflows/buildandtest.yml` | `env.CI_BUILD_BACKEND` | `ado` | `actions` | | Work | GitHub Actions | Azure Pipelines (managed pool) | | --- | --- | --- | | All-TFM solution builds | `build-all-tfm-windows` + `-linux` | two-TFM PR gate (`net10.0`, `net48`) | | Test matrix | ubuntu, every test project | Windows `net48` + `net10.0` fast PR legs | | Native AoT | ubuntu | Schedule/Manual only | | macOS | always (no managed-pool macOS image) | — | ## 2. Split the required gate from the advisory coverage check The coverage gate job was appended to *every* test stage, so `Fast PR test` emitted three sibling checks whose display names collided — `Coverage Release (net10.0)` appeared twice, once for the Windows leg and once for macOS, with nothing to tell them apart: ```text ✓ Coverage Release (net48) 55s ● Coverage Release (net10.0) ○ Coverage Release (net10.0) ``` That job also fused two unrelated concerns: whether the tests passed (which must block a merge) and whether coverage met its thresholds (which should not — the `net10.0` leg had `enforceCoverage: true`, so a coverage dip blocked merges). Both concerns now get a stage of their own, mirrored in both CI systems: | Concern | Azure Pipelines | GitHub Actions | In the branch ruleset? | | --- | --- | --- | --- | | Every test passed | **`Tests passed`** stage | **`build-and-test summary`** job | **Yes — required** | | Coverage meets thresholds | **`Code coverage`** stage | **`code coverage`** job | **No — advisory** | The coverage check reports a clean failure when the thresholds are missed, so the miss is visible, but it never blocks the merge. Coverage is now merged **once** across every leg of the run instead of once per leg, which also removes the `enforceCoverage` / `publishCoverage` "exactly one invocation may set this" parameters. Neither check re-runs any tests — the previous design re-ran the whole suite in one job with the collector attached, which serialised a suite that is deliberately fanned out and always exceeded the stage timeout. ### Coverage checks now exist in GitHub Actions `build-and-test` already ran with `--collect:"XPlat Code Coverage"` and uploaded the fragments, but nothing ever merged or evaluated them. It now runs the same `check-coverage.ps1` against the same `coverage-thresholds.json` as Azure. Since `CI_BUILD_BACKEND` defaults to `actions`, that ubuntu matrix — which runs *every* test project — is the fullest coverage signal in the repository. ### Reporting the numbers `check-coverage.ps1` gained `-SummaryPath`, rendering a markdown report that both systems surface: - **GitHub Actions** — appended to the job summary and upserted as a single **sticky pull-request comment**, updated in place on each run. Threshold misses also appear as `::error::` annotations. - **Azure Pipelines** — attached to the build summary via `##vso[task.uploadsummary]`. The report is a pass/fail table of the project floor, patch coverage and baseline delta, with the uncovered changed lines listed by file in a collapsible block. Rendered example: > ## Code coverage > > ❌ **Coverage gate failed.** This check is advisory and does not block the merge. > > | Check | Result | Threshold | > | --- | --- | --- | > | ✅ Project line rate | **72.10%** (40 731/56 496 lines) | >= 70.00% | > | ❌ Patch coverage | **0.00%** (0/1 changed lines) | >= 75.00% | The comment is fork-safe: a `pull_request` from a fork gets a read-only token regardless of the `permissions:` block, so it degrades to the job summary alone. The sticky comment uses `actions/github-script` rather than a community action, keeping this workflow on first-party actions only. ## Notable correctness details These were found and fixed during review, and are the parts most worth a second pair of eyes: - **`dependencies.<stage>.result` resolves at *stage* scope only.** At job scope `dependencies` refers to other jobs of the same stage, so reading stage results from a jobs template would have silently returned empty and the gate would have passed everything. `gate.yml` and `coverage.yml` are therefore **stage** templates that derive their own `dependsOn` from the same stage list they evaluate, so the two can never drift apart. An unresolved result is a hard error. - **Both rollups run on `not(canceled())` / `always()`**, not on success. A check that reports **skipped** is treated by GitHub as **satisfied**, so a required check that skips when its dependency fails would wave a red build straight through. - **`get-matrix.ps1` used to emit an empty matrix on a discovery miss.** The downstream job is then skipped, a skipped job rolls up as a *succeeded* stage, and the gate would have approved a run that executed no tests. It now fails, with an `-AllowEmpty` opt-out. - **Zero coverage reports used to be a green advisory check**, which would hide a broken collector indefinitely. Both systems now distinguish "the tests never ran" (warning) from "the tests ran and published nothing" (error). - The gate reads stage-level results rather than enumerating matrix-generated job names, which change whenever a test project or agent is added. ## Test-host hang and the reconnect defect behind it — both fixed Flipping `CI_BUILD_BACKEND` to `actions` started exercising the ubuntu matrix, which `master` never runs, and it immediately surfaced two real defects. **1. The test host hung for ten minutes.** `Assert.ThrowsAsync` blocks the calling thread with no timeout, and these tests drive a *fake clock* from that same thread — so once the reconnect parked on a fake-clock timer, nothing could ever advance it. NUnit sat in `WaitForCompletion` until `--blame-hang-timeout` killed the host, aborting the run at 3902 of 4238 tests and taking every other test in the assembly down with it. A hang dump named the culprit exactly. Replaced with an awaited, bounded `AssertThrowsAsync` helper (the repo's own rules forbid sync over async). The run now completes in **~1m29s** with all tests executed. **2. A budget-aware reconnect never returned.** With the hang gone, two tests failed with `Expected ServiceResultException but got TimeoutException`. `dumpasync` on a hang dump gave the whole chain: ``` ReconnectAsyncWithBudgetShrinksDelayToFitRemainingAsync → AssertThrowsAsync → ClientChannelManager.ReconnectAsync → ClientChannelManager.ReconnectLeaseAsync → ClientChannelManager.SwapFaultedEntryAsync ← parked here ``` `master` (via #4128) added `IsTerminalReconnectRace` so a reconnect that loses a race against a concurrent close recovers on a freshly swapped entry. The predicate only asks whether the entry ended up `Closed`/`Faulted` with `BadSecureChannelClosed` — which is *also* exactly how a reconnect ends when the caller's retry budget or the reconnect policy is deliberately exhausted. So a budget-exhausted reconnect was misread as a race: the manager swapped the entry and started a **second, unbudgeted** cycle, waiting out the swap back-off first. `ReconnectAsync(channel, budget)` therefore could not return within its budget, and under a fake clock it never returned at all. The fix requires the budget to have room left before treating a terminal failure as a race: ```csharp return !ct.IsCancellationRequested && !budget.IsExhausted && sre.StatusCode == StatusCodes.BadSecureChannelClosed && entry.State is ChannelState.Closed or ChannelState.Faulted; ``` The parameterless `ReconnectAsync(channel, ct)` overload passes an unlimited budget, so genuine race recovery is unchanged. Verified locally: Core 4152, Client 2122 and Sessions 776 tests all pass, and the full GitHub Actions PR run is green across 56 jobs. Two smaller fixes came along the way, neither weakening an assertion: - `ObservableFakeTimeProvider` waiters made **relative** ("N more timers from now") rather than absolute ("the Nth timer of the run"). Absolute numbering let an earlier reconnect consume the slots, so a waiter completed before the awaited timer existed and `Advance()` fired nothing. - The raw worker thread in `NodeStateHandlerConcurrencyTests` was a **foreground** thread whose stop flag was read without a barrier and set only on the success path — a failed assertion leaked a spinning thread that pins the process at exit. Now background, `Volatile`, stopped in a `finally`, joined with a bound. ## Background work now has an owner The reconnect hang above was one symptom of a pattern the codebase had **26 times** across seven assemblies: a bare `_ = Task.Run(...)`. Each one handed work to the thread pool and forgot it — nothing observed the exception if it threw, nothing bounded how many ran at once, and disposal raced them, so a component could finish tearing itself down while work it started was still touching its fields. Two of those sites already carried a standing `// TODO: Await the task completion in shutdown`. `BackgroundTaskScope` (in `Opc.Ua.Types`) gives them an owner: schedule with `Run`, drain with `DisposeAsync`, or signal-only with `Dispose` where the owner's teardown is synchronous and awaiting would be sync over async. `Run` never blocks and never throws — which matters, because most of these call sites went to the thread pool precisely to escape a lock they were holding. 11 unit tests cover drain, cancellation, exception observation, the concurrency cap, post-shutdown rejection and idempotent disposal. Two of the fixes were more than re-wiring: - **`SessionManager` / `SubscriptionManager` loops were structurally unawaitable.** Both return `ValueTask`, so `Task.Factory.StartNew(() => ...)` handed back a `Task<ValueTask>` that completes at the first yield. Storing it would not have helped — awaiting it would only have awaited the *scheduling*. Both now go through `AsTask().Unwrap()`, matching the neighbouring `StartConditionRefreshWorker`. - **`ChannelAsyncOperation` was swallowing subscriber exceptions.** Its `doNotBlock` branch had no try/catch while the inline branch beside it did. It is per-operation with no owner to hang a drain on, so it gained the missing handler rather than a scope. The Kubernetes readiness endpoint keeps a purpose-built semaphore instead: an unauthenticated endpoint needs a hard cap on concurrent handlers, not just an owner. ### ⚠️ Breaking: `ISessionManager.Shutdown()` is removed Replaced by `ShutdownAsync(CancellationToken)`. There is no correct synchronous way to wait for the session monitor loop, so leaving the overload would only have preserved the race. This mirrors `ISubscriptionManager`, which already had `ShutdownAsync`. Migration notes are in `docs/MigrationGuide.md`; only one in-repo implementer existed. Also converted the injected `object partitionLock` shared by `LogicalSubscription` and `CompositeMonitoredItemCollection` to `System.Threading.Lock`, per the repo rule. ## Patch coverage now scales with the size of the patch A flat floor over changed lines punishes small changes for arithmetic: two uncovered lines in a four-line fix reads as 50 % and failed an 80/5 floor — which is exactly how this branch earned a red coverage check for a four-line predicate change. A check that fires on changes nobody considers undertested is one people learn to ignore. | Coverable changed lines | Floor | Below it | | --- | --- | --- | | 1 – 10 | 50 % | warning, still passes | | 11 – 100 | 60 % | warning, still passes | | more than 100 | 75 % | **failure** | Only changes past the last band can fail. Below it the author still gets a warning naming the uncovered lines. Bands live in `patch.bands` in `coverage-thresholds.json` and each carries its own `enforced` flag. Verified end-to-end against synthetic repositories at every boundary. ## Codecov is back, as reporting Dropped in #4122 when the in-pipeline gate replaced it. Keeping the gate was right, but dropping Codecov also lost the PR comment, the file-by-file diff view and the trend. It returns with both statuses `informational: true` — two gates with two sets of thresholds would eventually disagree, and `coverage-thresholds.json` stays the single source of truth. The upload is switchable and can never fail a build: `enableCodecov` (Azure) and `ENABLE_CODECOV` (Actions), both defaulting on, both skipped when `CODECOV_TOKEN` is absent as on fork PRs. Each system uploads the single merged report it already builds, under its own flag. The old `codecov.io/bash` uploader was sunset, so Azure uses the current CLI. ## Follow-up actions for maintainers 1. **Update the branch ruleset.** The per-leg `Coverage Release (…)` checks are gone. Azure reports checks as `<pipeline> (<stage> <job>)`, so the exact names are: | Check | Require? | | --- | --- | | `OPCFoundation.UA-.NETStandard (Tests passed Verify stage results)` | **yes** | | `build-and-test summary` | **yes** (already required) | | `OPCFoundation.UA-.NETStandard (Code coverage Merge and evaluate)` | no | | `code coverage` | no | 2. **Re-calibrate the thresholds.** `coverage-thresholds.json` is still seeded from the last Codecov reading (73.61 %). The two systems now report *different* numbers by design — Actions merges every project on ubuntu, Azure only the Windows fast-PR legs, and scheduled runs read higher still because the Debug / .NET 8 / .NET 9 / netstandard stages also contribute. Re-seed from the GitHub Actions figure. 3. **Note:** a failed advisory coverage stage still turns the whole Azure run red, which the README build badge reflects on `master`. That follows from wanting a visible red X on a threshold miss; say the word if you would rather it were a warning. 4. **The Azure agent pool is the slowest thing here.** Several runs on this PR sat queued for hours, and one build failed with `Insufficient system resources` while writing an XML doc file — an agent exhaustion error, not a code error. It passed on re-run. Worth raising against the `netstandard` Managed DevOps Pool capacity separately. ## Related Issues _No tracking issue; this is CI infrastructure work requested directly. It builds on the switch introduced in #4122._ ## 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. - No product code changed, so there are no unit tests to add. The new logic was verified with purpose-built harnesses instead: - `check-coverage.ps1` run for real against a merged Cobertura report on all three paths — pass, threshold failure, and patch coverage with uncovered changed lines (a scratch commit adding an uncovered method was correctly resolved to `DataTypeException.cs:193`, reported 0 % patch coverage and exited 1). - The gate's accept/reject logic dry-run over `Succeeded` / `SucceededWithIssues` / `Skipped` / empty / `Failed` / `Canceled` — 6/6 correct. - `get-matrix.ps1` exit codes verified in clean processes: empty → 1, normal → 0, `-AllowEmpty` → 0. - All five pipeline/workflow files parse; both PowerShell scripts parse. - Action versions checked against the GitHub API and pinned to current majors. - [x] I have added all necessary documentation. - Rewrote the coverage section of `docs/DeveloperGuide.md` as *Required checks and coverage*, covering which checks belong in the ruleset and which must not, how coverage is measured, where the numbers appear, and why the two systems disagree. Corrected the stale "BLOCKS" wording in `coverage-thresholds.json`. - [x] I have verified that my changes do not introduce (new) build or analyzer warnings. - CI YAML, two PowerShell scripts and documentation only; no compilation involved. - [x] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - Not applicable — no product code changed. `Opc.Ua.Core.Schema.Tests` was run repeatedly on net10.0 (119 passed) to generate the real Cobertura fragments used to validate the gate. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [x] I have addressed **all** PR feedback received. - Feedback on the path-relevance diff (now a merge-base diff) and the target-framework numbering in `docs/DeveloperGuide.md` is fixed in `15b8e38df`. - A review pass over the gate/coverage split produced the four correctness fixes listed above, in `863eefc4d`. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 26aa55a2-5731-4a4c-b61c-df2f7a785eca
2265 lines
97 KiB
C#
2265 lines
97 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/
|
|
* ======================================================================*/
|
|
|
|
#nullable enable
|
|
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.Diagnostics.Metrics;
|
|
using System.Globalization;
|
|
using System.Linq;
|
|
using System.Reflection;
|
|
using System.Text;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Time.Testing;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Bindings;
|
|
using Opc.Ua.Security.Certificates;
|
|
using Opc.Ua.Tests;
|
|
|
|
// CA2000: test code; many disposables are ownership-transferred to test fixtures or short-lived.
|
|
#pragma warning disable CA2000
|
|
|
|
namespace Opc.Ua.Client.Tests.Stack.Client
|
|
{
|
|
/// <summary>
|
|
/// Unit tests for the managed (sharing, refcount, coalesced
|
|
/// reconnect, participant notification) behavior of
|
|
/// <see cref="ClientChannelManager"/> implementing
|
|
/// <see cref="IClientChannelManager"/>.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
[Parallelizable]
|
|
public sealed class ClientChannelManagerManagedTests
|
|
{
|
|
private static readonly ICertificateFactory s_factory = DefaultCertificateFactory.Instance;
|
|
|
|
/// <summary>
|
|
/// How long a test waits for an asynchronous operation it has already
|
|
/// unblocked to be observed as complete.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// This is a hang detector, not a latency assertion. These tests drive a
|
|
/// fake clock, so the work under test finishes in microseconds; what is
|
|
/// being waited on is purely the thread-pool scheduling of the
|
|
/// continuation chain. On a saturated CI agent the pool injects threads
|
|
/// at roughly one per second, so a short budget times out while the
|
|
/// operation is merely queued - the channel has already faulted and its
|
|
/// completion source has already been signalled. Keep this generous
|
|
/// enough that it never fires on a healthy run, and far below the
|
|
/// blame-hang timeout so a genuine deadlock still fails the job quickly.
|
|
/// </remarks>
|
|
private static readonly TimeSpan s_completionTimeout = TimeSpan.FromSeconds(60);
|
|
|
|
[Test]
|
|
public void ChannelKeyEqualityIsValueBased()
|
|
{
|
|
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
ConfiguredEndpoint endpoint1 = GetTestEndpoint(serverCert);
|
|
ConfiguredEndpoint endpoint2 = GetTestEndpoint(serverCert);
|
|
|
|
var k1 = ManagedChannelKey.FromEndpoint(endpoint1);
|
|
var k2 = ManagedChannelKey.FromEndpoint(endpoint2);
|
|
|
|
Assert.That(k1, Is.EqualTo(k2));
|
|
Assert.That(k1.GetHashCode(), Is.EqualTo(k2.GetHashCode()));
|
|
}
|
|
|
|
[Test]
|
|
public void ChannelKeyDistinguishesReverseFromForward()
|
|
{
|
|
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
|
|
var forward = ManagedChannelKey.FromEndpoint(endpoint);
|
|
var reverse = ManagedChannelKey.FromEndpoint(
|
|
endpoint,
|
|
reverseConnectionIdentity: new object());
|
|
|
|
Assert.That(forward, Is.Not.EqualTo(reverse));
|
|
}
|
|
|
|
[Test]
|
|
public void ChannelKeyDistinguishesDifferentReverseHandles()
|
|
{
|
|
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
|
|
var r1 = ManagedChannelKey.FromEndpoint(endpoint, reverseConnectionIdentity: new object());
|
|
var r2 = ManagedChannelKey.FromEndpoint(endpoint, reverseConnectionIdentity: new object());
|
|
|
|
Assert.That(r1, Is.Not.EqualTo(r2));
|
|
}
|
|
|
|
[Test]
|
|
public void ExponentialBackoffPolicyDoublesWithCap()
|
|
{
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.FromMilliseconds(100),
|
|
MaxDelay = TimeSpan.FromSeconds(1)
|
|
};
|
|
|
|
Assert.That(policy.GetDelay(0).TotalMilliseconds, Is.EqualTo(100));
|
|
Assert.That(policy.GetDelay(1).TotalMilliseconds, Is.EqualTo(200));
|
|
Assert.That(policy.GetDelay(2).TotalMilliseconds, Is.EqualTo(400));
|
|
Assert.That(policy.GetDelay(3).TotalMilliseconds, Is.EqualTo(800));
|
|
Assert.That(policy.GetDelay(4).TotalMilliseconds, Is.EqualTo(1000)); // capped
|
|
Assert.That(policy.GetDelay(10).TotalMilliseconds, Is.EqualTo(1000));
|
|
}
|
|
|
|
[Test]
|
|
public void ExponentialBackoffPolicyReturnsInfiniteWhenExhausted()
|
|
{
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MaxAttempts = 3
|
|
};
|
|
|
|
Assert.That(policy.GetDelay(2), Is.Not.EqualTo(Timeout.InfiniteTimeSpan));
|
|
Assert.That(policy.GetDelay(3), Is.EqualTo(Timeout.InfiniteTimeSpan));
|
|
Assert.That(policy.GetDelay(100), Is.EqualTo(Timeout.InfiniteTimeSpan));
|
|
}
|
|
|
|
[Test]
|
|
public async Task GetAsyncReturnsManagedChannelWithMatchingKeyAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
|
|
IManagedTransportChannel channel = await sut.GetAsync(participant, default)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(channel, Is.Not.Null);
|
|
Assert.That(channel.Key, Is.EqualTo(ManagedChannelKey.FromEndpoint(endpoint)));
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(channel.Manager, Is.SameAs(sut));
|
|
|
|
channel.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task GetAsyncWithFactoryConstructsParticipantInsideLockAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
IManagedTransportChannel? observedLease = null;
|
|
TestParticipant? participant = null;
|
|
int factoryCalls = 0;
|
|
|
|
IManagedTransportChannel channel = await sut.GetAsync(
|
|
endpoint,
|
|
lease =>
|
|
{
|
|
factoryCalls++;
|
|
observedLease = lease;
|
|
Assert.That(lease.Key, Is.EqualTo(ManagedChannelKey.FromEndpoint(endpoint)));
|
|
Assert.That(lease.Manager, Is.SameAs(sut));
|
|
var createdParticipant = new TestParticipant("p1", endpoint);
|
|
participant = createdParticipant;
|
|
return createdParticipant;
|
|
},
|
|
reverseConnection: null,
|
|
default)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(factoryCalls, Is.EqualTo(1));
|
|
Assert.That(channel, Is.SameAs(observedLease));
|
|
Assert.That(participant, Is.Not.Null);
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
|
|
|
|
channel.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task GetAsyncWithFactoryExceptionPropagates()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var expected = new InvalidOperationException("factory failed");
|
|
Exception? actual = null;
|
|
int factoryCalls = 0;
|
|
|
|
try
|
|
{
|
|
_ = await sut.GetAsync(
|
|
endpoint,
|
|
_ =>
|
|
{
|
|
factoryCalls++;
|
|
throw expected;
|
|
},
|
|
reverseConnection: null,
|
|
default)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
actual = ex;
|
|
}
|
|
|
|
Assert.That(actual, Is.SameAs(expected));
|
|
Assert.That(factoryCalls, Is.EqualTo(1));
|
|
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel channel = await sut.GetAsync(participant, default)
|
|
.ConfigureAwait(false);
|
|
|
|
chMock.Verify(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Exactly(2));
|
|
|
|
channel.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TwoParticipantsSameEndpointShareUnderlyingChannelAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var p1 = new TestParticipant("p1", endpoint);
|
|
var p2 = new TestParticipant("p2", endpoint);
|
|
|
|
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
|
|
|
|
Assert.That(ch1.Key, Is.EqualTo(ch2.Key));
|
|
Assert.That(ch1, Is.Not.SameAs(ch2)); // distinct lease wrappers
|
|
// underlying transport opened exactly once
|
|
chMock.Verify(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
ch1.Dispose();
|
|
ch2.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReleasingOneLeaseKeepsChannelAliveForOtherAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var p1 = new TestParticipant("p1", endpoint);
|
|
var p2 = new TestParticipant("p2", endpoint);
|
|
|
|
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
|
|
|
|
ch1.Dispose();
|
|
// underlying channel should still be alive
|
|
Assert.That(ch2.State, Is.EqualTo(ChannelState.Ready));
|
|
chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), Times.Never);
|
|
|
|
ch2.Dispose();
|
|
// give the close fiber a moment to run
|
|
await Task.Delay(100).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DiscoveryClientCreateAsyncSharesSessionChannelAndReleasesLeaseAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
var endpointConfiguration = new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
};
|
|
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(endpointConfiguration);
|
|
var sessionParticipant = new TestParticipant("session", endpoint);
|
|
IManagedTransportChannel sessionChannel = await sut.GetAsync(sessionParticipant, default)
|
|
.ConfigureAwait(false);
|
|
DiscoveryClient? discoveryClient = null;
|
|
|
|
try
|
|
{
|
|
discoveryClient = await DiscoveryClient.CreateAsync(
|
|
sut,
|
|
new Uri(endpoint.Description.EndpointUrl!),
|
|
endpointConfiguration,
|
|
NUnitTelemetryContext.Create(),
|
|
ct: default).ConfigureAwait(false);
|
|
|
|
Assert.That(discoveryClient.TransportChannel, Is.InstanceOf<IManagedTransportChannel>());
|
|
var discoveryChannel = (IManagedTransportChannel)discoveryClient.TransportChannel;
|
|
Assert.That(discoveryChannel.Key, Is.EqualTo(sessionChannel.Key));
|
|
Assert.That(discoveryChannel, Is.Not.SameAs(sessionChannel));
|
|
chMock.Verify(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
discoveryClient.Dispose();
|
|
discoveryClient = null;
|
|
|
|
chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), Times.Never);
|
|
Assert.That(sessionChannel.State, Is.EqualTo(ChannelState.Ready));
|
|
}
|
|
finally
|
|
{
|
|
discoveryClient?.Dispose();
|
|
sessionChannel.Dispose();
|
|
}
|
|
|
|
// lease.Dispose() is non-blocking; poll for the
|
|
// CloseAsync invocation before the strict verify.
|
|
await WaitForMockInvocationAsync(
|
|
() => chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), Times.Once))
|
|
.ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RegistrationClientCreateAsyncUsesManagedLeaseAndReleasesItAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
});
|
|
RegistrationClient registrationClient = await RegistrationClient.CreateAsync(
|
|
sut,
|
|
endpoint.Description,
|
|
endpoint.Configuration,
|
|
NUnitTelemetryContext.Create(),
|
|
ct: default).ConfigureAwait(false);
|
|
|
|
Assert.That(registrationClient.TransportChannel, Is.InstanceOf<IManagedTransportChannel>());
|
|
chMock.Verify(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
registrationClient.Dispose();
|
|
|
|
chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncNotifiesAttachedParticipantsAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
|
|
|
|
Assert.That(participant.NotificationCount, Is.GreaterThanOrEqualTo(1));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Ready));
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ConcurrentReconnectCallsCoalesceIntoOneCycleAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
int reconnectCalls = 0;
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(async () =>
|
|
{
|
|
Interlocked.Increment(ref reconnectCalls);
|
|
await Task.Delay(100).ConfigureAwait(false);
|
|
});
|
|
|
|
Task t1 = sut.ReconnectAsync(ch, default).AsTask();
|
|
Task t2 = sut.ReconnectAsync(ch, default).AsTask();
|
|
Task t3 = sut.ReconnectAsync(ch, default).AsTask();
|
|
|
|
await Task.WhenAll(t1, t2, t3).ConfigureAwait(false);
|
|
|
|
Assert.That(reconnectCalls, Is.EqualTo(1),
|
|
"Concurrent ReconnectAsync calls should coalesce into a single cycle.");
|
|
Assert.That(participant.NotificationCount, Is.EqualTo(1));
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FatalForChannelTransitionsToFaultedStateAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant(
|
|
"p1", endpoint,
|
|
onReconnect: (_, _, _) => ParticipantReconnectResult.FatalForChannel);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
|
|
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FatalForParticipantDetachesOnlyThatParticipantAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var p1 = new TestParticipant(
|
|
"p1", endpoint,
|
|
onReconnect: (_, _, _) => ParticipantReconnectResult.FatalForParticipant);
|
|
var p2 = new TestParticipant("p2", endpoint);
|
|
|
|
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch1, default).ConfigureAwait(false);
|
|
|
|
Assert.That(ch1.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(ch2.State, Is.EqualTo(ChannelState.Ready));
|
|
// p2 still got the notification
|
|
Assert.That(p2.NotificationCount, Is.GreaterThanOrEqualTo(1));
|
|
|
|
ch1.Dispose();
|
|
ch2.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAllAsyncTriggersAllEntriesAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var p1 = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAllAsync(default).ConfigureAwait(false);
|
|
|
|
Assert.That(p1.NotificationCount, Is.GreaterThanOrEqualTo(1));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Ready));
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RebindParticipantSwapsParticipantOnLeaseAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var bootstrap = new TestParticipant("boot", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(bootstrap, default)
|
|
.ConfigureAwait(false);
|
|
|
|
var real = new TestParticipant("real", endpoint);
|
|
#pragma warning disable CS0618 // Test verifies the obsolete compatibility shim remains functional.
|
|
sut.RebindParticipant(ch, real);
|
|
#pragma warning restore CS0618
|
|
// No exception means swap succeeded; participant list is internal,
|
|
// but downstream OnReconnect tests verify behavior.
|
|
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FailoverWithDifferentEndpointSwapsLeaseAsync()
|
|
{
|
|
var harness = new SessionChannelHarness();
|
|
ConfiguredEndpoint endpointA = GetSessionEndpoint("opc.tcp://localhost:4840");
|
|
ConfiguredEndpoint endpointB = GetSessionEndpoint("opc.tcp://localhost:4841");
|
|
Session? session = null;
|
|
try
|
|
{
|
|
session = await harness.CreateSessionAsync(endpointA).ConfigureAwait(false);
|
|
IManagedTransportChannel oldLease = session.ManagedChannel!;
|
|
ScriptedChannel oldTransport = harness.CreatedChannels[0];
|
|
|
|
await session.RecreateInPlaceAsync(endpointB, ct: default).ConfigureAwait(false);
|
|
|
|
IManagedTransportChannel newLease = session.ManagedChannel!;
|
|
Assert.That(newLease, Is.Not.SameAs(oldLease));
|
|
Assert.That(newLease.Key, Is.EqualTo(ManagedChannelKey.FromEndpoint(endpointB)));
|
|
Assert.That(newLease.State, Is.EqualTo(ChannelState.Ready));
|
|
// RecreateInPlaceAsync swaps the lease and fires the
|
|
// old-lease teardown asynchronously (the underlying
|
|
// CloseAsync runs on the thread pool). Poll for both
|
|
// the lease-state transition AND the underlying close
|
|
// before the hard assertions.
|
|
await WaitForConditionAsync(
|
|
() => oldLease.State == ChannelState.Closed &&
|
|
oldTransport.CloseCount == 1,
|
|
"oldLease.State == Closed && oldTransport.CloseCount == 1").ConfigureAwait(false);
|
|
Assert.That(oldLease.State, Is.EqualTo(ChannelState.Closed));
|
|
Assert.That(oldTransport.CloseCount, Is.EqualTo(1));
|
|
Assert.That(harness.CreatedChannels, Has.Count.EqualTo(2));
|
|
Assert.That(
|
|
harness.CreatedChannels[1].OpenedEndpointUrl,
|
|
Is.EqualTo(endpointB.Description.EndpointUrl));
|
|
}
|
|
finally
|
|
{
|
|
session?.Dispose();
|
|
await harness.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FailoverWithSameEndpointKeepsLeaseAsync()
|
|
{
|
|
var harness = new SessionChannelHarness();
|
|
ConfiguredEndpoint endpoint = GetSessionEndpoint("opc.tcp://localhost:4840");
|
|
Session? session = null;
|
|
try
|
|
{
|
|
session = await harness.CreateSessionAsync(endpoint).ConfigureAwait(false);
|
|
IManagedTransportChannel oldLease = session.ManagedChannel!;
|
|
ScriptedChannel transport = harness.CreatedChannels[0];
|
|
transport.SupportedFeatures = TransportChannelFeatures.Reconnect;
|
|
|
|
await session.RecreateInPlaceAsync(ct: default).ConfigureAwait(false);
|
|
|
|
Assert.That(session.ManagedChannel, Is.SameAs(oldLease));
|
|
Assert.That(transport.ReconnectCount, Is.EqualTo(1));
|
|
Assert.That(transport.CloseCount, Is.Zero);
|
|
Assert.That(harness.CreatedChannels, Has.Count.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
session?.Dispose();
|
|
await harness.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FailoverWithExplicitChannelTakesLegacyPathAsync()
|
|
{
|
|
var harness = new SessionChannelHarness();
|
|
ConfiguredEndpoint endpointA = GetSessionEndpoint("opc.tcp://localhost:4840");
|
|
ConfiguredEndpoint endpointB = GetSessionEndpoint("opc.tcp://localhost:4841");
|
|
Session? session = null;
|
|
try
|
|
{
|
|
session = await harness.CreateSessionAsync(endpointA).ConfigureAwait(false);
|
|
int managerCreatedChannels = harness.CreatedChannels.Count;
|
|
ScriptedChannel explicitChannel = harness.CreateOpenedStandaloneChannel(endpointB);
|
|
|
|
await session
|
|
.RecreateInPlaceAsync(endpointB, channel: explicitChannel.Channel, ct: default)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(harness.CreatedChannels, Has.Count.EqualTo(managerCreatedChannels));
|
|
Assert.That(session.TransportChannel, Is.SameAs(explicitChannel.Channel));
|
|
Assert.That(explicitChannel.SendRequestCount, Is.GreaterThan(0));
|
|
Assert.That(explicitChannel.CloseCount, Is.Zero);
|
|
}
|
|
finally
|
|
{
|
|
session?.Dispose();
|
|
await harness.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task MetricsAreEmittedForChannelLifetimeAsync()
|
|
{
|
|
using var metrics = new ChannelMetricListener();
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut(telemetry);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
string endpointUrl = endpoint.Description.EndpointUrl!;
|
|
var p1 = new TestParticipant("p1", endpoint);
|
|
var p2 = new TestParticipant("p2", endpoint);
|
|
|
|
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
|
|
metrics.RecordObservableInstruments();
|
|
|
|
ch1.Dispose();
|
|
ch2.Dispose();
|
|
|
|
// ch.Dispose() is non-blocking (lease teardown runs on
|
|
// the threadpool); poll for the close metric before the
|
|
// hard assertion so the test does not race with the
|
|
// asynchronous teardown.
|
|
await WaitForMeasurementAsync(
|
|
metrics,
|
|
"opc.ua.channel.close",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("reverse", false),
|
|
Tag("reason", "lease-released")).ConfigureAwait(false);
|
|
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.open",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("reverse", false)), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.active",
|
|
1,
|
|
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.active",
|
|
-1,
|
|
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.refcount",
|
|
2,
|
|
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.participants",
|
|
2,
|
|
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.close",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("reverse", false),
|
|
Tag("reason", "lease-released")), Is.True, metrics.FormatMeasurements());
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task MetricsAreEmittedForReconnectAndGateWaitAsync()
|
|
{
|
|
using var metrics = new ChannelMetricListener();
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut(telemetry);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
string endpointUrl = endpoint.Description.EndpointUrl!;
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
var reconnectEntered = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var allowReconnect = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(() => new ValueTask(WaitForReconnectReleaseAsync(
|
|
reconnectEntered,
|
|
allowReconnect)));
|
|
chMock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<IServiceResponse>(new ReadResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader { ServiceResult = StatusCodes.Good }
|
|
}));
|
|
|
|
Task reconnectTask = sut.ReconnectAsync(ch, default).AsTask();
|
|
await reconnectEntered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Task<IServiceResponse> sendTask = ch.SendRequestAsync(
|
|
new ReadRequest { RequestHeader = new RequestHeader() },
|
|
default).AsTask();
|
|
allowReconnect.SetResult(true);
|
|
|
|
await Task.WhenAll(reconnectTask, sendTask).ConfigureAwait(false);
|
|
|
|
// RunReconnectCycleAsync.RecordReconnectDuration is emitted from a
|
|
// finally block AFTER tcs.TrySetResult(true) returns, so on a fast
|
|
// runner the test thread can race past the assertion before the
|
|
// measurement lands. Poll briefly for the histogram measurement.
|
|
await WaitForMeasurementAsync(
|
|
metrics,
|
|
"opc.ua.channel.reconnect.duration",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("outcome", "success"))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.reconnect.attempts",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("outcome", "success")), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.reconnect.duration",
|
|
Tag("endpoint", endpointUrl),
|
|
Tag("outcome", "success")), Is.True, metrics.FormatMeasurements());
|
|
Assert.That(metrics.HasMeasurement(
|
|
"opc.ua.channel.gate.wait",
|
|
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SendRequestTransparentlyRetriesIdempotentRequestOnTransientDropAsync()
|
|
{
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
var sendCount = 0;
|
|
chMock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<IServiceRequest, CancellationToken>((_, _) =>
|
|
{
|
|
if (Interlocked.Increment(ref sendCount) == 1)
|
|
{
|
|
throw ServiceResultException.Create(
|
|
StatusCodes.BadConnectionClosed, "simulated in-flight drop");
|
|
}
|
|
|
|
return new ValueTask<IServiceResponse>(new ReadResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader { ServiceResult = StatusCodes.Good }
|
|
});
|
|
});
|
|
|
|
IServiceResponse response = await ch.SendRequestAsync(
|
|
new ReadRequest { RequestHeader = new RequestHeader() },
|
|
default).ConfigureAwait(false);
|
|
|
|
Assert.That(response, Is.Not.Null);
|
|
Assert.That(
|
|
sendCount,
|
|
Is.EqualTo(2),
|
|
"The idempotent request should be transparently resent after the channel recovered.");
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()), Times.AtLeastOnce);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SendRequestDoesNotRetryNonIdempotentRequestOnTransientDropAsync()
|
|
{
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
var sendCount = 0;
|
|
chMock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<IServiceRequest, CancellationToken>((_, _) =>
|
|
{
|
|
Interlocked.Increment(ref sendCount);
|
|
throw ServiceResultException.Create(
|
|
StatusCodes.BadConnectionClosed, "simulated in-flight drop");
|
|
});
|
|
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
ch.SendRequestAsync(
|
|
new WriteRequest { RequestHeader = new RequestHeader() },
|
|
default).AsTask(),
|
|
TimeSpan.FromSeconds(30)).ConfigureAwait(false);
|
|
|
|
Assert.That(ex, Is.Not.Null);
|
|
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadConnectionClosed));
|
|
Assert.That(
|
|
sendCount,
|
|
Is.EqualTo(1),
|
|
"A non-idempotent request must not be resent; the error surfaces for higher-level recovery.");
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()), Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SendRequestDoesNotRetryNonTransientErrorAsync()
|
|
{
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
var sendCount = 0;
|
|
chMock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<IServiceRequest, CancellationToken>((_, _) =>
|
|
{
|
|
Interlocked.Increment(ref sendCount);
|
|
throw ServiceResultException.Create(
|
|
StatusCodes.BadNodeIdUnknown, "non-transient application error");
|
|
});
|
|
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
ch.SendRequestAsync(
|
|
new ReadRequest { RequestHeader = new RequestHeader() },
|
|
default).AsTask(),
|
|
TimeSpan.FromSeconds(30)).ConfigureAwait(false);
|
|
|
|
Assert.That(ex, Is.Not.Null);
|
|
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadNodeIdUnknown));
|
|
Assert.That(
|
|
sendCount,
|
|
Is.EqualTo(1),
|
|
"A non-transient error must surface immediately without a reconnect.");
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()), Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectGenerationIncrementsOnTransportReconnectingAsync()
|
|
{
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
object entry = GetLeaseEntry(ch);
|
|
long before = (long)GetInternalPropertyValue(entry, "ReconnectGeneration");
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
|
|
|
|
long after = (long)GetInternalPropertyValue(entry, "ReconnectGeneration");
|
|
Assert.That(
|
|
after,
|
|
Is.GreaterThan(before),
|
|
"Entering TransportReconnecting must bump the reconnect generation.");
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
private static async Task WaitForMeasurementAsync(
|
|
ChannelMetricListener metrics,
|
|
string instrumentName,
|
|
params KeyValuePair<string, object?>[] tags)
|
|
{
|
|
const int kMaxPollMs = 2000;
|
|
const int kPollIntervalMs = 25;
|
|
int elapsed = 0;
|
|
while (!metrics.HasMeasurement(instrumentName, tags) && elapsed < kMaxPollMs)
|
|
{
|
|
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
|
|
elapsed += kPollIntervalMs;
|
|
}
|
|
}
|
|
|
|
private static async Task WaitForMockInvocationAsync(Action verify)
|
|
{
|
|
// Used to bridge Moq.Verify against state mutated by fire-
|
|
// and-forget tasks (e.g. ManagedTransportChannelLease.Dispose
|
|
// which posts the actual underlying CloseAsync onto the
|
|
// thread pool). Polls the verify until it stops throwing or
|
|
// the budget is exhausted; the final invocation is allowed
|
|
// to throw and surface the failure to NUnit.
|
|
const int kMaxPollMs = 2000;
|
|
const int kPollIntervalMs = 25;
|
|
int elapsed = 0;
|
|
while (elapsed < kMaxPollMs)
|
|
{
|
|
try
|
|
{
|
|
verify();
|
|
return;
|
|
}
|
|
catch (MockException)
|
|
{
|
|
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
|
|
elapsed += kPollIntervalMs;
|
|
}
|
|
}
|
|
verify();
|
|
}
|
|
|
|
private static async Task WaitForConditionAsync(
|
|
Func<bool> condition,
|
|
string description)
|
|
{
|
|
// Generic test-side poll for state mutated by fire-and-forget
|
|
// tasks. Returns once the condition holds; if the budget is
|
|
// exhausted the caller's subsequent assertion is allowed to
|
|
// run and surface the failure to NUnit.
|
|
const int kMaxPollMs = 2000;
|
|
const int kPollIntervalMs = 25;
|
|
int elapsed = 0;
|
|
while (!condition() && elapsed < kMaxPollMs)
|
|
{
|
|
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
|
|
elapsed += kPollIntervalMs;
|
|
}
|
|
_ = description;
|
|
}
|
|
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task ActivitySpanIsRecordedForReconnectAsync()
|
|
{
|
|
using var listener = new ChannelActivityListener();
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut(telemetry);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
string endpointUrl = endpoint.Description.EndpointUrl!;
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
|
|
|
|
Activity activity = await listener
|
|
.WaitForStoppedActivityAsync("OpcUaChannelReconnect")
|
|
.ConfigureAwait(false);
|
|
var tags = activity.TagObjects.ToDictionary(t => t.Key, t => t.Value);
|
|
Assert.That(tags, Does.ContainKey("endpoint"));
|
|
Assert.That(tags["endpoint"], Is.EqualTo(endpointUrl));
|
|
Assert.That(tags, Does.ContainKey("attempt.count"));
|
|
Assert.That(tags["attempt.count"], Is.EqualTo(1));
|
|
Assert.That(tags, Does.ContainKey("outcome"));
|
|
Assert.That(tags["outcome"], Is.EqualTo("success"));
|
|
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task StructuredLogsCaptureStateTransitionsAsync()
|
|
{
|
|
using var loggerProvider = new RecordingLoggerProvider();
|
|
ITelemetryContext telemetry = DefaultTelemetry.Create(
|
|
builder => builder.AddProvider(loggerProvider));
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) = CreateMockedSut(telemetry);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
|
|
ch.Dispose();
|
|
|
|
// ch.Dispose() is non-blocking; teardown logs are emitted by
|
|
// the background cleanup task.
|
|
await WaitForConditionAsync(
|
|
() => loggerProvider.Records.Any(record =>
|
|
record.EventId.Name == "ParticipantDetached") &&
|
|
loggerProvider.Records.Any(record =>
|
|
record.EventId.Name == "ChannelClosed"),
|
|
"ParticipantDetached + ChannelClosed events").ConfigureAwait(false);
|
|
|
|
RecordedLogRecord[] records = loggerProvider.Records
|
|
.Where(record => record.CategoryName == "Opc.Ua.ChannelManager")
|
|
.ToArray();
|
|
string formatted = string.Join(
|
|
Environment.NewLine,
|
|
records.Select(record => $"{record.EventId.Name} {record.Message}"));
|
|
string?[] eventNames = records.Select(record => record.EventId.Name).ToArray();
|
|
Assert.That(eventNames, Does.Contain("StateChanged"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ReconnectStarted"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ReconnectCompleted"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ParticipantAttached"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ParticipantDetached"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ChannelOpened"), formatted);
|
|
Assert.That(eventNames, Does.Contain("ChannelClosed"), formatted);
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task GetChannelDiagnosticsReturnsSnapshot()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint1 = GetTestEndpoint(serverCert);
|
|
ConfiguredEndpoint endpoint2 = GetTestEndpoint(serverCert);
|
|
endpoint2.Description.EndpointUrl = "opc.tcp://localhost:4841";
|
|
var p1 = new TestParticipant("p1", endpoint1);
|
|
var p2 = new TestParticipant("p2", endpoint2);
|
|
|
|
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
|
|
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
|
|
|
|
IReadOnlyList<ManagedChannelDiagnostic> snapshot = sut.GetChannelDiagnostics();
|
|
|
|
Assert.That(snapshot, Has.Count.EqualTo(2));
|
|
ManagedChannelDiagnostic d1 = snapshot.Single(d => d.Key == ch1.Key);
|
|
ManagedChannelDiagnostic d2 = snapshot.Single(d => d.Key == ch2.Key);
|
|
AssertChannelDiagnostic(d1, ch1.Key);
|
|
AssertChannelDiagnostic(d2, ch2.Key);
|
|
|
|
ch1.Dispose();
|
|
ch2.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncWithBudgetStopsWhenExhaustedAsync()
|
|
{
|
|
var timeProvider = new FakeTimeProvider();
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
var budget = new RetryBudget(TimeSpan.Zero, timeProvider);
|
|
|
|
// Await rather than Assert.ThrowsAsync: see AssertThrowsAsync.
|
|
Task exhaustedReconnect = sut.ReconnectAsync(ch, budget, default).AsTask();
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
exhaustedReconnect,
|
|
s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(ex, Is.Not.Null);
|
|
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncSwapsFaultedLeaseEntryAsync()
|
|
{
|
|
var timeProvider = new FakeTimeProvider();
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.FromMilliseconds(100),
|
|
MaxDelay = TimeSpan.FromMilliseconds(100),
|
|
MaxAttempts = 3
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, _) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
object originalEntry = GetLeaseEntry(ch);
|
|
var exhaustedBudget = new RetryBudget(TimeSpan.Zero, timeProvider);
|
|
|
|
// Await rather than Assert.ThrowsAsync: see AssertThrowsAsync.
|
|
// This test drives a fake clock from this thread, so blocking
|
|
// here would prevent the clock from ever moving again.
|
|
Task faultedReconnect = sut
|
|
.ReconnectAsync(ch, exhaustedBudget, default)
|
|
.AsTask();
|
|
await AssertThrowsAsync<ServiceResultException>(
|
|
faultedReconnect,
|
|
s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
|
|
Task reconnectTask = sut.ReconnectAsync(ch, default).AsTask();
|
|
await Task.Delay(10).ConfigureAwait(false);
|
|
|
|
Assert.That(reconnectTask.IsCompleted, Is.False, "Swap back-off should delay the reset.");
|
|
|
|
for (int i = 0; i < 4 && !reconnectTask.IsCompleted; i++)
|
|
{
|
|
timeProvider.Advance(TimeSpan.FromMilliseconds(100));
|
|
await Task.Delay(10).ConfigureAwait(false);
|
|
}
|
|
|
|
await reconnectTask.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
|
|
object freshEntry = GetLeaseEntry(ch);
|
|
ManagedChannelDiagnostic diagnostic = sut.GetChannelDiagnostics()
|
|
.Single(d => d.Key.Equals(ch.Key));
|
|
Assert.That(freshEntry, Is.Not.SameAs(originalEntry));
|
|
Assert.That(GetEntryState(freshEntry), Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(GetInternalIntProperty(freshEntry, "RefCount"), Is.EqualTo(1));
|
|
Assert.That(GetInternalIntProperty(freshEntry, "ParticipantCount"), Is.EqualTo(1));
|
|
Assert.That(GetInternalIntProperty(originalEntry, "RefCount"), Is.Zero);
|
|
Assert.That(GetInternalIntProperty(ch, "SwapCount"), Is.EqualTo(1));
|
|
Assert.That(diagnostic.Refcount, Is.EqualTo(1));
|
|
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(1));
|
|
Assert.That(participant.NotificationCount, Is.GreaterThanOrEqualTo(2));
|
|
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncWithBudgetShrinksDelayToFitRemainingAsync()
|
|
{
|
|
var timeProvider = new ObservableFakeTimeProvider();
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.FromSeconds(10),
|
|
MaxDelay = TimeSpan.FromSeconds(10)
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
var budget = new RetryBudget(TimeSpan.FromMilliseconds(100), timeProvider);
|
|
var reconnecting = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ch.StateChanged += (_, change) =>
|
|
{
|
|
if (change.NewState == ChannelState.TransportReconnecting)
|
|
{
|
|
reconnecting.TrySetResult(true);
|
|
}
|
|
};
|
|
|
|
// Arm before starting the reconnect so the waiter cannot be
|
|
// satisfied by an unrelated timer created earlier in the run.
|
|
Task<bool> shrunkBackoff = timeProvider.WaitForTimersCreatedAsync();
|
|
Task reconnectTask = sut.ReconnectAsync(ch, budget, default).AsTask();
|
|
await reconnecting.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
await shrunkBackoff.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(reconnectTask.IsCompleted, Is.False);
|
|
|
|
timeProvider.Advance(TimeSpan.FromMilliseconds(100));
|
|
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
reconnectTask,
|
|
s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(ex, Is.Not.Null);
|
|
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncWithExhaustedPolicyDoesNotSwapTheEntryAsync()
|
|
{
|
|
var timeProvider = new ObservableFakeTimeProvider();
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.FromMilliseconds(10),
|
|
MaxDelay = TimeSpan.FromMilliseconds(10),
|
|
MaxAttempts = 0
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
|
|
var participant = new TestParticipant("p1", endpoint);
|
|
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
|
|
|
|
// Deliberately generous. The policy, not the budget, is what ends this
|
|
// cycle: GetDelay returns the infinite sentinel as soon as the attempt
|
|
// count reaches MaxAttempts, before the budget is ever consulted. A
|
|
// race check that only asked whether the budget still had room would
|
|
// see plenty here, mistake the deliberate stop for a lost race against
|
|
// a concurrent close, and swap the entry to run a second, unbudgeted
|
|
// reconnect cycle behind the swap back-off.
|
|
var budget = new RetryBudget(TimeSpan.FromMinutes(1), timeProvider);
|
|
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
sut.ReconnectAsync(ch, budget, default).AsTask(),
|
|
s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(ex.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(GetInternalIntProperty(ch, "SwapCount"), Is.Zero);
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Awaits <paramref name="task"/> and returns the exception it faulted
|
|
/// with, failing the test if it succeeded or threw something else.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Deliberately not <c>Assert.ThrowsAsync</c>. That blocks the calling
|
|
/// thread until the task completes - sync over async - so on a
|
|
/// constrained CI agent it can starve the very continuation it is
|
|
/// waiting for, and it has no timeout, so the block is unbounded. Tests
|
|
/// that drive a fake clock from the test thread cannot survive either:
|
|
/// the clock cannot advance while the thread is blocked, NUnit's runner
|
|
/// thread never returns, and the whole test host hangs until the blame
|
|
/// collector kills it.
|
|
/// </remarks>
|
|
private static async Task<TException> AssertThrowsAsync<TException>(
|
|
Task task,
|
|
TimeSpan timeout)
|
|
where TException : Exception
|
|
{
|
|
try
|
|
{
|
|
await task.WaitAsync(timeout).ConfigureAwait(false);
|
|
}
|
|
catch (TException expected)
|
|
{
|
|
return expected;
|
|
}
|
|
catch (Exception other)
|
|
{
|
|
Assert.Fail(
|
|
$"Expected {typeof(TException).Name} but got " +
|
|
$"{other.GetType().Name}: {other}");
|
|
throw;
|
|
}
|
|
|
|
Assert.Fail(
|
|
$"Expected {typeof(TException).Name} but the operation completed successfully.");
|
|
throw new InvalidOperationException("unreachable");
|
|
}
|
|
|
|
private static (ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) CreateMockedSut(
|
|
ITelemetryContext? telemetry = null,
|
|
IChannelReconnectPolicy? reconnectPolicy = null,
|
|
TimeProvider? timeProvider = null)
|
|
{
|
|
telemetry ??= NUnitTelemetryContext.Create();
|
|
Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
|
|
var chMock = new Mock<IChannel>();
|
|
chMock.Setup(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<Uri, TransportChannelSettings, CancellationToken>(
|
|
(_, settings, _) => DisposeMockOwnedCertificates(settings))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.OpenAsync(
|
|
It.IsAny<ITransportWaitingConnection>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<ITransportWaitingConnection, TransportChannelSettings, CancellationToken>(
|
|
(_, settings, _) => DisposeMockOwnedCertificates(settings))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.None);
|
|
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings.Setup(b => b.Create(It.IsAny<string>(), It.IsAny<ITelemetryContext>()))
|
|
.Returns(chMock.Object);
|
|
|
|
var configuration = new ApplicationConfiguration(telemetry);
|
|
var sut = new ClientChannelManager(
|
|
configuration,
|
|
telemetry,
|
|
bindings.Object,
|
|
reconnectPolicy,
|
|
timeProvider);
|
|
return (sut, serverCert, chMock);
|
|
}
|
|
|
|
private static void DisposeMockOwnedCertificates(TransportChannelSettings settings)
|
|
{
|
|
settings.ServerCertificate?.Dispose();
|
|
settings.ServerCertificate = null;
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetTestEndpoint(Certificate serverCert)
|
|
{
|
|
var endpoint = new ConfiguredEndpoint
|
|
{
|
|
Configuration = new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
}
|
|
};
|
|
endpoint.Description.EndpointUrl = "opc.tcp://localhost:4840";
|
|
endpoint.Description.SecurityMode = MessageSecurityMode.SignAndEncrypt;
|
|
endpoint.Description.SecurityPolicyUri = SecurityPolicies.Basic256Sha256;
|
|
endpoint.Description.ServerCertificate = serverCert.RawData.ToByteString();
|
|
return endpoint;
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetNoneSecurityEndpoint(EndpointConfiguration endpointConfiguration)
|
|
{
|
|
var description = new EndpointDescription
|
|
{
|
|
EndpointUrl = "opc.tcp://localhost:4840",
|
|
SecurityMode = MessageSecurityMode.None,
|
|
SecurityPolicyUri = SecurityPolicies.None
|
|
};
|
|
description.Server.ApplicationUri = description.EndpointUrl;
|
|
description.Server.ApplicationType = ApplicationType.DiscoveryServer;
|
|
|
|
return new ConfiguredEndpoint(null, description, endpointConfiguration)
|
|
{
|
|
UpdateBeforeConnect = false
|
|
};
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetSessionEndpoint(string endpointUrl)
|
|
{
|
|
var endpointConfiguration = new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
};
|
|
var description = new EndpointDescription
|
|
{
|
|
EndpointUrl = endpointUrl,
|
|
SecurityMode = MessageSecurityMode.None,
|
|
SecurityPolicyUri = SecurityPolicies.None,
|
|
TransportProfileUri = Profiles.UaTcpTransport,
|
|
UserIdentityTokens =
|
|
[
|
|
new UserTokenPolicy
|
|
{
|
|
PolicyId = "anonymous",
|
|
TokenType = UserTokenType.Anonymous,
|
|
SecurityPolicyUri = SecurityPolicies.None
|
|
}
|
|
]
|
|
};
|
|
description.Server.ApplicationUri = endpointUrl;
|
|
description.Server.ApplicationType = ApplicationType.Server;
|
|
|
|
return new ConfiguredEndpoint(null, description, endpointConfiguration)
|
|
{
|
|
UpdateBeforeConnect = false
|
|
};
|
|
}
|
|
|
|
private static KeyValuePair<string, object?> Tag(string key, object? value)
|
|
{
|
|
return new KeyValuePair<string, object?>(key, value);
|
|
}
|
|
|
|
private static async Task WaitForReconnectReleaseAsync(
|
|
TaskCompletionSource<bool> reconnectEntered,
|
|
TaskCompletionSource<bool> allowReconnect)
|
|
{
|
|
reconnectEntered.TrySetResult(true);
|
|
await allowReconnect.Task.ConfigureAwait(false);
|
|
}
|
|
|
|
private static object GetLeaseEntry(IManagedTransportChannel channel)
|
|
{
|
|
return GetInternalPropertyValue(channel, "Entry");
|
|
}
|
|
|
|
private static ChannelState GetEntryState(object entry)
|
|
{
|
|
return (ChannelState)GetInternalPropertyValue(entry, "State");
|
|
}
|
|
|
|
private static int GetInternalIntProperty(object target, string propertyName)
|
|
{
|
|
return (int)GetInternalPropertyValue(target, propertyName);
|
|
}
|
|
|
|
private static object GetInternalPropertyValue(object target, string propertyName)
|
|
{
|
|
const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic;
|
|
PropertyInfo? property = target.GetType().GetProperty(propertyName, flags);
|
|
Assert.That(property, Is.Not.Null, $"Expected property {propertyName} on {target.GetType()}.");
|
|
object? value = property!.GetValue(target);
|
|
Assert.That(value, Is.Not.Null, $"Expected property {propertyName} to return a value.");
|
|
return value!;
|
|
}
|
|
|
|
private static void AssertChannelDiagnostic(
|
|
ManagedChannelDiagnostic diagnostic,
|
|
ManagedChannelKey key)
|
|
{
|
|
Assert.That(diagnostic.Key, Is.EqualTo(key));
|
|
Assert.That(diagnostic.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(diagnostic.Refcount, Is.EqualTo(1));
|
|
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(1));
|
|
Assert.That(diagnostic.OpenedAt, Is.Not.Default);
|
|
Assert.That(diagnostic.LastStateChange, Is.Not.Default);
|
|
Assert.That(diagnostic.LastReconnectAttempt, Is.Zero);
|
|
Assert.That(diagnostic.LastError, Is.Null);
|
|
}
|
|
|
|
public interface IChannel : ITransportChannel, ISecureChannel;
|
|
|
|
private sealed class SessionChannelHarness : IAsyncDisposable
|
|
{
|
|
public SessionChannelHarness()
|
|
{
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
Configuration = CreateConfiguration(telemetry);
|
|
m_bindings = new Mock<ITransportChannelBindings>();
|
|
m_bindings.Setup(b => b.Create(It.IsAny<string>(), It.IsAny<ITelemetryContext>()))
|
|
.Returns((string _, ITelemetryContext context) => CreateManagedChannel(context));
|
|
Manager = new ClientChannelManager(
|
|
Configuration,
|
|
telemetry,
|
|
m_bindings.Object,
|
|
new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
MaxAttempts = 1
|
|
});
|
|
}
|
|
|
|
public ApplicationConfiguration Configuration { get; }
|
|
|
|
public ClientChannelManager Manager { get; }
|
|
|
|
public List<ScriptedChannel> CreatedChannels { get; } = [];
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
await Manager.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
public Task<Session> CreateSessionAsync(ConfiguredEndpoint endpoint)
|
|
{
|
|
return Session.CreateAsync(
|
|
Manager,
|
|
Configuration,
|
|
endpoint,
|
|
updateBeforeConnect: false,
|
|
checkDomain: false,
|
|
sessionName: "ClientChannelManagerManagedTests",
|
|
sessionTimeout: 60000,
|
|
identity: new UserIdentity(),
|
|
ct: default);
|
|
}
|
|
|
|
public ScriptedChannel CreateOpenedStandaloneChannel(ConfiguredEndpoint endpoint)
|
|
{
|
|
var channel = new ScriptedChannel(Configuration.CreateMessageContext());
|
|
channel.OpenForEndpoint(endpoint);
|
|
return channel;
|
|
}
|
|
|
|
private static ApplicationConfiguration CreateConfiguration(ITelemetryContext telemetry)
|
|
{
|
|
return new ApplicationConfiguration(telemetry)
|
|
{
|
|
ApplicationName = "ClientChannelManagerManagedTests",
|
|
ApplicationType = ApplicationType.Client,
|
|
ApplicationUri = "urn:localhost:ClientChannelManagerManagedTests",
|
|
ProductUri = "urn:localhost:ClientChannelManagerManagedTests",
|
|
ClientConfiguration = new ClientConfiguration
|
|
{
|
|
DefaultSessionTimeout = 60000,
|
|
MinSubscriptionLifetime = 10000
|
|
},
|
|
TransportQuotas = new TransportQuotas
|
|
{
|
|
OperationTimeout = 6000,
|
|
MaxMessageSize = 1_048_576,
|
|
MaxStringLength = 1_048_576,
|
|
MaxByteStringLength = 1_048_576,
|
|
MaxArrayLength = 65_535
|
|
}
|
|
};
|
|
}
|
|
|
|
private ITransportChannel CreateManagedChannel(ITelemetryContext telemetry)
|
|
{
|
|
var channel = new ScriptedChannel(Configuration.CreateMessageContext(), telemetry);
|
|
CreatedChannels.Add(channel);
|
|
return channel.Channel;
|
|
}
|
|
|
|
private readonly Mock<ITransportChannelBindings> m_bindings;
|
|
}
|
|
|
|
private sealed class ScriptedChannel
|
|
{
|
|
public ScriptedChannel(IServiceMessageContext messageContext, ITelemetryContext? telemetry = null)
|
|
{
|
|
m_messageContext = messageContext;
|
|
Mock = new Mock<IChannel>();
|
|
Mock.Setup(c => c.SupportedFeatures).Returns(() => SupportedFeatures);
|
|
Mock.Setup(c => c.EndpointDescription).Returns(() => m_description);
|
|
Mock.Setup(c => c.EndpointConfiguration).Returns(() => m_endpointConfiguration);
|
|
Mock.Setup(c => c.MessageContext).Returns(m_messageContext);
|
|
Mock.Setup(c => c.ChannelThumbprint).Returns([]);
|
|
Mock.Setup(c => c.ClientChannelCertificate).Returns([]);
|
|
Mock.Setup(c => c.ServerChannelCertificate).Returns([]);
|
|
Mock.Setup(c => c.OperationTimeout).Returns(() => m_operationTimeout);
|
|
Mock.SetupSet(c => c.OperationTimeout = It.IsAny<int>())
|
|
.Callback<int>(value => m_operationTimeout = value);
|
|
Mock.Setup(c => c.CurrentToken).Returns((ChannelToken?)null);
|
|
Mock.Setup(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<Uri, TransportChannelSettings, CancellationToken>(OpenAsync);
|
|
Mock.Setup(c => c.OpenAsync(
|
|
It.IsAny<ITransportWaitingConnection>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<ITransportWaitingConnection, TransportChannelSettings, CancellationToken>(
|
|
OpenReverseAsync);
|
|
Mock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<ITransportWaitingConnection?, CancellationToken>(ReconnectAsync);
|
|
Mock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns<IServiceRequest, CancellationToken>(SendRequestAsync);
|
|
Mock.Setup(c => c.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns<CancellationToken>(CloseAsync);
|
|
Mock.Setup(c => c.Dispose()).Callback(() => DisposeCount++);
|
|
_ = telemetry;
|
|
}
|
|
|
|
public Mock<IChannel> Mock { get; }
|
|
|
|
public ITransportChannel Channel => Mock.Object;
|
|
|
|
public TransportChannelFeatures SupportedFeatures { get; set; }
|
|
|
|
public int CloseCount { get; private set; }
|
|
|
|
public int DisposeCount { get; private set; }
|
|
|
|
public int ReconnectCount { get; private set; }
|
|
|
|
public int SendRequestCount { get; private set; }
|
|
|
|
public string? OpenedEndpointUrl { get; private set; }
|
|
|
|
public void OpenForEndpoint(ConfiguredEndpoint endpoint)
|
|
{
|
|
m_description = endpoint.Description;
|
|
m_endpointConfiguration = endpoint.Configuration!;
|
|
OpenedEndpointUrl = endpoint.Description.EndpointUrl;
|
|
}
|
|
|
|
private ValueTask OpenAsync(
|
|
Uri uri,
|
|
TransportChannelSettings settings,
|
|
CancellationToken ct)
|
|
{
|
|
_ = uri;
|
|
_ = ct;
|
|
OpenWithSettings(settings);
|
|
return new ValueTask();
|
|
}
|
|
|
|
private ValueTask OpenReverseAsync(
|
|
ITransportWaitingConnection connection,
|
|
TransportChannelSettings settings,
|
|
CancellationToken ct)
|
|
{
|
|
_ = connection;
|
|
_ = ct;
|
|
OpenWithSettings(settings);
|
|
return new ValueTask();
|
|
}
|
|
|
|
private ValueTask ReconnectAsync(
|
|
ITransportWaitingConnection? connection,
|
|
CancellationToken ct)
|
|
{
|
|
_ = connection;
|
|
_ = ct;
|
|
ReconnectCount++;
|
|
return new ValueTask();
|
|
}
|
|
|
|
private ValueTask<IServiceResponse> SendRequestAsync(
|
|
IServiceRequest request,
|
|
CancellationToken ct)
|
|
{
|
|
_ = ct;
|
|
SendRequestCount++;
|
|
return new ValueTask<IServiceResponse>(CreateResponse(request));
|
|
}
|
|
|
|
private ValueTask CloseAsync(CancellationToken ct)
|
|
{
|
|
_ = ct;
|
|
CloseCount++;
|
|
return new ValueTask();
|
|
}
|
|
|
|
private void OpenWithSettings(TransportChannelSettings settings)
|
|
{
|
|
EndpointDescription description = settings.Description
|
|
?? throw new InvalidOperationException("Transport settings do not include an endpoint.");
|
|
EndpointConfiguration endpointConfiguration = settings.Configuration
|
|
?? throw new InvalidOperationException("Transport settings do not include endpoint configuration.");
|
|
|
|
m_description = description;
|
|
m_endpointConfiguration = endpointConfiguration;
|
|
OpenedEndpointUrl = description.EndpointUrl;
|
|
}
|
|
|
|
private IServiceResponse CreateResponse(IServiceRequest request)
|
|
{
|
|
return request switch
|
|
{
|
|
CreateSessionRequest => CreateSessionResponse(),
|
|
ActivateSessionRequest => CreateActivateSessionResponse(),
|
|
ReadRequest readRequest => CreateReadResponse(readRequest),
|
|
CloseSessionRequest => new CloseSessionResponse { ResponseHeader = CreateGoodHeader() },
|
|
_ => throw ServiceResultException.Create(
|
|
StatusCodes.BadServiceUnsupported,
|
|
"Unexpected request type {0}.",
|
|
request.GetType().Name)
|
|
};
|
|
}
|
|
|
|
private CreateSessionResponse CreateSessionResponse()
|
|
{
|
|
m_sessionCounter++;
|
|
string suffix = m_sessionCounter.ToString(CultureInfo.InvariantCulture);
|
|
return new CreateSessionResponse
|
|
{
|
|
ResponseHeader = CreateGoodHeader(),
|
|
SessionId = new NodeId($"session-{suffix}", 1),
|
|
AuthenticationToken = new NodeId($"token-{suffix}", 1),
|
|
RevisedSessionTimeout = 60000,
|
|
ServerNonce = ByteString.Empty,
|
|
ServerCertificate = ByteString.Empty,
|
|
ServerSignature = new SignatureData(),
|
|
ServerEndpoints = [m_description],
|
|
MaxRequestMessageSize = 1_048_576
|
|
};
|
|
}
|
|
|
|
private static ActivateSessionResponse CreateActivateSessionResponse()
|
|
{
|
|
return new ActivateSessionResponse
|
|
{
|
|
ResponseHeader = CreateGoodHeader(),
|
|
ServerNonce = ByteString.Empty,
|
|
Results = [],
|
|
DiagnosticInfos = []
|
|
};
|
|
}
|
|
|
|
private static ReadResponse CreateReadResponse(ReadRequest request)
|
|
{
|
|
return new ReadResponse
|
|
{
|
|
ResponseHeader = CreateGoodHeader(),
|
|
Results = CreateReadResults(request),
|
|
DiagnosticInfos = []
|
|
};
|
|
}
|
|
|
|
private static ResponseHeader CreateGoodHeader()
|
|
{
|
|
return new ResponseHeader
|
|
{
|
|
ServiceResult = StatusCodes.Good,
|
|
Timestamp = DateTime.UtcNow
|
|
};
|
|
}
|
|
|
|
private static ArrayOf<DataValue> CreateReadResults(ReadRequest request)
|
|
{
|
|
int count = request.NodesToRead.Count;
|
|
if (count == 1 &&
|
|
Equals(request.NodesToRead[0].NodeId, VariableIds.Server_ServerStatus_State))
|
|
{
|
|
return [CreateDataValue(new Variant((int)ServerState.Running))];
|
|
}
|
|
|
|
if (count == 2)
|
|
{
|
|
return
|
|
[
|
|
CreateDataValue(new Variant(ArrayOf.Wrapped(Namespaces.OpcUa))),
|
|
CreateDataValue(new Variant(ArrayOf.Wrapped("urn:localhost:server")))
|
|
];
|
|
}
|
|
|
|
var values = new DataValue[count];
|
|
for (int index = 0; index < count; index++)
|
|
{
|
|
if (count > 1 && index is 12 or 13 or 14)
|
|
{
|
|
values[index] = CreateDataValue(new Variant((ushort)0));
|
|
}
|
|
else if (count > 1 && index == 18)
|
|
{
|
|
values[index] = CreateDataValue(new Variant(0d));
|
|
}
|
|
else
|
|
{
|
|
values[index] = CreateDataValue(new Variant(0u));
|
|
}
|
|
}
|
|
return new ArrayOf<DataValue>(values);
|
|
}
|
|
|
|
private static DataValue CreateDataValue(Variant value)
|
|
{
|
|
return new DataValue(value, StatusCodes.Good);
|
|
}
|
|
|
|
private readonly IServiceMessageContext m_messageContext;
|
|
private EndpointDescription m_description = new();
|
|
private EndpointConfiguration m_endpointConfiguration = new();
|
|
private int m_operationTimeout;
|
|
private int m_sessionCounter;
|
|
}
|
|
|
|
private sealed class ChannelActivityListener : IDisposable
|
|
{
|
|
public ChannelActivityListener()
|
|
{
|
|
m_listener = new ActivityListener
|
|
{
|
|
ShouldListenTo = source => source.Name == "Opc.Ua.ChannelManager",
|
|
Sample = (ref _) =>
|
|
ActivitySamplingResult.AllDataAndRecorded,
|
|
ActivityStopped = activity =>
|
|
{
|
|
StoppedActivities.Add(activity);
|
|
_ = m_stoppedActivity.TrySetResult(activity);
|
|
}
|
|
};
|
|
ActivitySource.AddActivityListener(m_listener);
|
|
}
|
|
|
|
public List<Activity> StoppedActivities { get; } = [];
|
|
|
|
public async Task<Activity> WaitForStoppedActivityAsync(string operationName)
|
|
{
|
|
Activity activity = await m_stoppedActivity.Task
|
|
.WaitAsync(s_completionTimeout)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(activity.OperationName, Is.EqualTo(operationName));
|
|
return activity;
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
m_listener.Dispose();
|
|
}
|
|
|
|
private readonly ActivityListener m_listener;
|
|
|
|
private readonly TaskCompletionSource<Activity> m_stoppedActivity = new(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
}
|
|
|
|
private sealed class ChannelMetricListener : IDisposable
|
|
{
|
|
public ChannelMetricListener()
|
|
{
|
|
m_listener = new MeterListener
|
|
{
|
|
InstrumentPublished = (instrument, listener) =>
|
|
{
|
|
if (instrument.Name.StartsWith("opc.ua.channel.", StringComparison.Ordinal))
|
|
{
|
|
listener.EnableMeasurementEvents(instrument);
|
|
}
|
|
}
|
|
};
|
|
m_listener.SetMeasurementEventCallback<long>(OnLongMeasurementRecorded);
|
|
m_listener.SetMeasurementEventCallback<double>(OnDoubleMeasurementRecorded);
|
|
m_listener.Start();
|
|
}
|
|
|
|
public ConcurrentQueue<MeasurementRecord> Measurements { get; } = new();
|
|
|
|
public void RecordObservableInstruments()
|
|
{
|
|
m_listener.RecordObservableInstruments();
|
|
}
|
|
|
|
public bool HasMeasurement(
|
|
string instrumentName,
|
|
params KeyValuePair<string, object?>[] tags)
|
|
{
|
|
return HasMeasurement(instrumentName, null, tags);
|
|
}
|
|
|
|
public bool HasMeasurement(
|
|
string instrumentName,
|
|
double? value,
|
|
params KeyValuePair<string, object?>[] tags)
|
|
{
|
|
// Snapshot under enumeration to avoid races with the
|
|
// metric callbacks that fire concurrently from the
|
|
// channel manager's threadpool teardown work.
|
|
foreach (MeasurementRecord m in Measurements)
|
|
{
|
|
if (m.InstrumentName == instrumentName &&
|
|
(value == null || m.Value == value.Value) &&
|
|
tags.All(tag => m.Tags.TryGetValue(tag.Key, out object? actual) &&
|
|
Equals(actual, tag.Value)))
|
|
{
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
public string FormatMeasurements()
|
|
{
|
|
var builder = new StringBuilder();
|
|
// ConcurrentQueue enumeration is snapshot-stable.
|
|
foreach (MeasurementRecord measurement in Measurements)
|
|
{
|
|
string tags = string.Join(
|
|
", ",
|
|
measurement.Tags.Select(t => $"{t.Key}={t.Value}"));
|
|
builder
|
|
.Append(measurement.InstrumentName)
|
|
.Append('=')
|
|
.Append(measurement.Value.ToString(CultureInfo.InvariantCulture))
|
|
.Append(" {")
|
|
.Append(tags)
|
|
.AppendLine("}");
|
|
}
|
|
return builder.ToString();
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
m_listener.Dispose();
|
|
}
|
|
|
|
private void OnLongMeasurementRecorded(
|
|
Instrument instrument,
|
|
long measurement,
|
|
ReadOnlySpan<KeyValuePair<string, object?>> tags,
|
|
object? state)
|
|
{
|
|
Measurements.Enqueue(new MeasurementRecord(instrument.Name, measurement, tags.ToArray()));
|
|
}
|
|
|
|
private void OnDoubleMeasurementRecorded(
|
|
Instrument instrument,
|
|
double measurement,
|
|
ReadOnlySpan<KeyValuePair<string, object?>> tags,
|
|
object? state)
|
|
{
|
|
Measurements.Enqueue(new MeasurementRecord(instrument.Name, measurement, tags.ToArray()));
|
|
}
|
|
|
|
private readonly MeterListener m_listener;
|
|
}
|
|
|
|
private sealed class MeasurementRecord
|
|
{
|
|
public MeasurementRecord(
|
|
string instrumentName,
|
|
double value,
|
|
KeyValuePair<string, object?>[] tags)
|
|
{
|
|
InstrumentName = instrumentName;
|
|
Value = value;
|
|
Tags = tags.ToDictionary(t => t.Key, t => t.Value);
|
|
}
|
|
|
|
public string InstrumentName { get; }
|
|
|
|
public double Value { get; }
|
|
|
|
public Dictionary<string, object?> Tags { get; }
|
|
}
|
|
|
|
private sealed class TestParticipant : IReconnectParticipant
|
|
{
|
|
private readonly Func<IManagedTransportChannel, int, CancellationToken,
|
|
ParticipantReconnectResult>? m_onReconnect;
|
|
|
|
private int m_notificationCount;
|
|
|
|
public TestParticipant(
|
|
string id,
|
|
ConfiguredEndpoint endpoint,
|
|
Func<IManagedTransportChannel, int, CancellationToken,
|
|
ParticipantReconnectResult>? onReconnect = null)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
m_onReconnect = onReconnect;
|
|
}
|
|
|
|
public string Id { get; }
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
public int NotificationCount => Volatile.Read(ref m_notificationCount);
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref m_notificationCount);
|
|
ParticipantReconnectResult result = m_onReconnect?.Invoke(channel, reconnectAttempt, ct)
|
|
?? ParticipantReconnectResult.Reactivated;
|
|
return new ValueTask<ParticipantReconnectResult>(result);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A <see cref="FakeTimeProvider"/> that lets a test wait until the code
|
|
/// under test has actually registered its back-off timers.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Waiting is deliberately <b>relative</b>: a waiter is armed for "N
|
|
/// more timers from now" rather than for "the Nth timer of the run".
|
|
/// Absolute numbering is a race - anything else that happens to create a
|
|
/// timer on this provider first (an earlier reconnect in the same test,
|
|
/// or manager housekeeping) consumes the low numbers, the waiter then
|
|
/// completes before the timer the test cares about exists, and the
|
|
/// subsequent Advance fires nothing. The reconnect is left parked on a
|
|
/// fake clock that nobody will move again, which hangs the test - and,
|
|
/// because NUnit blocks the runner thread, the whole test host.
|
|
/// </remarks>
|
|
private sealed class ObservableFakeTimeProvider : FakeTimeProvider
|
|
{
|
|
public override ITimer CreateTimer(
|
|
TimerCallback callback,
|
|
object? state,
|
|
TimeSpan dueTime,
|
|
TimeSpan period)
|
|
{
|
|
ITimer timer = base.CreateTimer(callback, state, dueTime, period);
|
|
|
|
List<TaskCompletionSource<bool>>? ready = null;
|
|
lock (m_lock)
|
|
{
|
|
m_timerCount++;
|
|
for (int i = m_waiters.Count - 1; i >= 0; i--)
|
|
{
|
|
if (m_timerCount >= m_waiters[i].Target)
|
|
{
|
|
(ready ??= []).Add(m_waiters[i].Completion);
|
|
m_waiters.RemoveAt(i);
|
|
}
|
|
}
|
|
}
|
|
|
|
if (ready != null)
|
|
{
|
|
foreach (TaskCompletionSource<bool> completion in ready)
|
|
{
|
|
completion.TrySetResult(true);
|
|
}
|
|
}
|
|
|
|
return timer;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns a task that completes once <paramref name="count"/>
|
|
/// further timers have been created, counted from this call. Arm it
|
|
/// <b>before</b> starting the operation whose timers are awaited.
|
|
/// </summary>
|
|
public Task<bool> WaitForTimersCreatedAsync(int count = 1)
|
|
{
|
|
if (count < 1)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(count));
|
|
}
|
|
|
|
var completion = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
lock (m_lock)
|
|
{
|
|
m_waiters.Add((m_timerCount + count, completion));
|
|
}
|
|
return completion.Task;
|
|
}
|
|
|
|
private readonly System.Threading.Lock m_lock = new();
|
|
private readonly List<(int Target, TaskCompletionSource<bool> Completion)> m_waiters = [];
|
|
private int m_timerCount;
|
|
}
|
|
}
|
|
}
|