Compare commits

...

4 Commits

Author SHA1 Message Date
Joseph Doherty
0433d3a35e Phase 3 PR 69 -- OPC UA Client ISubscribable + IHostConnectivityProbe. Completes the OpcUaClientDriver capability surface — now matches the Galaxy + Modbus + S7 driver coverage. ISubscribable: SubscribeAsync creates a new upstream Subscription via the non-obsolete Subscription(ITelemetryContext, SubscriptionOptions) ctor + AddItem/CreateItemsAsync flow, wires each MonitoredItem's Notification event into OnDataChange. Tag strings round-trip through MonitoredItem.Handle so the notification handler can identify which tag changed without a second lookup. Publishing interval floored at 50ms (servers negotiate up anyway; sub-50ms wastes round-trip). SubscriptionOptions uses KeepAliveCount=10, LifetimeCount=1000, TimestampsToReturn=Both so SourceTimestamp passthrough for the cascading-quality rule works through subscription paths too. UnsubscribeAsync calls Subscription.DeleteAsync(silent:true) and tolerates unknown handles (returns cleanly) because the caller's race with server-side cleanup after a session drop shouldn't crash either side. Session shutdown explicitly deletes every remote subscription before closing — avoids BadSubscriptionIdInvalid noise in the upstream server's log on Close. IHostConnectivityProbe: HostName surfaced as the EndpointUrl (not host:port like the Modbus/S7 drivers) so the Admin /hosts dashboard can render the full opc.tcp:// URL as a clickable target back at the remote server. HostState tracked via session.KeepAlive event — OPC UA's built-in keep-alive is authoritative for session liveness (the SDK pings on KeepAliveInterval, sets KeepAliveStopped after N missed pings), strictly better than a driver-side polling probe: no extra wire round-trip, no duplicate semantic with the native protocol. Handler transitions Running on healthy keep-alives and Stopped on any Bad service-result. Initial Running raised at end of InitializeAsync once the session is up; Shutdown transitions back to Unknown + unwires the handler. Unit tests (OpcUaClientSubscribeAndProbeTests, 3 facts): SubscribeAsync_without_initialize_throws_InvalidOperationException, UnsubscribeAsync_with_unknown_handle_is_noop (session-drop-race safety), GetHostStatuses_returns_endpoint_url_row_pre_init (asserts EndpointUrl as the host identity -- the full opc.tcp://plc.example:4840 URL). Live-session subscribe/unsubscribe round-trip + keep-alive state transition coverage lands in a follow-up PR once we scaffold the in-process OPC UA server fixture. 13/13 OpcUaClient.Tests pass. dotnet build clean. All six capability interfaces (IDriver / ITagDiscovery / IReadable / IWritable / ISubscribable / IHostConnectivityProbe) implemented — OPC UA Client driver surface complete. 2026-04-19 01:22:14 -04:00
141673fc80 Merge pull request 'Phase 3 PR 68 -- OPC UA Client ITagDiscovery (Full browse)' (#67) from phase-3-pr68-opcua-client-discovery into v2 2026-04-19 01:19:27 -04:00
Joseph Doherty
db56a95819 Phase 3 PR 68 -- OPC UA Client ITagDiscovery via recursive browse (Full strategy). Adds ITagDiscovery to OpcUaClientDriver. DiscoverAsync opens a single Remote folder on the IAddressSpaceBuilder and recursively browses from the configured root (default: ObjectsFolder i=85; override via OpcUaClientDriverOptions.BrowseRoot for scoped discovery). Browse uses non-obsolete Session.BrowseAsync(RequestHeader, ViewDescription, uint maxReferences, BrowseDescriptionCollection, ct) with HierarchicalReferences forward, subtypes included, NodeClassMask Object+Variable, ResultMask pulling BrowseName + DisplayName + NodeClass + TypeDefinition. Objects become sub-folders via builder.Folder; Variables become builder.Variable entries with FullName set to the NodeId.ToString() serialization so IReadable/IWritable can round-trip without re-resolving. Three safety caps added to OpcUaClientDriverOptions to bound runaway discovery: (1) MaxBrowseDepth default 10 -- deep enough for realistic OPC UA information models, shallow enough that cyclic graphs can't spin the browse forever. (2) MaxDiscoveredNodes default 10_000 -- caps memory on pathological remote servers. Once the cap is hit, recursion short-circuits and the partially-discovered tree is still projected into the local address space (graceful degradation rather than all-or-nothing). (3) BrowseRoot as an opt-in scope restriction string per driver-specs.md \u00A78 -- defaults to ObjectsFolder but operators with 100k-node servers can point it at a single subtree. Visited-set tracks NodeIds already visited to prevent infinite cycles on graphs with non-strict hierarchy (OPC UA models can have back-references). Transient browse failures on a subtree are swallowed -- the sub-branch stops but the rest of discovery continues, matching the Modbus driver's 'transient poll errors don't kill the loop' pattern. The driver's health surface reflects the network-level cascade via the probe loop (PR 69). Deferred to a follow-up PR: DataType resolution via a batch Session.ReadAsync(Attributes.DataType) after the browse so DriverAttributeInfo.DriverDataType is accurate instead of the current conservative DriverDataType.Int32 default; AccessLevel-derived SecurityClass instead of the current ViewOnly default; array-type detection via Attributes.ValueRank + ArrayDimensions. These need an extra wire round-trip per batch of variables + a NodeId -> DriverDataType mapping table; out of scope for PR 68 to keep browse path landable. Unit tests (OpcUaClientDiscoveryTests, 3 facts): DiscoverAsync_without_initialize_throws_InvalidOperationException (pre-init hits RequireSession); DiscoverAsync_rejects_null_builder (ArgumentNullException); Discovery_caps_are_sensible_defaults (asserts 10000 / 10 / null defaults documented above). NullAddressSpaceBuilder stub implements the full IAddressSpaceBuilder shape including IVariableHandle.MarkAsAlarmCondition (throws NotSupportedException since this PR doesn't wire alarms). Live-browse coverage against a real remote server is deferred to the in-process-server-fixture PR. 10/10 OpcUaClient.Tests pass. dotnet build clean. 2026-04-19 01:17:21 -04:00
89bd726fa8 Merge pull request 'Phase 3 PR 67 -- OPC UA Client IReadable + IWritable' (#66) from phase-3-pr67-opcua-client-read-write into v2 2026-04-19 01:15:42 -04:00
4 changed files with 403 additions and 1 deletions

View File

@@ -27,8 +27,20 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// </para>
/// </remarks>
public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string driverInstanceId)
: IDriver, IReadable, IWritable, IDisposable, IAsyncDisposable
: IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IDisposable, IAsyncDisposable
{
// ---- ISubscribable + IHostConnectivityProbe state ----
private readonly System.Collections.Concurrent.ConcurrentDictionary<long, RemoteSubscription> _subscriptions = new();
private long _nextSubscriptionId;
private readonly object _probeLock = new();
private HostState _hostState = HostState.Unknown;
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
private KeepAliveEventHandler? _keepAliveHandler;
public event EventHandler<DataChangeEventArgs>? OnDataChange;
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
// OPC UA StatusCode constants the driver surfaces for local-side faults. Upstream-server
// StatusCodes are passed through verbatim per driver-specs.md §8 "cascading quality" —
// downstream clients need to distinguish 'remote source down' from 'local driver failure'.
@@ -110,8 +122,21 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
session.KeepAliveInterval = (int)_options.KeepAliveInterval.TotalMilliseconds;
// Wire the session's keep-alive channel into HostState. OPC UA keep-alives are
// authoritative for session liveness: the SDK pings on KeepAliveInterval and sets
// KeepAliveStopped when N intervals elapse without a response. That's strictly
// better than a driver-side polling probe — no extra round-trip, no duplicate
// semantic.
_keepAliveHandler = (_, e) =>
{
var healthy = !ServiceResult.IsBad(e.Status);
TransitionTo(healthy ? HostState.Running : HostState.Stopped);
};
session.KeepAlive += _keepAliveHandler;
Session = session;
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
TransitionTo(HostState.Running);
}
catch (Exception ex)
{
@@ -214,10 +239,29 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
// Tear down remote subscriptions first — otherwise Session.Close will try and may fail
// with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared
// whether or not the wire-side delete succeeds since the local handles are useless
// after close anyway.
foreach (var rs in _subscriptions.Values)
{
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
}
_subscriptions.Clear();
if (_keepAliveHandler is not null && Session is not null)
{
try { Session.KeepAlive -= _keepAliveHandler; } catch { }
}
_keepAliveHandler = null;
try { if (Session is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
try { Session?.Dispose(); } catch { }
Session = null;
TransitionTo(HostState.Unknown);
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
@@ -380,6 +424,235 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
private ISession RequireSession() =>
Session ?? throw new InvalidOperationException("OpcUaClientDriver not initialized");
// ---- ITagDiscovery ----
public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
var session = RequireSession();
var root = !string.IsNullOrEmpty(_options.BrowseRoot)
? NodeId.Parse(session.MessageContext, _options.BrowseRoot)
: ObjectIds.ObjectsFolder;
var rootFolder = builder.Folder("Remote", "Remote");
var visited = new HashSet<NodeId>();
var discovered = 0;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await BrowseRecursiveAsync(session, root, rootFolder, visited,
depth: 0, discovered: () => discovered, increment: () => discovered++,
ct: cancellationToken).ConfigureAwait(false);
}
finally { _gate.Release(); }
}
private async Task BrowseRecursiveAsync(
ISession session, NodeId node, IAddressSpaceBuilder folder, HashSet<NodeId> visited,
int depth, Func<int> discovered, Action increment, CancellationToken ct)
{
if (depth >= _options.MaxBrowseDepth) return;
if (discovered() >= _options.MaxDiscoveredNodes) return;
if (!visited.Add(node)) return;
var browseDescriptions = new BrowseDescriptionCollection
{
new()
{
NodeId = node,
BrowseDirection = BrowseDirection.Forward,
ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences,
IncludeSubtypes = true,
NodeClassMask = (uint)(NodeClass.Object | NodeClass.Variable),
ResultMask = (uint)(BrowseResultMask.BrowseName | BrowseResultMask.DisplayName
| BrowseResultMask.NodeClass | BrowseResultMask.TypeDefinition),
}
};
BrowseResponse resp;
try
{
resp = await session.BrowseAsync(
requestHeader: null,
view: null,
requestedMaxReferencesPerNode: 0,
nodesToBrowse: browseDescriptions,
ct: ct).ConfigureAwait(false);
}
catch
{
// Transient browse failure on a sub-tree — don't kill the whole discovery, just
// skip this branch. The driver's health surface will reflect the cascade via the
// probe loop (PR 69).
return;
}
if (resp.Results.Count == 0) return;
var refs = resp.Results[0].References;
foreach (var rf in refs)
{
if (discovered() >= _options.MaxDiscoveredNodes) break;
var childId = ExpandedNodeId.ToNodeId(rf.NodeId, session.NamespaceUris);
if (NodeId.IsNull(childId)) continue;
var browseName = rf.BrowseName?.Name ?? childId.ToString();
var displayName = rf.DisplayName?.Text ?? browseName;
if (rf.NodeClass == NodeClass.Object)
{
var subFolder = folder.Folder(browseName, displayName);
increment();
await BrowseRecursiveAsync(session, childId, subFolder, visited,
depth + 1, discovered, increment, ct).ConfigureAwait(false);
}
else if (rf.NodeClass == NodeClass.Variable)
{
// Serialize the NodeId so the IReadable/IWritable surface receives a
// round-trippable string. Deferring the DataType + AccessLevel fetch to a
// follow-up PR — initial browse uses a conservative ViewOnly + Int32 default.
var nodeIdString = childId.ToString() ?? string.Empty;
folder.Variable(browseName, displayName, new DriverAttributeInfo(
FullName: nodeIdString,
DriverDataType: DriverDataType.Int32,
IsArray: false,
ArrayDim: null,
SecurityClass: SecurityClassification.ViewOnly,
IsHistorized: false,
IsAlarm: false));
increment();
}
}
}
// ---- ISubscribable ----
public async Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var session = RequireSession();
var id = Interlocked.Increment(ref _nextSubscriptionId);
var handle = new OpcUaSubscriptionHandle(id);
// 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;
var subscription = new Subscription(telemetry: null!, new SubscriptionOptions
{
DisplayName = $"opcua-sub-{id}",
PublishingInterval = intervalMs,
KeepAliveCount = 10,
LifetimeCount = 1000,
MaxNotificationsPerPublish = 0,
PublishingEnabled = true,
Priority = 0,
TimestampsToReturn = TimestampsToReturn.Both,
});
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
session.AddSubscription(subscription);
await subscription.CreateAsync(cancellationToken).ConfigureAwait(false);
foreach (var fullRef in fullReferences)
{
if (!TryParseNodeId(session, fullRef, out var nodeId)) continue;
// The tag string is routed through MonitoredItem.Handle so the Notification
// handler can identify which tag changed without an extra lookup.
var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions
{
DisplayName = fullRef,
StartNodeId = nodeId,
AttributeId = Attributes.Value,
MonitoringMode = MonitoringMode.Reporting,
SamplingInterval = intervalMs,
QueueSize = 1,
DiscardOldest = true,
})
{
Handle = fullRef,
};
item.Notification += (mi, args) => OnMonitoredItemNotification(handle, mi, args);
subscription.AddItem(item);
}
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
_subscriptions[id] = new RemoteSubscription(subscription, handle);
}
finally { _gate.Release(); }
return handle;
}
public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is not OpcUaSubscriptionHandle h) return;
if (!_subscriptions.TryRemove(h.Id, out var rs)) return;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort — the subscription may already be gone on reconnect */ }
}
finally { _gate.Release(); }
}
private void OnMonitoredItemNotification(OpcUaSubscriptionHandle handle, MonitoredItem item, MonitoredItemNotificationEventArgs args)
{
// args.NotificationValue arrives as a MonitoredItemNotification for value-change
// subscriptions; extract its DataValue. The Handle property carries our tag string.
if (args.NotificationValue is not MonitoredItemNotification mn) return;
var dv = mn.Value;
if (dv is null) return;
var fullRef = (item.Handle as string) ?? item.DisplayName ?? string.Empty;
var snapshot = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot));
}
private sealed record RemoteSubscription(Subscription Subscription, OpcUaSubscriptionHandle Handle);
private sealed record OpcUaSubscriptionHandle(long Id) : ISubscriptionHandle
{
public string DiagnosticId => $"opcua-sub-{Id}";
}
// ---- IHostConnectivityProbe ----
/// <summary>Endpoint-URL-keyed host identity for the Admin /hosts dashboard.</summary>
public string HostName => _options.EndpointUrl;
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
{
lock (_probeLock)
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
private void TransitionTo(HostState newState)
{
HostState old;
lock (_probeLock)
{
old = _hostState;
if (old == newState) return;
_hostState = newState;
_hostStateChangedUtc = DateTime.UtcNow;
}
OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState));
}
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
public async ValueTask DisposeAsync()

