dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(opcuaclient): subscription tuning, OperationLimits chunking, diagnostics counters, revocation options, DiscoveryUrl (#273 #275 #276 #277 #278)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 14:40:20 -04:00
parent 55eac5f68b
commit ba62022e88
12 changed files with 1308 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+388 -96
View File
@@ -144,7 +144,24 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <param name="map">The namespace map to install, or <c>null</c> to clear it.</param>
internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map;
/// <summary>Test-only seam installing the OperationLimits that connect would have read (#275).</summary>
/// <param name="limits">The limits to apply to subsequent Read / Write / subscribe calls.</param>
internal void SetOperationLimitsForTest(OpcUaClientOperationLimits limits) => _limits = limits;
/// <summary>Test-only view of the diagnostics counters (#276).</summary>
internal OpcUaClientDiagnostics DiagnosticsForTest => _diag;
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// <summary>Counters surfaced through <see cref="GetHealth"/> (#276).</summary>
private readonly OpcUaClientDiagnostics _diag = new();
/// <summary>Server OperationLimits read at connect (#275); <c>None</c> when unread / disabled.
/// Replaced wholesale (never mutated), so readers just snapshot the reference.</summary>
private OpcUaClientOperationLimits _limits = OpcUaClientOperationLimits.None;
private NotificationEventHandler? _notificationHandler;
private PublishErrorEventHandler? _publishErrorHandler;
private bool _disposed;
/// <summary>URL of the endpoint the driver actually connected to. Exposed via <see cref="HostName"/>.</summary>
private string? _connectedEndpointUrl;
@@ -208,9 +225,31 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// §8 "Namespace Assignment" — a misconfigured remote fails draft validation here,
// not as a runtime surprise.
ValidateNamespaceKind(_options);
ValidateTuning(_options);
var appConfig = await BuildApplicationConfigurationAsync(cancellationToken).ConfigureAwait(false);
var candidates = ResolveEndpointCandidates(_options);
// #278: when a DiscoveryUrl is configured, FindServers/GetEndpoints on it first and try the
// discovered URLs ahead of the static list. Non-fatal — an LDS outage must not block init.
IReadOnlyList<string> discovered = [];
if (!string.IsNullOrWhiteSpace(_options.DiscoveryUrl))
{
try
{
discovered = await DiscoverEndpointsAsync(
appConfig, _options.DiscoveryUrl!, _options.SecurityPolicy, _options.SecurityMode,
cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; }
catch (Exception ex)
{
_logger.LogWarning(ex,
"OpcUaClientDriver '{DriverInstanceId}': discovery against '{DiscoveryUrl}' failed; " +
"falling back to the configured endpoint list.", _driverInstanceId, _options.DiscoveryUrl);
}
}
var candidates = ResolveEndpointCandidates(_options, discovered);
// G-2b: resolve any secret:-prefixed Password / UserCertificatePassword through the
// shared secret store ONCE, here in the async connect flow, just before the credentials
@@ -262,6 +301,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// via TransferSubscriptions so local MonitoredItem handles stay valid.
_keepAliveHandler = OnKeepAlive;
session.KeepAlive += _keepAliveHandler;
AttachDiagnostics(session);
_limits = _options.HonorServerOperationLimits
? await OpcUaClientOperationLimits.ReadAsync(session, cancellationToken).ConfigureAwait(false)
: OpcUaClientOperationLimits.None;
// Build the bidirectional namespace map from the freshly negotiated session's
// NamespaceUris (driver-specs.md §8 "Namespace Remapping"). Stored NodeIds carry
@@ -327,7 +370,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
TrustedIssuerCertificates = new CertificateTrustList
{
StoreType = CertificateStoreType.Directory,
StorePath = Path.Combine(pkiRoot, "issuers"),
// #277: the issuer store's crl/ sub-folder is where the SDK reads issuer CRLs, so an
// operator-supplied path is the CRL knob. Unset = the historical default.
StorePath = string.IsNullOrWhiteSpace(_options.CertificateValidation.IssuerStorePath)
? Path.Combine(pkiRoot, "issuers")
: _options.CertificateValidation.IssuerStorePath,
},
RejectedCertificateStore = new CertificateTrustList
{
@@ -335,6 +382,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
StorePath = Path.Combine(pkiRoot, "rejected"),
},
AutoAcceptUntrustedCertificates = _options.AutoAcceptCertificates,
RejectUnknownRevocationStatus = _options.CertificateValidation.RejectUnknownRevocationStatus,
},
TransportQuotas = new TransportQuotas { OperationTimeout = (int)_options.Timeout.TotalMilliseconds },
ClientConfiguration = new ClientConfiguration
@@ -360,8 +408,22 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// Accept the full set of certificate-validation error codes: a real dev cert can
// fail with BadCertificateChainIncomplete, BadCertificateTimeInvalid, or
// BadCertificateHostNameInvalid, not only BadCertificateUntrusted. Only accepting
// the latter would silently fail for those certs.
CertificateValidationEventHandler handler = (_, e) => e.Accept = true;
// the latter would silently fail for those certs. A REVOKED certificate is the one
// exception (#277): auto-accept is for dev self-signed certs, never for a cert its issuer
// has explicitly withdrawn.
CertificateValidationEventHandler handler = (_, e) =>
{
if (IsRevocationFailure(e.Error.StatusCode))
{
_logger.LogWarning(
"OpcUaClientDriver '{DriverInstanceId}': REVOKED server certificate rejected " +
"(subject={Subject}, status=0x{Status:X8}) — AutoAcceptCertificates does not apply to revoked certs.",
_driverInstanceId, e.Certificate?.Subject, e.Error.StatusCode.Code);
e.Accept = false;
return;
}
e.Accept = true;
};
config.CertificateValidator.CertificateValidation += handler;
// Store refs so ShutdownAsync + Dispose can detach the delegate and avoid
// leaking a closure on a potentially process-shared validator.
@@ -377,6 +439,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
return config;
}
/// <summary>True for the two hard "this certificate (or its issuer) is revoked" results (#277).</summary>
/// <param name="status">The SDK's certificate-validation status.</param>
/// <returns><c>true</c> when the status is Revoked / IssuerRevoked.</returns>
internal static bool IsRevocationFailure(StatusCode status) =>
status.Code == StatusCodes.BadCertificateRevoked || status.Code == StatusCodes.BadCertificateIssuerRevoked;
/// <summary>
/// Resolve the ordered failover candidate list. <c>EndpointUrls</c> wins when
/// non-empty; otherwise fall back to <c>EndpointUrl</c> as a single-URL shortcut so
@@ -384,10 +452,136 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// </summary>
/// <param name="opts">Driver options containing endpoint configuration.</param>
/// <returns>The ordered list of endpoint URLs to try.</returns>
internal static IReadOnlyList<string> ResolveEndpointCandidates(OpcUaClientDriverOptions opts)
internal static IReadOnlyList<string> ResolveEndpointCandidates(OpcUaClientDriverOptions opts) =>
ResolveEndpointCandidates(opts, []);
/// <summary>
/// Resolve the ordered failover list with optional discovery results (#278). Discovered URLs are
/// <b>prepended</b> so a discovery sweep gets first attempt; static entries follow, de-duplicated
/// case-insensitively so one server listed in both is not attempted twice in a row. With no
/// discovered URLs the result is exactly the pre-discovery behaviour.
/// </summary>
/// <param name="opts">Driver options containing endpoint configuration.</param>
/// <param name="discovered">URLs resolved via <see cref="DiscoverEndpointsAsync"/> (may be empty).</param>
/// <returns>The ordered list of endpoint URLs to try.</returns>
internal static IReadOnlyList<string> ResolveEndpointCandidates(
OpcUaClientDriverOptions opts, IReadOnlyList<string> discovered)
{
if (opts.EndpointUrls is { Count: > 0 }) return opts.EndpointUrls;
return [opts.EndpointUrl];
IReadOnlyList<string> staticList = opts.EndpointUrls is { Count: > 0 } ? opts.EndpointUrls : [opts.EndpointUrl];
if (discovered.Count == 0) return staticList;
var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
var merged = new List<string>(discovered.Count + staticList.Count);
foreach (var u in discovered)
if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u);
foreach (var u in staticList)
if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u);
return merged;
}
/// <summary>
/// <c>FindServers</c> on <paramref name="discoveryUrl"/> (an LDS, or the server itself), then
/// <c>GetEndpoints</c> on each returned server's discovery URL, keeping the endpoints whose security
/// policy + mode match. One unreachable server in the list is skipped, not fatal. The discovery
/// channel is unsecured by design (OPC UA Part 4 §5.4) regardless of the session's security mode.
/// </summary>
/// <param name="appConfig">Application configuration (client identity for the discovery channel).</param>
/// <param name="discoveryUrl">The LDS / server discovery URL.</param>
/// <param name="policy">Required security policy.</param>
/// <param name="mode">Required security mode.</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>Matching endpoint URLs in discovery order, de-duplicated.</returns>
internal static async Task<IReadOnlyList<string>> DiscoverEndpointsAsync(
ApplicationConfiguration appConfig, string discoveryUrl,
OpcUaSecurityPolicy policy, OpcUaSecurityMode mode, CancellationToken ct)
{
var wantedPolicyUri = MapSecurityPolicy(policy);
var wantedMode = ToMessageSecurityMode(mode);
ApplicationDescriptionCollection servers;
using (var lds = await DiscoveryClient.CreateAsync(
appConfig, new Uri(discoveryUrl), Opc.Ua.DiagnosticsMasks.None, ct).ConfigureAwait(false))
{
servers = await lds.FindServersAsync(null, ct).ConfigureAwait(false);
}
var urls = new List<string>();
foreach (var server in servers)
{
if (server.DiscoveryUrls is null) continue;
foreach (var serverDiscoveryUrl in server.DiscoveryUrls)
{
if (string.IsNullOrWhiteSpace(serverDiscoveryUrl)) continue;
EndpointDescriptionCollection endpoints;
try
{
using var ep = await DiscoveryClient.CreateAsync(
appConfig, new Uri(serverDiscoveryUrl), Opc.Ua.DiagnosticsMasks.None, ct).ConfigureAwait(false);
endpoints = await ep.GetEndpointsAsync(null, ct).ConfigureAwait(false);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
catch { continue; }
urls.AddRange(FilterEndpointUrls(endpoints, wantedPolicyUri, wantedMode));
}
}
return urls.Distinct(StringComparer.OrdinalIgnoreCase).ToList();
}
/// <summary>The endpoint URLs in <paramref name="endpoints"/> matching the wanted policy URI + mode.</summary>
/// <param name="endpoints">Endpoints advertised by a server.</param>
/// <param name="wantedPolicyUri">Required <c>SecurityPolicyUri</c>.</param>
/// <param name="wantedMode">Required message security mode.</param>
/// <returns>Matching <c>EndpointUrl</c>s, in advertised order.</returns>
internal static IEnumerable<string> FilterEndpointUrls(
IEnumerable<EndpointDescription> endpoints, string wantedPolicyUri, MessageSecurityMode wantedMode) =>
endpoints
.Where(e => e.SecurityPolicyUri == wantedPolicyUri && e.SecurityMode == wantedMode
&& !string.IsNullOrWhiteSpace(e.EndpointUrl))
.Select(e => e.EndpointUrl);
private static MessageSecurityMode ToMessageSecurityMode(OpcUaSecurityMode mode) => mode switch
{
OpcUaSecurityMode.None => MessageSecurityMode.None,
OpcUaSecurityMode.Sign => MessageSecurityMode.Sign,
OpcUaSecurityMode.SignAndEncrypt => MessageSecurityMode.SignAndEncrypt,
_ => throw new ArgumentOutOfRangeException(nameof(mode)),
};
/// <summary>
/// Range checks for the subscription-tuning and discovery options (#273/#278). Throws at
/// <see cref="InitializeAsync"/> so a bad config fails like the namespace-kind check does rather
/// than surfacing later as a server-side <c>BadInvalidArgument</c> on every subscribe.
/// </summary>
/// <param name="opts">The options to validate.</param>
internal static void ValidateTuning(OpcUaClientDriverOptions opts)
{
if (opts.SubscriptionKeepAliveCount < 1)
throw new InvalidOperationException("OpcUaClient SubscriptionKeepAliveCount must be >= 1.");
if (opts.SubscriptionLifetimeCount < 3L * opts.SubscriptionKeepAliveCount)
throw new InvalidOperationException(
"OpcUaClient SubscriptionLifetimeCount must be at least 3x SubscriptionKeepAliveCount " +
$"(got lifetime={opts.SubscriptionLifetimeCount}, keep-alive={opts.SubscriptionKeepAliveCount}).");
if (opts.SubscriptionPublishingInterval is { } pi && pi <= TimeSpan.Zero)
throw new InvalidOperationException("OpcUaClient SubscriptionPublishingInterval must be positive when set.");
if (!string.IsNullOrWhiteSpace(opts.DiscoveryUrl)
&& !Uri.TryCreate(opts.DiscoveryUrl, UriKind.Absolute, out _))
throw new InvalidOperationException(
$"OpcUaClient DiscoveryUrl '{opts.DiscoveryUrl}' is not an absolute URI.");
}
/// <summary>
/// Publishing interval (ms) a new subscription gets: the configured override if set, else the
/// interval the runtime requested, floored at 50 ms either way (servers negotiate up anyway, and
/// sub-50 ms just wastes negotiation).
/// </summary>
/// <param name="opts">Driver options.</param>
/// <param name="requested">The per-call interval from <c>SubscribeAsync</c>.</param>
/// <returns>The publishing interval in milliseconds.</returns>
internal static int ResolvePublishingIntervalMs(OpcUaClientDriverOptions opts, TimeSpan requested)
{
var effective = opts.SubscriptionPublishingInterval ?? requested;
return effective < TimeSpan.FromMilliseconds(50) ? 50 : (int)effective.TotalMilliseconds;
}
/// <summary>
@@ -506,13 +700,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var all = await client.GetEndpointsAsync(null, ct).ConfigureAwait(false);
var wantedPolicyUri = MapSecurityPolicy(policy);
var wantedMode = mode switch
{
OpcUaSecurityMode.None => MessageSecurityMode.None,
OpcUaSecurityMode.Sign => MessageSecurityMode.Sign,
OpcUaSecurityMode.SignAndEncrypt => MessageSecurityMode.SignAndEncrypt,
_ => throw new ArgumentOutOfRangeException(nameof(mode)),
};
var wantedMode = ToMessageSecurityMode(mode);
var match = all.FirstOrDefault(e =>
e.SecurityPolicyUri == wantedPolicyUri && e.SecurityMode == wantedMode);
@@ -639,8 +827,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { sessionToClose.KeepAlive -= _keepAliveHandler; } catch { }
}
_keepAliveHandler = null;
if (sessionToClose is not null) DetachDiagnostics(sessionToClose);
Session = null;
}
_limits = OpcUaClientOperationLimits.None;
try { if (sessionToClose is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
@@ -662,7 +852,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
/// <inheritdoc />
public DriverHealth GetHealth() => _health;
public DriverHealth GetHealth() => _health with { Diagnostics = _diag.Snapshot() };
// ---- IReadable ----
@@ -712,44 +902,59 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (toSend.Count == 0) return results;
try
// #275: one wire Read per OperationLimits.MaxNodesPerRead-sized chunk. A chunk that fails
// only fails ITS tags — the rest of the batch keeps its values.
var chunks = OpcUaClientOperationLimits.Chunk(toSend.Count, _limits.MaxNodesPerRead).ToList();
if (chunks.Count > 1) _diag.RecordChunkedRequest();
Exception? firstFailure = null;
foreach (var (start, count) in chunks)
{
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Both,
nodesToRead: toSend,
ct: cancellationToken).ConfigureAwait(false);
var slice = new ReadValueIdCollection(count);
for (var w = start; w < start + count; w++) slice.Add(toSend[w]);
try
{
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Both,
nodesToRead: slice,
ct: cancellationToken).ConfigureAwait(false);
var values = resp.Results;
for (var w = 0; w < values.Count; w++)
{
var r = indexMap[w];
var dv = values[w];
// Preserve the upstream StatusCode verbatim — including Bad codes per
// §8's cascading-quality rule. Also preserve SourceTimestamp so downstream
// clients can detect stale upstream data.
results[r] = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp);
var values = resp.Results;
for (var w = 0; w < values.Count && w < count; w++)
{
var r = indexMap[start + w];
var dv = values[w];
// Preserve the upstream StatusCode verbatim — including Bad codes per
// §8's cascading-quality rule. Also preserve SourceTimestamp so downstream
// clients can detect stale upstream data.
results[r] = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp);
}
}
_health = new DriverHealth(DriverState.Healthy, now, null);
}
catch (Exception ex)
{
// Transport / timeout / session-dropped — fan out the same fault across every
// tag in this batch. Per-tag StatusCode stays BadCommunicationError (not
// BadInternalError) so operators distinguish "upstream unreachable" from
// "driver bug".
for (var w = 0; w < indexMap.Count; w++)
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
var r = indexMap[w];
results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
throw;
}
catch (Exception ex)
{
// Transport / timeout / session-dropped — fan out the same fault across every
// tag in this chunk. Per-tag StatusCode stays BadCommunicationError (not
// BadInternalError) so operators distinguish "upstream unreachable" from
// "driver bug".
for (var w = start; w < start + count; w++)
results[indexMap[w]] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
firstFailure ??= ex;
}
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
_diag.RecordRead(failed: firstFailure is not null);
_health = firstFailure is null
? new DriverHealth(DriverState.Healthy, now, null)
: new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, firstFailure.Message);
}
finally { _gate.Release(); }
return results;
@@ -799,40 +1004,55 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (toSend.Count == 0) return results;
try
// #275: one wire Write per OperationLimits.MaxNodesPerWrite-sized chunk. Chunks are sent in
// order and each fails independently: a later chunk that times out leaves earlier chunks'
// real results intact (those writes DID happen).
var chunks = OpcUaClientOperationLimits.Chunk(toSend.Count, _limits.MaxNodesPerWrite).ToList();
if (chunks.Count > 1) _diag.RecordChunkedRequest();
var anyFailed = false;
foreach (var (start, count) in chunks)
{
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: toSend,
ct: cancellationToken).ConfigureAwait(false);
var codes = resp.Results;
for (var w = 0; w < codes.Count; w++)
var slice = new WriteValueCollection(count);
for (var w = start; w < start + count; w++) slice.Add(toSend[w]);
try
{
var r = indexMap[w];
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[r] = new WriteResult(codes[w].Code);
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: slice,
ct: cancellationToken).ConfigureAwait(false);
var codes = resp.Results;
for (var w = 0; w < codes.Count && w < count; w++)
{
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[indexMap[start + w]] = new WriteResult(codes[w].Code);
}
}
catch (OperationCanceledException)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue. Every not-yet-answered chunk (this one and
// the ones after it) gets the same code.
const uint StatusBadTimeout = 0x800A0000u;
anyFailed = true;
for (var w = start; w < toSend.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
break;
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
anyFailed = true;
for (var w = start; w < start + count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
}
}
catch (OperationCanceledException)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue.
const uint StatusBadTimeout = 0x800A0000u;
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
}
_diag.RecordWrite(anyFailed);
}
finally { _gate.Release(); }
return results;
@@ -1099,16 +1319,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.Historizing });
}
DataValueCollection values;
// #275: split the attribute reads to the server's MaxNodesPerRead (each variable contributes 4).
var values = new DataValueCollection(nodesToRead.Count);
try
{
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Neither,
nodesToRead: nodesToRead,
ct: ct).ConfigureAwait(false);
values = resp.Results;
foreach (var (start, count) in OpcUaClientOperationLimits.Chunk(nodesToRead.Count, _limits.MaxNodesPerRead))
{
var slice = new ReadValueIdCollection(count);
for (var w = start; w < start + count; w++) slice.Add(nodesToRead[w]);
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Neither,
nodesToRead: slice,
ct: ct).ConfigureAwait(false);
values.AddRange(resp.Results);
}
if (values.Count < nodesToRead.Count) throw new InvalidOperationException("Short enrichment read.");
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
}
catch
{
@@ -1230,20 +1461,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// Floor the publishing interval at 50ms — OPC UA servers routinely negotiate
// minimum-supported intervals up anyway, but sending sub-50ms wastes negotiation
// bandwidth on every subscription create.
var intervalMs = publishingInterval < TimeSpan.FromMilliseconds(50)
? 50
: (int)publishingInterval.TotalMilliseconds;
// bandwidth on every subscription create. #273: the options may pin the interval.
var intervalMs = ResolvePublishingIntervalMs(_options, publishingInterval);
var subscription = new Subscription(telemetry: null!, new SubscriptionOptions
{
DisplayName = $"opcua-sub-{id}",
PublishingInterval = intervalMs,
KeepAliveCount = 10,
LifetimeCount = 1000,
MaxNotificationsPerPublish = 0,
KeepAliveCount = _options.SubscriptionKeepAliveCount,
LifetimeCount = _options.SubscriptionLifetimeCount,
MaxNotificationsPerPublish = _options.SubscriptionMaxNotificationsPerPublish,
PublishingEnabled = true,
Priority = 0,
Priority = _options.SubscriptionPriority,
TimestampsToReturn = TimestampsToReturn.Both,
});
@@ -1263,6 +1492,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// captures `handle`, so we must hold the exact delegate instance returned by
// `+=` to be able to remove it.
var itemHandlers = new List<MonitoredItemNotificationHandle>();
var itemsAdded = 0;
foreach (var fullRef in fullReferences)
{
if (!TryResolveTagNode(session, fullRef, out var nodeId)) continue;
@@ -1287,6 +1517,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
item.Notification += notifHandler;
itemHandlers.Add(new MonitoredItemNotificationHandle(item, notifHandler));
subscription.AddItem(item);
itemsAdded++;
// #275: flush a CreateMonitoredItems call whenever the server's per-call cap is reached.
// CreateItemsAsync only sends the items not yet created, so calling it per batch is safe.
if (_limits.MaxMonitoredItemsPerCall > 0 && itemsAdded % _limits.MaxMonitoredItemsPerCall == 0)
{
_diag.RecordChunkedRequest();
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
}
}
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
@@ -1336,6 +1575,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var dv = mn.Value;
if (dv is null) return;
var fullRef = (item.Handle as string) ?? item.DisplayName ?? string.Empty;
_diag.RecordDataChange(bad: StatusCode.IsBad(dv.StatusCode));
var snapshot = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
@@ -1405,9 +1645,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
DisplayName = $"opcua-alarm-sub-{id}",
PublishingInterval = 500, // 500ms — alarms don't need fast polling; the server pushes
KeepAliveCount = 10,
LifetimeCount = 1000,
MaxNotificationsPerPublish = 0,
KeepAliveCount = _options.SubscriptionKeepAliveCount,
LifetimeCount = _options.SubscriptionLifetimeCount,
MaxNotificationsPerPublish = _options.SubscriptionMaxNotificationsPerPublish,
PublishingEnabled = true,
Priority = 0,
TimestampsToReturn = TimestampsToReturn.Both,
@@ -1988,6 +2228,49 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
/// <summary>Hook the SDK's publish-side events into the diagnostics counters (#276).</summary>
/// <param name="session">The session to observe.</param>
private void AttachDiagnostics(ISession session)
{
_notificationHandler ??= (_, e) =>
{
_diag.RecordPublishResponse();
var msg = e.NotificationMessage;
if (msg?.NotificationData is { } data)
foreach (var nd in data)
_diag.RecordNotifications(CountNotifications(nd));
};
_publishErrorHandler ??= (_, _) => _diag.RecordPublishError();
session.Notification -= _notificationHandler;
session.Notification += _notificationHandler;
session.PublishError -= _publishErrorHandler;
session.PublishError += _publishErrorHandler;
}
private void DetachDiagnostics(ISession session)
{
try
{
if (_notificationHandler is not null) session.Notification -= _notificationHandler;
if (_publishErrorHandler is not null) session.PublishError -= _publishErrorHandler;
}
catch { /* best-effort — SDK may have already cleared its invocation lists on session loss */ }
}
private static int CountNotifications(ExtensionObject nd) => nd.Body switch
{
DataChangeNotification dc => dc.MonitoredItems?.Count ?? 0,
EventNotificationList ev => ev.Events?.Count ?? 0,
StatusChangeNotification => 1,
_ => 1,
};
private async Task RefreshOperationLimitsAsync(ISession session)
{
try { _limits = await OpcUaClientOperationLimits.ReadAsync(session, CancellationToken.None).ConfigureAwait(false); }
catch { /* best-effort: keep the previous limits */ }
}
/// <summary>
/// Session keep-alive handler. On a healthy ping, bumps HostState back to Running
/// (typical bounce after a transient network blip). On a bad ping, starts the SDK's
@@ -2003,6 +2286,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
TransitionTo(HostState.Stopped);
_diag.RecordKeepAliveMiss();
// Kick off the SDK's reconnect loop exactly once per drop. Keep-alive callbacks fire
// from the SDK keep-alive timer thread and the SDK can fire this handler repeatedly
@@ -2022,6 +2306,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// BeginReconnect is started outside the lock — it does no _reconnectHandler mutation
// and we don't want to hold _probeLock across an SDK call.
_diag.RecordReconnectAttempt();
handler.BeginReconnect(
sender,
(int)_options.ReconnectPeriod.TotalMilliseconds,
@@ -2062,6 +2347,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
newSession.KeepAlive += _keepAliveHandler;
}
// Detach-then-attach: the SDK may hand back the SAME session object, and -= on a handler that
// is not attached is a no-op, so this never double-counts.
if (oldSession is not null) DetachDiagnostics(oldSession);
if (newSession is not null) AttachDiagnostics(newSession);
Session = newSession;
@@ -2084,6 +2373,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
RecordSkippedTags(_namespaceMap);
TransitionTo(HostState.Running);
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_diag.RecordReconnectSucceeded();
if (_options.HonorServerOperationLimits) _ = RefreshOperationLimitsAsync(newSession);
return;
}
@@ -2096,6 +2387,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// the last-known session for an always-on gateway rather than abandoning recovery
// entirely.
TransitionTo(HostState.Faulted);
_diag.RecordReconnectExhausted();
_health = new DriverHealth(
DriverState.Faulted, _health.LastSuccessfulRead,
"OPC UA session reconnect exhausted its retry window without recovering. " +