using System; using System.Collections.Generic; using ZB.MOM.WW.MxGateway.Contracts.Proto; namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; /// /// In-process dispatcher that owns the lifetime of an /// + /// pair, and wires the consumer's AlarmTransitionEmitted stream /// onto the sink's EnqueueTransition path so transitions land on /// the worker's as proto /// messages ready for IPC dispatch. /// /// /// /// The dispatcher carries the consumer→sink→queue pipeline. The /// worker's IPC layer issues SubscribeAlarmsCommand / /// AcknowledgeAlarmCommand / QueryActiveAlarmsCommand /// through , which owns one /// dispatcher per session. Acknowledgement resolves the alarm to its /// MXAccess ack item through the consumer and hands the write to the /// , which authenticates as the operator. /// /// /// Threading: owns no internal /// timer — the worker's STA drives polling via /// StaRuntime.InvokeAsync(() => PollOnce()), so the /// consumer's AlarmTransitionEmitted event fires on the STA. /// The dispatcher is purely a pass-through, so it inherits that /// thread. Fan-out into EnqueueTransition uses the /// thread-safe . /// /// public sealed class AlarmDispatcher : IDisposable { private readonly IMxAccessAlarmConsumer consumer; private readonly MxAccessAlarmEventSink sink; private readonly string sessionId; private readonly EventHandler handler; private readonly IAlarmAckWriter ackWriter; private bool disposed; /// Initializes a new alarm dispatcher for the given consumer, sink, session ID and ack writer. /// The alarm consumer. /// The alarm event sink. /// The session identifier. /// /// Writes the resolved ack item as the authenticated operator. Owned by /// the caller (), which disposes it; the /// dispatcher never does, so the writer's registered MXAccess server /// outlives an unsubscribe/resubscribe cycle. /// public AlarmDispatcher( IMxAccessAlarmConsumer consumer, MxAccessAlarmEventSink sink, string sessionId, IAlarmAckWriter ackWriter) { this.consumer = consumer ?? throw new ArgumentNullException(nameof(consumer)); this.sink = sink ?? throw new ArgumentNullException(nameof(sink)); this.sessionId = sessionId ?? string.Empty; this.ackWriter = ackWriter ?? throw new ArgumentNullException(nameof(ackWriter)); // Sink.Attach is the seam that propagates the session id onto the // proto SessionId field of every emitted MxEvent. Pass the consumer // as the "associated COM object" — sink ignores the object reference // for the alarm path, but the existing IMxAccessEventSink contract // requires a non-null first arg. this.sink.Attach(this.consumer, this.sessionId); this.handler = OnTransition; consumer.AlarmTransitionEmitted += handler; } /// /// Begin polling the configured AVEVA alarm provider for /// transitions. The supplied subscription expression follows the /// canonical \\<machine>\Galaxy!<area> format. /// /// The subscription expression (e.g., \\HOST\Galaxy!Area). public void Subscribe(string subscription) { if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher)); consumer.Subscribe(subscription); } /// Acknowledges the alarm with the given GUID by writing its ack item as the operator. /// The alarm GUID. /// The acknowledgement comment. /// The operator to authenticate as. /// The outcome. public AlarmAckResult Acknowledge(Guid alarmGuid, string ackComment, AlarmAckOperator ackOperator) { if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher)); AlarmAckTarget target = consumer.ResolveAckTarget(alarmGuid); return target.Found ? ackWriter.Acknowledge(target.ItemAddress, ackComment, ackOperator) : AlarmAckResult.FromTarget(target); } /// Acknowledges the alarm identified by name/provider/group as the operator. /// The tag/alarm name. /// The provider. /// The group/area. /// The acknowledgement comment. /// The operator to authenticate as. /// The outcome. public AlarmAckResult AcknowledgeByName( string alarmName, string providerName, string groupName, string ackComment, AlarmAckOperator ackOperator) { if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher)); AlarmAckTarget target = consumer.ResolveAckTargetByName(alarmName, providerName, groupName); return target.Found ? ackWriter.Acknowledge(target.ItemAddress, ackComment, ackOperator) : AlarmAckResult.FromTarget(target); } /// /// Drives a single synchronous poll of the underlying consumer. /// Must be called on the STA thread that owns the wnwrap COM object. /// No-op if the dispatcher has been disposed. /// public void PollOnce() { if (disposed) return; consumer.PollOnce(); } /// /// Snapshot the currently-active alarm set as /// protos for the /// QueryActiveAlarms RPC's ConditionRefresh stream. /// /// /// Receives whether the fetch behind the snapshot hit the per-fetch cap, /// so the returned set may omit active alarms. Forwarded from the single /// atomic consumer read, and stamped onto every returned record as /// ActiveAlarmSnapshot.FromTruncatedSnapshot. Also returned /// separately because a snapshot of zero records still has to report it. /// /// The currently active alarm snapshots. public IReadOnlyList SnapshotActiveAlarms(out bool truncated) { if (disposed) throw new ObjectDisposedException(nameof(AlarmDispatcher)); // One consumer call yields the records and the verdict together, under a // single acquisition of the consumer's snapshot lock. Reading them // separately would let a poll interleave and pair a stale not-truncated // verdict with a capped snapshot — a set that reads complete while // missing actives. The atomicity is structural here, not a consequence of // the STA happening to serialize the two calls. IReadOnlyList records = consumer.SnapshotActiveAlarms(out truncated); if (records.Count == 0) return Array.Empty(); List snapshots = new List(records.Count); foreach (MxAlarmSnapshotRecord record in records) { snapshots.Add(MapToSnapshot(record, truncated)); } return snapshots; } private void OnTransition(object? sender, MxAlarmTransitionEvent transition) { if (disposed) return; if (transition is null) return; MxAlarmSnapshotRecord record = transition.Record; AlarmTransitionKind kind = AlarmRecordTransitionMapper.MapTransition( transition.PreviousState, record.State); if (kind == AlarmTransitionKind.Unspecified) return; string fullReference = AlarmRecordTransitionMapper.ComposeFullReference( record.ProviderName, record.Group, record.TagName); bool useAck = kind == AlarmTransitionKind.Acknowledge && HasAckAttribution(record); sink.EnqueueTransition( alarmFullReference: fullReference, sourceObjectReference: record.TagName, alarmTypeName: record.Type, transitionKind: kind, severity: record.Priority, originalRaiseTimestampUtc: null, transitionTimestampUtc: record.TransitionTimestampUtc, operatorUser: useAck ? record.AckOperatorName : record.OperatorName, operatorComment: useAck ? record.AckComment : record.AlarmComment, category: record.Group, description: string.Empty, degraded: record.Degraded); } private static bool HasAckAttribution(MxAlarmSnapshotRecord record) => !string.IsNullOrEmpty(record.AckOperatorName) || !string.IsNullOrEmpty(record.AckComment); private static bool IsAcked(MxAlarmStateKind state) => state == MxAlarmStateKind.AckAlm || state == MxAlarmStateKind.AckRtn; private static ActiveAlarmSnapshot MapToSnapshot(MxAlarmSnapshotRecord record, bool truncated) { bool useAck = IsAcked(record.State) && HasAckAttribution(record); ActiveAlarmSnapshot snapshot = new ActiveAlarmSnapshot { AlarmFullReference = AlarmRecordTransitionMapper.ComposeFullReference( record.ProviderName, record.Group, record.TagName), SourceObjectReference = record.TagName, AlarmTypeName = record.Type, CurrentState = MapConditionState(record.State), Severity = record.Priority, OperatorUser = useAck ? record.AckOperatorName : record.OperatorName, OperatorComment = useAck ? record.AckComment : record.AlarmComment, Category = record.Group, Description = string.Empty, Degraded = record.Degraded, SourceProvider = record.Degraded ? AlarmProviderMode.Subtag : AlarmProviderMode.Alarmmgr, // Set-level status, stamped identically on every record of the // snapshot: QueryActiveAlarms streams bare ActiveAlarmSnapshot // messages with no envelope to hang it off. Independent of // Degraded above — that is about this record's provider, this is // about whether the set it belongs to is complete. FromTruncatedSnapshot = truncated, }; if (record.TransitionTimestampUtc != DateTime.MinValue) { snapshot.LastTransitionTimestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime( DateTime.SpecifyKind(record.TransitionTimestampUtc, DateTimeKind.Utc)); } return snapshot; } private static AlarmConditionState MapConditionState(MxAlarmStateKind state) { // The proto's AlarmConditionState only distinguishes Active / // ActiveAcked / Inactive — both Rtn states collapse to Inactive // (the ack-vs-unack distinction on a cleared alarm is not exposed // through OPC UA's Part 9 condition state model anyway). return state switch { MxAlarmStateKind.UnackAlm => AlarmConditionState.Active, MxAlarmStateKind.AckAlm => AlarmConditionState.ActiveAcked, MxAlarmStateKind.UnackRtn => AlarmConditionState.Inactive, MxAlarmStateKind.AckRtn => AlarmConditionState.Inactive, _ => AlarmConditionState.Unspecified, }; } /// Gets the session ID. public string SessionId => sessionId; /// public void Dispose() { if (disposed) return; disposed = true; try { consumer.AlarmTransitionEmitted -= handler; } catch { /* swallow */ } try { sink.Detach(); } catch { /* swallow */ } try { consumer.Dispose(); } catch { /* swallow */ } } }