View File

@@ -62,6 +62,30 @@ public sealed class OpcUaClientDriverOptions
/// <summary>Connect + per-operation timeout.</summary>
public TimeSpan Timeout { get; init; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Root NodeId to mirror. Default <c>null</c> = <c>ObjectsFolder</c> (i=85). Set to
/// a scoped root to restrict the address space the driver exposes locally — useful
/// when the remote server has tens of thousands of nodes and only a subset is
/// needed downstream.
/// </summary>
public string? BrowseRoot { get; init; }
/// <summary>
/// Cap on total nodes discovered during <c>DiscoverAsync</c>. Default 10_000 —
/// bounds memory on runaway remote servers without being so low that normal
/// deployments hit it. When the cap is reached discovery stops and a warning is
/// written to the driver health surface; the partially-discovered tree is still
/// projected into the local address space.
/// </summary>
public int MaxDiscoveredNodes { get; init; } = 10_000;
/// <summary>
/// Max hierarchical depth of the browse. Default 10 — deep enough for realistic
/// OPC UA information models, shallow enough that cyclic graphs can't spin the
/// browse forever.
/// </summary>
public int MaxBrowseDepth { get; init; } = 10;
}
/// <summary>OPC UA message security mode.</summary>

View File

@@ -0,0 +1,55 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Scaffold tests for <see cref="OpcUaClientDriver"/>'s <see cref="ITagDiscovery"/>
/// surface that don't require a live remote server. Live-browse coverage lands in a
/// follow-up PR once the in-process OPC UA server fixture is scaffolded.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientDiscoveryTests
{
[Fact]
public async Task DiscoverAsync_without_initialize_throws_InvalidOperationException()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-disco");
var builder = new NullAddressSpaceBuilder();
await Should.ThrowAsync<InvalidOperationException>(async () =>
await drv.DiscoverAsync(builder, TestContext.Current.CancellationToken));
}
[Fact]
public void DiscoverAsync_rejects_null_builder()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-disco");
Should.ThrowAsync<ArgumentNullException>(async () =>
await drv.DiscoverAsync(null!, TestContext.Current.CancellationToken));
}
[Fact]
public void Discovery_caps_are_sensible_defaults()
{
var opts = new OpcUaClientDriverOptions();
opts.MaxDiscoveredNodes.ShouldBe(10_000, "bounds memory on runaway servers without clipping normal models");
opts.MaxBrowseDepth.ShouldBe(10, "deep enough for realistic info models; shallow enough for cycle safety");
opts.BrowseRoot.ShouldBeNull("null = default to ObjectsFolder i=85");
}
private sealed class NullAddressSpaceBuilder : IAddressSpaceBuilder
{
public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
=> new StubHandle();
public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
public void AttachAlarmCondition(IVariableHandle sourceVariable, string alarmName, DriverAttributeInfo alarmInfo) { }
private sealed class StubHandle : IVariableHandle
{
public string FullReference => "stub";
public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => throw new NotSupportedException();
}
}
}

