Merge b640-619alm: OpcUaClient native alarms route to RawPath conditions + ConditionRefresh; docker-dev opc-plc seed (#619)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -92,6 +92,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
// ---- IAlarmSource state ----
|
||||
|
||||
private readonly System.Collections.Concurrent.ConcurrentDictionary<long, RemoteAlarmSubscription> _alarmSubscriptions = new();
|
||||
|
||||
/// <summary>#619: RawPath → the upstream condition NodeId last seen for it. The host acknowledges by
|
||||
/// RawPath (its only identity for the condition) but Part 9 Acknowledge must be called on the CONDITION
|
||||
/// node, which differs from the authored source node.</summary>
|
||||
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, AlarmAckTarget> _alarmConditionByRawPath =
|
||||
new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>#619 review: the (Active, Acked) state last told to the host per reported condition id, from
|
||||
/// which <see cref="AlarmTransitions"/> picks the Kind sequence that moves the host's projector to the
|
||||
/// upstream event's state.</summary>
|
||||
private readonly Dictionary<string, AlarmToldState> _alarmStateByRawPath = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>#619 re-review: serialises the read-compute-emit over <see cref="_alarmStateByRawPath"/>. Event
|
||||
/// notifications arrive on SDK publish threads (and, across a resubscribe, from two subscriptions at once), so
|
||||
/// a lock-free prior/target exchange could emit two kind sequences computed from the same prior.</summary>
|
||||
private readonly object _alarmStateLock = new();
|
||||
|
||||
/// <summary>#619 re-review: the ref set the last <see cref="SubscribeAlarmsAsync"/> carried. Survives
|
||||
/// <see cref="ShutdownAsync"/> deliberately — see <see cref="BeginAlarmSubscription"/>.</summary>
|
||||
private HashSet<string>? _lastAlarmRefSet;
|
||||
|
||||
/// <summary>Latched once the upstream server has refused ConditionRefresh as unsupported (logged once).</summary>
|
||||
private int _conditionRefreshUnsupportedLogged;
|
||||
private long _nextAlarmSubscriptionId;
|
||||
|
||||
/// <summary>Occurs when an alarm event is received from the remote OPC UA server.</summary>
|
||||
@@ -879,6 +902,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
catch { /* best-effort */ }
|
||||
}
|
||||
_alarmSubscriptions.Clear();
|
||||
// #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A
|
||||
// ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the
|
||||
// same set over condition nodes it did NOT rebuild; forgetting here would make the post-subscribe
|
||||
// ConditionRefresh re-raise every active+acked alarm (an un-ack flicker). BeginAlarmSubscription forgets
|
||||
// them when the set changes, which is when the host did rebuild.
|
||||
|
||||
await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
|
||||
_namespaceMap = null;
|
||||
@@ -1621,6 +1649,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
private const int AlarmFieldSeverity = 4;
|
||||
private const int AlarmFieldTime = 5;
|
||||
private const int AlarmFieldConditionId = 6;
|
||||
private const int AlarmFieldActiveState = 7;
|
||||
private const int AlarmFieldAckedState = 8;
|
||||
private const int AlarmFieldBranchId = 9;
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
|
||||
@@ -1635,11 +1666,24 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
|
||||
// Pre-resolve the source-node filter set so the per-event notification handler can
|
||||
// match in O(1) without re-parsing on every event.
|
||||
var sourceFilter = new HashSet<string>(sourceNodeIds, StringComparer.Ordinal);
|
||||
// #619: under v3 the host passes RawPaths, never upstream node ids. Translate each to the upstream
|
||||
// source id its tag authored (TagConfig.nodeId) so the per-event match compares like with like.
|
||||
// Keys are server-stable (AlarmIndexKey) so a namespace reorder across an in-place SDK reconnect —
|
||||
// which transfers this subscription without re-running SubscribeAlarmsAsync — cannot orphan them.
|
||||
var sessionForResolve = Session;
|
||||
var mapForResolve = _namespaceMap;
|
||||
var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw =>
|
||||
{
|
||||
if (sessionForResolve is null) return null;
|
||||
if (!TryResolveTagNode(sessionForResolve, raw, out var live)
|
||||
&& !NamespaceMap.TryResolve(sessionForResolve, raw, out live)) return null;
|
||||
return AlarmIndexKey(mapForResolve, live);
|
||||
});
|
||||
BeginAlarmSubscription(sourceNodeIds);
|
||||
|
||||
var subscription = new Subscription(telemetry: null!, new SubscriptionOptions
|
||||
{
|
||||
DisplayName = $"opcua-alarm-sub-{id}",
|
||||
DisplayName = $"{AlarmSubscriptionNamePrefix}{id}",
|
||||
PublishingInterval = 500, // 500ms — alarms don't need fast polling; the server pushes
|
||||
KeepAliveCount = _options.SubscriptionKeepAliveCount,
|
||||
LifetimeCount = _options.SubscriptionLifetimeCount,
|
||||
@@ -1674,6 +1718,28 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
BrowsePath = [], // empty path = the condition node itself
|
||||
AttributeId = Attributes.NodeId,
|
||||
});
|
||||
// #619 review: the host derives Active/Acked solely from AlarmEventArgs.Kind, so the condition state
|
||||
// must be selected. TWO-element browse paths — a single "ActiveState/Id" QualifiedName names nothing.
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand
|
||||
{
|
||||
TypeDefinitionId = ObjectTypeIds.AlarmConditionType,
|
||||
BrowsePath = [new QualifiedName(BrowseNames.ActiveState), new QualifiedName(BrowseNames.Id)],
|
||||
AttributeId = Attributes.Value,
|
||||
});
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand
|
||||
{
|
||||
TypeDefinitionId = ObjectTypeIds.AcknowledgeableConditionType,
|
||||
BrowsePath = [new QualifiedName(BrowseNames.AckedState), new QualifiedName(BrowseNames.Id)],
|
||||
AttributeId = Attributes.Value,
|
||||
});
|
||||
// #619 re-review: BranchId — a non-null one marks a condition BRANCH event (a previous state, with its own
|
||||
// EventId). Only the trunk (null BranchId) is the condition's current state.
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand
|
||||
{
|
||||
TypeDefinitionId = ObjectTypeIds.ConditionType,
|
||||
BrowsePath = [new QualifiedName(BrowseNames.BranchId)],
|
||||
AttributeId = Attributes.Value,
|
||||
});
|
||||
|
||||
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
@@ -1710,6 +1776,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
_alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler);
|
||||
|
||||
// #619: a new event item only reports FUTURE transitions — a condition already active upstream would
|
||||
// stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's
|
||||
// current state through the normal path (deduplicated against the last-told state).
|
||||
await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
finally { _gate.Release(); }
|
||||
|
||||
@@ -1765,15 +1836,30 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
var callRequests = new CallMethodRequestCollection();
|
||||
foreach (var ack in acknowledgements)
|
||||
{
|
||||
// ConditionId is an upstream condition-node id captured from the alarm event (not a
|
||||
// tag RawPath), so it re-binds directly against the session via the static resolve.
|
||||
if (!NamespaceMap.TryResolve(session, ack.ConditionId, out var conditionId)) continue;
|
||||
// #619: the host acks by RawPath; Acknowledge must hit the upstream CONDITION node.
|
||||
// The remembered target pairs the condition node with the EventId of the event that named it,
|
||||
// so the ack can never mix one condition's node with another's EventId. Nothing seen yet ⇒ the
|
||||
// tag's authored node (correct when the tag was authored on the condition node), else the id
|
||||
// verbatim (pre-v3 callers that pass an upstream condition id directly).
|
||||
NodeId conditionId;
|
||||
byte[] eventId;
|
||||
if (_alarmConditionByRawPath.TryGetValue(ack.ConditionId, out var seen))
|
||||
{
|
||||
if (!NamespaceMap.TryResolve(session, seen.ConditionRef, out conditionId)) continue;
|
||||
eventId = seen.EventId;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (!TryResolveTagNode(session, ack.ConditionId, out conditionId)
|
||||
&& !NamespaceMap.TryResolve(session, ack.ConditionId, out conditionId)) continue;
|
||||
eventId = Array.Empty<byte>();
|
||||
}
|
||||
callRequests.Add(new CallMethodRequest
|
||||
{
|
||||
ObjectId = conditionId,
|
||||
MethodId = MethodIds.AcknowledgeableConditionType_Acknowledge,
|
||||
InputArguments = [
|
||||
new Variant(Array.Empty<byte>()), // EventId — server-side best-effort; empty resolves to 'most recent'
|
||||
new Variant(eventId), // Part 9 §5.7.3: the EventId being acknowledged (empty only when none seen)
|
||||
new Variant(new LocalizedText(ack.Comment ?? string.Empty)),
|
||||
],
|
||||
});
|
||||
@@ -1831,32 +1917,339 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
|
||||
private void OnEventNotification(
|
||||
OpcUaAlarmSubscriptionHandle handle,
|
||||
HashSet<string> sourceFilter,
|
||||
IReadOnlyDictionary<string, string> sourceFilter,
|
||||
MonitoredItem item,
|
||||
MonitoredItemNotificationEventArgs args)
|
||||
{
|
||||
if (args.NotificationValue is not EventFieldList efl) return;
|
||||
if (efl.EventFields.Count <= AlarmFieldConditionId) return;
|
||||
RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
|
||||
}
|
||||
|
||||
var sourceNode = efl.EventFields[AlarmFieldSourceNode].Value?.ToString() ?? string.Empty;
|
||||
if (sourceFilter.Count > 0 && !sourceFilter.Contains(sourceNode)) return;
|
||||
/// <summary>
|
||||
/// #619 re-review: open (or re-open) the host's alarm subscription over <paramref name="sourceNodeIds"/>.
|
||||
/// A CHANGED set means the host rebuilt its address space (an alarm-bearing delta always does), recreating
|
||||
/// every condition node inactive+acked — so every last-told state is forgotten and the post-subscribe
|
||||
/// ConditionRefresh re-reports from that default. The SAME set (a reinit / reconnect re-subscribe) keeps
|
||||
/// them, so the refresh copy of an already-reported state emits nothing. Ack targets of refs no longer
|
||||
/// subscribed are pruned either way.
|
||||
/// </summary>
|
||||
/// <param name="sourceNodeIds">The refs the host is subscribing.</param>
|
||||
internal void BeginAlarmSubscription(IReadOnlyList<string> sourceNodeIds)
|
||||
{
|
||||
var set = new HashSet<string>(sourceNodeIds, StringComparer.Ordinal);
|
||||
lock (_alarmStateLock)
|
||||
{
|
||||
if (_lastAlarmRefSet is null || !_lastAlarmRefSet.SetEquals(set)) _alarmStateByRawPath.Clear();
|
||||
_lastAlarmRefSet = set;
|
||||
foreach (var known in _alarmConditionByRawPath.Keys)
|
||||
if (!set.Contains(known)) _alarmConditionByRawPath.TryRemove(known, out _);
|
||||
}
|
||||
}
|
||||
|
||||
var eventType = efl.EventFields[AlarmFieldEventType].Value?.ToString() ?? "BaseEventType";
|
||||
var message = (efl.EventFields[AlarmFieldMessage].Value as LocalizedText)?.Text ?? string.Empty;
|
||||
var severity = efl.EventFields[AlarmFieldSeverity].Value is ushort sev ? sev : (ushort)0;
|
||||
var time = efl.EventFields[AlarmFieldTime].Value is DateTime t ? t : DateTime.UtcNow;
|
||||
var conditionId = efl.EventFields[AlarmFieldConditionId].Value?.ToString() ?? string.Empty;
|
||||
/// <summary>
|
||||
/// #619 — route one upstream event: drop refresh markers (a RefreshRequired asks for a new refresh), project
|
||||
/// it, and emit the kinds that move the host from the state last told to the event's. An event that changes
|
||||
/// nothing the host shows (a ConditionRefresh copy) emits nothing.
|
||||
/// </summary>
|
||||
/// <param name="fields">The event's fields, in select-clause order.</param>
|
||||
/// <param name="sourceFilter">The subscription's upstream-key → RawPath index.</param>
|
||||
/// <param name="handle">The subscription handle to stamp.</param>
|
||||
internal void RouteAlarmEvent(
|
||||
IReadOnlyList<Variant> fields, IReadOnlyDictionary<string, string> sourceFilter, IAlarmSubscriptionHandle handle)
|
||||
{
|
||||
if (IsRefreshRequired(fields))
|
||||
{
|
||||
// Part 9 §5.11.7: the server lost events (queue overflow / restart) and asks for a refresh.
|
||||
_ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
|
||||
return;
|
||||
}
|
||||
|
||||
OnAlarmEvent?.Invoke(this, new AlarmEventArgs(
|
||||
// Projection, the per-RawPath bookkeeping AND the emission run under one lock, so two notifications for
|
||||
// one condition (two publish threads; two subscriptions across a resubscribe) can never compute their
|
||||
// kinds from the same prior, nor reach the host out of order. The host's handler only marshals.
|
||||
lock (_alarmStateLock)
|
||||
{
|
||||
var e = ProjectEvent(fields, sourceFilter, handle,
|
||||
n => AlarmIndexKey(_namespaceMap, n), (raw, target) => _alarmConditionByRawPath[raw] = target,
|
||||
out var state);
|
||||
if (e is null) return;
|
||||
if (state is not { } target)
|
||||
{
|
||||
// Not a condition (no ActiveState/AckedState) — a message/severity-only event, as before.
|
||||
OnAlarmEvent?.Invoke(this, e);
|
||||
return;
|
||||
}
|
||||
|
||||
// Fresh prior = the host projector's own default (inactive, acked).
|
||||
var told = new AlarmToldState(target.Active, target.Acked, e.Message, e.OpcUaSeverity);
|
||||
var prior = _alarmStateByRawPath.TryGetValue(e.ConditionId, out var p)
|
||||
? p
|
||||
: new AlarmToldState(false, true, null, null);
|
||||
_alarmStateByRawPath[e.ConditionId] = told;
|
||||
|
||||
var kinds = AlarmTransitions((prior.Active, prior.Acked), target);
|
||||
if (kinds.Count == 0)
|
||||
{
|
||||
// Same state: a message/severity change still reaches the host; an identical copy does not.
|
||||
if (prior == told) return;
|
||||
kinds = [AlarmTransitionKind.Unspecified];
|
||||
}
|
||||
foreach (var kind in kinds)
|
||||
OnAlarmEvent?.Invoke(this, e with { Kind = kind });
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>#619 re-review — what the host was last told about one condition: its state plus the message and
|
||||
/// exact severity, so a refresh copy is recognised as a duplicate.</summary>
|
||||
/// <param name="Active">ActiveState/Id last told.</param>
|
||||
/// <param name="Acked">AckedState/Id last told.</param>
|
||||
/// <param name="Message">Message last told (null before any event).</param>
|
||||
/// <param name="Severity">Exact severity last told (null before any event).</param>
|
||||
private readonly record struct AlarmToldState(bool Active, bool Acked, string? Message, int? Severity);
|
||||
|
||||
/// <summary>True for a <c>RefreshRequiredEventType</c> event (the server asks the client to ConditionRefresh).</summary>
|
||||
/// <param name="fields">The event's fields.</param>
|
||||
/// <returns>Whether the event is a RefreshRequired marker.</returns>
|
||||
internal static bool IsRefreshRequired(IReadOnlyList<Variant> fields) =>
|
||||
fields.Count > AlarmFieldEventType && fields[AlarmFieldEventType].Value is NodeId t
|
||||
&& t == ObjectTypeIds.RefreshRequiredEventType;
|
||||
|
||||
/// <summary>True for the three ConditionRefresh marker event types — never alarms.</summary>
|
||||
/// <param name="eventType">The event's EventType.</param>
|
||||
/// <returns>Whether it is a refresh marker.</returns>
|
||||
private static bool IsRefreshMarker(object? eventType) =>
|
||||
eventType is NodeId t
|
||||
&& (t == ObjectTypeIds.RefreshStartEventType || t == ObjectTypeIds.RefreshEndEventType
|
||||
|| t == ObjectTypeIds.RefreshRequiredEventType);
|
||||
|
||||
/// <summary>A ConditionRefresh refusal meaning "this server does not do refresh" (tolerated, logged once).</summary>
|
||||
/// <param name="code">The refusal status.</param>
|
||||
/// <returns>Whether the server simply lacks ConditionRefresh.</returns>
|
||||
internal static bool IsConditionRefreshUnsupported(StatusCode code) =>
|
||||
code == StatusCodes.BadNotSupported || code == StatusCodes.BadMethodInvalid
|
||||
|| code == StatusCodes.BadNotImplemented || code == StatusCodes.BadServiceUnsupported
|
||||
|| code == StatusCodes.BadNodeIdUnknown;
|
||||
|
||||
/// <summary>
|
||||
/// #619 — call ConditionRefresh for <paramref name="subscription"/> so conditions already active upstream
|
||||
/// are reported. Never throws: an unsupported server is logged once, any other failure warns.
|
||||
/// </summary>
|
||||
/// <param name="subscription">The (created) alarm subscription.</param>
|
||||
/// <param name="cancellationToken">Cancellation.</param>
|
||||
/// <returns>A task completing when the refresh call returns.</returns>
|
||||
private async Task TryConditionRefreshAsync(Subscription subscription, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await subscription.ConditionRefreshAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (ServiceResultException ex) when (IsConditionRefreshUnsupported(ex.StatusCode))
|
||||
{
|
||||
if (Interlocked.Exchange(ref _conditionRefreshUnsupportedLogged, 1) == 0)
|
||||
_logger.LogInformation(
|
||||
"OpcUaClientDriver '{DriverInstanceId}': upstream server does not support ConditionRefresh " +
|
||||
"({StatusCode}); conditions already active are reported only on their next transition.",
|
||||
_driverInstanceId, ex.StatusCode);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"OpcUaClientDriver '{DriverInstanceId}': ConditionRefresh on alarm subscription {SubscriptionId} failed; " +
|
||||
"conditions already active upstream may not be reported until their next transition.",
|
||||
_driverInstanceId, subscription.Id);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// The caller cancelled; the subscription itself is in place.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #619 — ConditionRefresh every alarm subscription on the current session: after an in-place reconnect
|
||||
/// (transferred or SDK-recreated subscriptions — matched by display name, since a recreate clones the
|
||||
/// <see cref="Subscription"/> objects) and on a RefreshRequired event. Events missed during the outage
|
||||
/// reach the host through the normal deduplicated path.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Cancellation.</param>
|
||||
/// <returns>A task completing when every refresh call has returned.</returns>
|
||||
private async Task RefreshAllAlarmSubscriptionsAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_alarmSubscriptions.IsEmpty) return;
|
||||
try
|
||||
{
|
||||
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (ObjectDisposedException) { return; }
|
||||
try
|
||||
{
|
||||
var session = Session;
|
||||
if (session is null) return;
|
||||
foreach (var sub in session.Subscriptions.ToList())
|
||||
{
|
||||
if (sub.DisplayName?.StartsWith(AlarmSubscriptionNamePrefix, StringComparison.Ordinal) == true && sub.Created)
|
||||
await TryConditionRefreshAsync(sub, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': alarm ConditionRefresh sweep failed.",
|
||||
_driverInstanceId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
try { _gate.Release(); } catch (ObjectDisposedException) { }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Display-name prefix of every alarm subscription (how a refresh sweep finds them on a session).</summary>
|
||||
private const string AlarmSubscriptionNamePrefix = "opcua-alarm-sub-";
|
||||
|
||||
/// <summary>
|
||||
/// #619 review — the <see cref="AlarmTransitionKind"/> sequence that moves the host's
|
||||
/// <c>NativeAlarmProjector</c> (which knows only kinds: Raise ⇒ active+unacked, Acknowledge ⇒ acked,
|
||||
/// Clear ⇒ inactive keeping acked) from <paramref name="prior"/> to <paramref name="target"/>. A single
|
||||
/// upstream event can need two (an alarm that went active and was acknowledged between two notifications).
|
||||
/// No state change ⇒ empty (the caller decides whether a message/severity change warrants an Unspecified).
|
||||
/// </summary>
|
||||
/// <param name="prior">The state last told to the host.</param>
|
||||
/// <param name="target">The upstream event's ActiveState/Id + AckedState/Id.</param>
|
||||
/// <returns>The kinds to emit, in order.</returns>
|
||||
internal static IReadOnlyList<AlarmTransitionKind> AlarmTransitions(
|
||||
(bool Active, bool Acked) prior, (bool Active, bool Acked) target)
|
||||
{
|
||||
if (prior == target) return [];
|
||||
return target switch
|
||||
{
|
||||
// Active+unacked: a fresh raise, or a re-arm of an active-acked condition.
|
||||
(true, false) => [prior.Active ? AlarmTransitionKind.Retrigger : AlarmTransitionKind.Raise],
|
||||
// Active+acked: raise first when it was inactive (Raise always lands unacked).
|
||||
(true, true) => prior.Active
|
||||
? [AlarmTransitionKind.Acknowledge]
|
||||
: [AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge],
|
||||
// Inactive+acked: Clear keeps the prior ack, so ack after it when that was unacked.
|
||||
(false, true) => prior.Active
|
||||
? prior.Acked ? [AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear, AlarmTransitionKind.Acknowledge]
|
||||
: [AlarmTransitionKind.Acknowledge],
|
||||
// Inactive+UNacked (went active and cleared unacknowledged — e.g. first seen by a ConditionRefresh):
|
||||
// the only kinds that un-ack are Raise/Retrigger, so pass through active, then Clear.
|
||||
(false, false) => prior.Active
|
||||
? prior.Acked ? [AlarmTransitionKind.Retrigger, AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear]
|
||||
: [AlarmTransitionKind.Raise, AlarmTransitionKind.Clear],
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #619 — index an alarm subscription's source refs by UPSTREAM source id. The host (v3) subscribes with
|
||||
/// the alarm tags' <b>RawPaths</b>; an upstream event carries its own source NodeId, so the match key is
|
||||
/// the upstream id the tag authored and the value is the RawPath to report back. A ref
|
||||
/// <paramref name="resolveUpstream"/> cannot resolve is kept verbatim (direct upstream-id use).
|
||||
/// </summary>
|
||||
/// <param name="refs">The refs handed to <c>SubscribeAlarmsAsync</c>.</param>
|
||||
/// <param name="resolveUpstream">RawPath → serialized live upstream NodeId, or null when unknown.</param>
|
||||
/// <returns>Upstream source id → ref to report.</returns>
|
||||
internal static Dictionary<string, string> BuildAlarmSourceIndex(
|
||||
IReadOnlyList<string> refs, Func<string, string?> resolveUpstream)
|
||||
{
|
||||
var index = new Dictionary<string, string>(StringComparer.Ordinal);
|
||||
foreach (var r in refs)
|
||||
index[resolveUpstream(r) ?? r] = r;
|
||||
return index;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #619 — project one upstream event into an <see cref="AlarmEventArgs"/>. <c>ConditionId</c> and
|
||||
/// <c>SourceNodeId</c> are the authored RawPath, because <c>DriverHostActor</c> routes native alarms on
|
||||
/// (driver, RawPath); the upstream condition node (needed to acknowledge) is handed to
|
||||
/// <paramref name="rememberCondition"/>. Null when the event's source is not an authored one.
|
||||
/// </summary>
|
||||
internal static AlarmEventArgs? ProjectEvent(
|
||||
IReadOnlyList<Variant> fields,
|
||||
IReadOnlyDictionary<string, string> sourceIndex,
|
||||
IAlarmSubscriptionHandle handle,
|
||||
Func<NodeId, string> keyOf,
|
||||
Action<string, AlarmAckTarget> rememberCondition) =>
|
||||
ProjectEvent(fields, sourceIndex, handle, keyOf, rememberCondition, out _);
|
||||
|
||||
/// <summary>As the five-argument overload, also returning the event's (ActiveState/Id, AckedState/Id) through
|
||||
/// <c>state</c> when it is an alarm condition (null otherwise).</summary>
|
||||
internal static AlarmEventArgs? ProjectEvent(
|
||||
IReadOnlyList<Variant> fields,
|
||||
IReadOnlyDictionary<string, string> sourceIndex,
|
||||
IAlarmSubscriptionHandle handle,
|
||||
Func<NodeId, string> keyOf,
|
||||
Action<string, AlarmAckTarget> rememberCondition,
|
||||
out (bool Active, bool Acked)? state)
|
||||
{
|
||||
state = null;
|
||||
if (fields.Count <= AlarmFieldConditionId) return null;
|
||||
// ConditionRefresh brackets its re-delivery with RefreshStart/RefreshEnd — markers, never alarms.
|
||||
if (IsRefreshMarker(fields[AlarmFieldEventType].Value)) return null;
|
||||
// A condition BRANCH event reports a previous state with its own EventId: projecting it would flip the
|
||||
// trunk's state, remembering it would make Acknowledge ack the branch.
|
||||
if (fields.Count > AlarmFieldBranchId && fields[AlarmFieldBranchId].Value is NodeId branch && !NodeId.IsNull(branch))
|
||||
return null;
|
||||
if (fields.Count > AlarmFieldAckedState
|
||||
&& fields[AlarmFieldActiveState].Value is bool active
|
||||
&& fields[AlarmFieldAckedState].Value is bool acked)
|
||||
state = (active, acked);
|
||||
|
||||
// Pass-through (no authored refs) keeps the raw serialized ids; an authored index compares
|
||||
// server-stable keys (see AlarmIndexKey), matching the CONDITION node first — the only way to tell
|
||||
// several conditions on one source apart — and then the source node.
|
||||
var sourceField = fields[AlarmFieldSourceNode].Value;
|
||||
var conditionField = fields[AlarmFieldConditionId].Value;
|
||||
string reported, upstreamCondition;
|
||||
if (sourceIndex.Count == 0)
|
||||
{
|
||||
reported = sourceField?.ToString() ?? string.Empty;
|
||||
upstreamCondition = conditionField?.ToString() ?? string.Empty;
|
||||
}
|
||||
else
|
||||
{
|
||||
upstreamCondition = conditionField is NodeId cn && !NodeId.IsNull(cn) ? keyOf(cn) : string.Empty;
|
||||
var sourceKey = sourceField is NodeId sn && !NodeId.IsNull(sn) ? keyOf(sn) : string.Empty;
|
||||
if (!(upstreamCondition.Length > 0 && sourceIndex.TryGetValue(upstreamCondition, out reported!))
|
||||
&& !(sourceKey.Length > 0 && sourceIndex.TryGetValue(sourceKey, out reported!)))
|
||||
return null;
|
||||
}
|
||||
|
||||
var eventType = fields[AlarmFieldEventType].Value?.ToString() ?? "BaseEventType";
|
||||
var message = (fields[AlarmFieldMessage].Value as LocalizedText)?.Text ?? string.Empty;
|
||||
var severity = fields[AlarmFieldSeverity].Value is ushort sev ? sev : (ushort)0;
|
||||
var time = fields[AlarmFieldTime].Value is DateTime t ? t : DateTime.UtcNow;
|
||||
|
||||
if (sourceIndex.Count != 0 && upstreamCondition.Length > 0)
|
||||
rememberCondition(reported, new AlarmAckTarget(
|
||||
upstreamCondition,
|
||||
// A copy: the decoded notification's array is not ours to alias for the ack's lifetime.
|
||||
(fields[AlarmFieldEventId].Value as byte[])?.ToArray() ?? Array.Empty<byte>()));
|
||||
|
||||
return new AlarmEventArgs(
|
||||
SubscriptionHandle: handle,
|
||||
SourceNodeId: sourceNode,
|
||||
ConditionId: conditionId,
|
||||
SourceNodeId: reported,
|
||||
ConditionId: sourceIndex.Count == 0 ? upstreamCondition : reported,
|
||||
AlarmType: eventType,
|
||||
Message: message,
|
||||
Severity: MapSeverity(severity),
|
||||
SourceTimestampUtc: time));
|
||||
SourceTimestampUtc: time,
|
||||
// #626: the upstream severity is already the exact Part 9 value — carry it, don't re-bucket.
|
||||
OpcUaSeverity: severity == 0 ? null : severity);
|
||||
}
|
||||
|
||||
/// <summary>#619 — what Acknowledge needs for one RawPath: the upstream condition node (server-stable
|
||||
/// reference) and the EventId of the event that named it.</summary>
|
||||
/// <param name="ConditionRef">Upstream condition NodeId, rendered by <see cref="AlarmIndexKey"/>.</param>
|
||||
/// <param name="EventId">EventId of the most recent event seen for that condition.</param>
|
||||
internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId);
|
||||
|
||||
/// <summary>#619 — the key both sides of the alarm index use: server-stable (<c>nsu=</c>) when a map is
|
||||
/// available, so a remote namespace-table reorder across an in-place reconnect cannot orphan it.</summary>
|
||||
/// <param name="map">The current session's namespace map, or null.</param>
|
||||
/// <param name="nodeId">The NodeId to key.</param>
|
||||
/// <returns>The index key.</returns>
|
||||
internal static string AlarmIndexKey(NamespaceMap? map, NodeId nodeId) =>
|
||||
map?.ToStableReference(nodeId) ?? nodeId.ToString() ?? string.Empty;
|
||||
|
||||
/// <summary>
|
||||
/// Map an OPC UA <c>BaseEventType.Severity</c> (1..1000) to our coarse-grained
|
||||
/// <see cref="AlarmSeverity"/> bucket. Thresholds match the OPC UA A&C Part 9
|
||||
@@ -2364,6 +2757,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
TransitionTo(HostState.Running);
|
||||
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
|
||||
_diag.Record(OpcUaClientDiagnostics.ReconnectSuccess);
|
||||
// #619: transitions during the outage were missed (a recreated subscription starts empty; a transferred
|
||||
// one may have overflowed). ConditionRefresh re-reports current states; unchanged ones dedup to nothing.
|
||||
if (!_alarmSubscriptions.IsEmpty)
|
||||
_ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user