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

Auto: opcuaclient-2 — per-tag advanced subscription tuning

Browse Source 
Closes #274
This commit is contained in:
Joseph Doherty
2026-04-25 15:25:20 -04:00
parent bf200e813e
commit fae00749ca
4 changed files with 400 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
src/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
View File

@@ -851,8 +851,20 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
// ---- ISubscribable ----
public async Task<ISubscriptionHandle> SubscribeAsync(
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
// Route the simple-string overload through the per-tag overload with all knobs at
// their defaults. Single code path for subscription create — keeps the wire-side
// identical for callers that don't need per-tag tuning.
var specs = new MonitoredTagSpec[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
specs[i] = new MonitoredTagSpec(fullReferences[i]);
return SubscribeAsync(specs, publishingInterval, cancellationToken);
}
public async Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<MonitoredTagSpec> tags, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var session = RequireSession();
var id = Interlocked.Increment(ref _nextSubscriptionId);
@@ -885,29 +897,28 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
session.AddSubscription(subscription);
await subscription.CreateAsync(cancellationToken).ConfigureAwait(false);
foreach (var fullRef in fullReferences)
foreach (var spec in tags)
{
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);
if (!TryParseNodeId(session, spec.TagName, out var nodeId)) continue;
var monItem = BuildMonitoredItem(spec, nodeId, intervalMs);
monItem.Notification += (mi, args) => OnMonitoredItemNotification(handle, mi, args);
subscription.AddItem(monItem);
}
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
try
{
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
}
catch (Opc.Ua.ServiceResultException sre)
{
// PercentDeadband requires the server to expose EURange on the variable; if
// it isn't set the server returns BadFilterNotAllowed during item creation.
// We swallow the exception here so other items in the batch still get created
// — per-item failure surfaces through MonitoredItem.Status.Error rather than
// tearing down the whole subscription.
if (sre.StatusCode != StatusCodes.BadFilterNotAllowed) throw;
}
_subscriptions[id] = new RemoteSubscription(subscription, handle);
}
finally { _gate.Release(); }
@@ -915,6 +926,84 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
return handle;
}
/// <summary>
/// Map a <see cref="MonitoredTagSpec"/> to a SDK <see cref="MonitoredItem"/> with the
/// per-tag knobs applied. Defaults match the original hard-coded values
/// (Reporting / SamplingInterval=publishInterval / QueueSize=1 / DiscardOldest=true)
/// so a spec with all knobs <c>null</c> behaves identically to the legacy path.
/// </summary>
internal static MonitoredItem BuildMonitoredItem(MonitoredTagSpec spec, NodeId nodeId, int defaultIntervalMs)
{
var sampling = spec.SamplingIntervalMs.HasValue ? (int)spec.SamplingIntervalMs.Value : defaultIntervalMs;
var queueSize = spec.QueueSize ?? 1u;
var discardOldest = spec.DiscardOldest ?? true;
var monitoringMode = spec.MonitoringMode is { } mm ? MapMonitoringMode(mm) : MonitoringMode.Reporting;
var filter = BuildDataChangeFilter(spec.DataChangeFilter);
var options = new MonitoredItemOptions
{
DisplayName = spec.TagName,
StartNodeId = nodeId,
AttributeId = Attributes.Value,
MonitoringMode = monitoringMode,
SamplingInterval = sampling,
QueueSize = queueSize,
DiscardOldest = discardOldest,
Filter = filter,
};
return new MonitoredItem(telemetry: null!, options)
{
// The tag string is routed through MonitoredItem.Handle so the Notification
// handler can identify which tag changed without an extra lookup.
Handle = spec.TagName,
};
}
/// <summary>
/// Build the OPC UA <see cref="DataChangeFilter"/> from a <see cref="DataChangeFilterSpec"/>,
/// or return <c>null</c> if the caller didn't supply a filter. PercentDeadband requires
/// server-side EURange — if the server rejects with BadFilterNotAllowed, the caller's
/// <c>SubscribeAsync</c> swallows it so other items in the batch still get created.
/// </summary>
internal static DataChangeFilter? BuildDataChangeFilter(DataChangeFilterSpec? spec)
{
if (spec is null) return null;
return new DataChangeFilter
{
Trigger = MapTrigger(spec.Trigger),
DeadbandType = (uint)MapDeadbandType(spec.DeadbandType),
DeadbandValue = spec.DeadbandValue,
};
}
/// <summary>Map our SDK-free <see cref="SubscriptionMonitoringMode"/> to the OPC UA SDK's enum.</summary>
internal static MonitoringMode MapMonitoringMode(SubscriptionMonitoringMode mode) => mode switch
{
SubscriptionMonitoringMode.Disabled => MonitoringMode.Disabled,
SubscriptionMonitoringMode.Sampling => MonitoringMode.Sampling,
SubscriptionMonitoringMode.Reporting => MonitoringMode.Reporting,
_ => MonitoringMode.Reporting,
};
/// <summary>Map our <see cref="Core.Abstractions.DataChangeTrigger"/> to the SDK enum.</summary>
internal static Opc.Ua.DataChangeTrigger MapTrigger(Core.Abstractions.DataChangeTrigger trigger) => trigger switch
{
Core.Abstractions.DataChangeTrigger.Status => Opc.Ua.DataChangeTrigger.Status,
Core.Abstractions.DataChangeTrigger.StatusValue => Opc.Ua.DataChangeTrigger.StatusValue,
Core.Abstractions.DataChangeTrigger.StatusValueTimestamp => Opc.Ua.DataChangeTrigger.StatusValueTimestamp,
_ => Opc.Ua.DataChangeTrigger.StatusValue,
};
/// <summary>Map our <see cref="Core.Abstractions.DeadbandType"/> to the SDK enum.</summary>
internal static Opc.Ua.DeadbandType MapDeadbandType(Core.Abstractions.DeadbandType type) => type switch
{
Core.Abstractions.DeadbandType.None => Opc.Ua.DeadbandType.None,
Core.Abstractions.DeadbandType.Absolute => Opc.Ua.DeadbandType.Absolute,
Core.Abstractions.DeadbandType.Percent => Opc.Ua.DeadbandType.Percent,
_ => Opc.Ua.DeadbandType.None,
};
public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is not OpcUaSubscriptionHandle h) return;