View File

@@ -0,0 +1,50 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Scaffold tests for <c>ISubscribable</c> + <c>IHostConnectivityProbe</c> that don't
/// need a live remote server. Live-session tests (subscribe/unsubscribe round-trip,
/// keep-alive transitions) land in a follow-up PR once the in-process OPC UA server
/// fixture is scaffolded.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientSubscribeAndProbeTests
{
[Fact]
public async Task SubscribeAsync_without_initialize_throws_InvalidOperationException()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-sub-uninit");
await Should.ThrowAsync<InvalidOperationException>(async () =>
await drv.SubscribeAsync(["ns=2;s=Demo"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken));
}
[Fact]
public async Task UnsubscribeAsync_with_unknown_handle_is_noop()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-sub-unknown");
// UnsubscribeAsync returns cleanly for handles it doesn't recognise — protects against
// the caller's race with server-side cleanup after a session drop.
await drv.UnsubscribeAsync(new FakeHandle(), TestContext.Current.CancellationToken);
}
[Fact]
public void GetHostStatuses_returns_endpoint_url_row_pre_init()
{
using var drv = new OpcUaClientDriver(
new OpcUaClientDriverOptions { EndpointUrl = "opc.tcp://plc.example:4840" },
"opcua-hosts");
var rows = drv.GetHostStatuses();
rows.Count.ShouldBe(1);
rows[0].HostName.ShouldBe("opc.tcp://plc.example:4840",
"host identity mirrors the endpoint URL so the Admin /hosts dashboard can link back to the remote server");
rows[0].State.ShouldBe(HostState.Unknown);
}
private sealed class FakeHandle : ISubscriptionHandle
{
public string DiagnosticId => "fake";
}
}