MonitoredItemNotification.cs

namespace Opc.Ua.Cloud.Publisher
{
    using Microsoft.Extensions.Logging;
    using Opc.Ua;
    using Opc.Ua.Client;
    using Opc.Ua.Client.ComplexTypes;
    using Opc.Ua.Cloud.Publisher.Interfaces;
    using Opc.Ua.Cloud.Publisher.Models;
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Runtime.CompilerServices;
    using System.Threading.Tasks;

    public class MonitoredItemNotification : IMessageSource
    {
        private readonly ILogger _logger;
        private readonly ConditionalWeakTable<ISession, ComplexTypeSystem> _complexTypeSystems = new();

        // caches the resolved browse path per node so that not every notification triggers a browse session
        private readonly ConcurrentDictionary<string, string> _browsePathCache = new();

        // tracks which complex data types have already been (attempted to be) loaded per session so we don't
        // re-issue a LoadType request on every single notification (which floods BadRequestTimeout errors)
        private readonly ConcurrentDictionary<string, bool> _loadedComplexTypes = new();

        public ConcurrentDictionary<string, bool> SkipFirst { get; set; } = new ConcurrentDictionary<string, bool>();

        private NodeId[] KnownEventTypes = new NodeId[]
{
            ObjectTypeIds.BaseEventType,
            ObjectTypeIds.ConditionType,
            ObjectTypeIds.DialogConditionType,
            ObjectTypeIds.AlarmConditionType,
            ObjectTypeIds.ExclusiveLimitAlarmType,
            ObjectTypeIds.NonExclusiveLimitAlarmType,
            ObjectTypeIds.AuditEventType,
            ObjectTypeIds.AuditUpdateMethodEventType
        };

        public MonitoredItemNotification(ILoggerFactory loggerFactory)
        {
            _logger = loggerFactory.CreateLogger("MonitoredItemNotification");
        }

        public async void EventNotificationHandlerAsync(MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e)
        {
            try
            {
                EventFieldList notification = e.NotificationValue as EventFieldList;
                if (notification == null)
                {
                    return;
                }

                NodeId eventTypeId = FindEventType(monitoredItem, notification);
                if (NodeId.IsNull(eventTypeId))
                {
                    return;
                }

                if (eventTypeId == ObjectTypeIds.RefreshStartEventType)
                {
                    return;
                }

                if (eventTypeId == ObjectTypeIds.RefreshEndEventType)
                {
                    return;
                }

                ConditionState condition = await ConstructEventAsync(
                    (Session)monitoredItem.Subscription.Session,
                    monitoredItem,
                    notification,
                    new Dictionary<NodeId, NodeId>()).ConfigureAwait(false) as ConditionState;
                if (condition == null)
                {
                    return;
                }

                MessageProcessorModel messageData = new()
                {
                    ExpandedNodeId = NodeId.ToExpandedNodeId(monitoredItem.ResolvedNodeId, monitoredItem.Subscription.Session.NamespaceUris).ToString(),
                    ApplicationUri = monitoredItem.Subscription.Session.Endpoint.Server.ApplicationUri,
                    MessageContext = (ServiceMessageContext)monitoredItem.Subscription.Session.MessageContext
                };

                if (Settings.Instance.AddBrowsePathToName)
                {
                    messageData.Name = await GetNameWithBrowsePathAsync(monitoredItem.Subscription.Session, monitoredItem.ResolvedNodeId, monitoredItem.DisplayName).ConfigureAwait(false);
                }

                // Source
                if (condition.SourceName != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Source",
                        Value = new DataValue(condition.SourceName.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Condition
                if (condition.ConditionName != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Condition",
                        Value = new DataValue(condition.ConditionName.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Branch
                if (condition.BranchId != null && !NodeId.IsNull(condition.BranchId.Value))
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Branch",
                        Value = new DataValue(condition.BranchId.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Type
                INode type = await monitoredItem.Subscription.Session.NodeCache.FindAsync(condition.TypeDefinitionId).ConfigureAwait(false);
                if (type != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Type",
                        Value = new DataValue(type.ToString() + " (" + NodeId.ToExpandedNodeId(new NodeId(type.NodeId.ToString()), monitoredItem.Subscription.Session.NamespaceUris).ToString() + ")")
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Severity
                if (condition.Severity != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Severity",
                        Value = new DataValue(((EventSeverity)condition.Severity.Value).ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Time
                if (condition.Time != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Time",
                        Value = new DataValue(condition.Time.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // State
                if (condition.EnabledState != null && condition.EnabledState.EffectiveDisplayName != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "State",
                        Value = new DataValue(condition.EnabledState.EffectiveDisplayName.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Message
                if (condition.Message != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Message",
                        Value = new DataValue(condition.Message.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                // Comment
                if (condition.Comment != null)
                {
                    EventValueModel eventValue = new EventValueModel()
                    {
                        Name = "Comment",
                        Value = new DataValue(condition.Comment.Value.ToString())
                    };

                    messageData.EventValues.Add(eventValue);
                }

                if (messageData.EventValues.Count > 0)
                {
                    MessageProcessor.Enqueue(messageData);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing monitored item notification");
            }
        }

        private async Task<string> GetNameWithBrowsePathAsync(ISession session, NodeId nodeId, string displayName)
        {
            if (session == null || nodeId == null || NodeId.IsNull(nodeId))
            {
                return displayName;
            }

            // nodeIds are not unique across OPC UA servers, so combine the session id with the expanded node id
            string cacheKey = session.SessionId + "|" + NodeId.ToExpandedNodeId(nodeId, session.NamespaceUris)?.ToString();

            // only browse the first time we see a node; every subsequent notification uses the cached path
            if (_browsePathCache.TryGetValue(cacheKey, out string cachedPath))
            {
                return cachedPath;
            }

            try
            {
                List<string> segments = new List<string>();
                NodeId currentNodeId = nodeId;

                // walk up the hierarchy towards the Objects/Root folder, collecting browse names
                while (currentNodeId != null
                    && !NodeId.IsNull(currentNodeId)
                    && currentNodeId != ObjectIds.ObjectsFolder
                    && currentNodeId != ObjectIds.RootFolder)
                {
                    INode node = await session.NodeCache.FindAsync(currentNodeId).ConfigureAwait(false);
                    if (node == null)
                    {
                        break;
                    }

                    string segment = (node.BrowseName != null && !string.IsNullOrEmpty(node.BrowseName.Name))
                        ? node.BrowseName.Name
                        : node.DisplayName?.Text;
                    segments.Insert(0, segment);

                    // browse the parent via inverse hierarchical references
                    (_, _, ReferenceDescriptionCollection parents) = await session.BrowseAsync(
                        null,
                        null,
                        currentNodeId,
                        1,
                        BrowseDirection.Inverse,
                        ReferenceTypeIds.HierarchicalReferences,
                        true,
                        0).ConfigureAwait(false);

                    if (parents == null || parents.Count == 0 || parents[0].NodeId == null || parents[0].NodeId.IsAbsolute)
                    {
                        break;
                    }

                    currentNodeId = (NodeId)parents[0].NodeId;
                }

                string browsePath = (segments.Count > 0) ? string.Join("/", segments) : displayName;
                _browsePathCache[cacheKey] = browsePath;
                return browsePath;
            }
            catch (Exception ex)
            {
                _logger.LogWarning("Could not build browse path for node {nodeId}; falling back to display name. {message}", nodeId, ex.Message);

                // cache the fallback so we don't repeatedly browse a node that fails
                _browsePathCache[cacheKey] = displayName;
                return displayName;
            }
        }

        private NodeId FindEventType(MonitoredItem monitoredItem, EventFieldList notification)
        {
            EventFilter filter = monitoredItem.Status.Filter as EventFilter;

            if (filter != null)
            {
                for (int i = 0; i < filter.SelectClauses.Count; i++)
                {
                    SimpleAttributeOperand clause = filter.SelectClauses[i];

                    if (clause.BrowsePath.Count == 1 && clause.BrowsePath[0] == BrowseNames.EventType)
                    {
                        return notification.EventFields[i].Value as NodeId;
                    }
                }
            }

            return null;
        }

        private async Task<BaseEventState> ConstructEventAsync(
            Session session,
            MonitoredItem monitoredItem,
            EventFieldList notification,
            Dictionary<NodeId, NodeId> eventTypeMappings)
        {
            // find the event type.
            NodeId eventTypeId = FindEventType(monitoredItem, notification);

            if (eventTypeId == null)
            {
                return null;
            }

            // look up the known event type.
            NodeId knownTypeId = null;

            if (!eventTypeMappings.TryGetValue(eventTypeId, out knownTypeId))
            {
                // check for a known type
                for (int j = 0; j < KnownEventTypes.Length; j++)
                {
                    if (KnownEventTypes[j] == eventTypeId)
                    {
                        knownTypeId = eventTypeId;
                        eventTypeMappings.Add(eventTypeId, eventTypeId);
                        break;
                    }
                }

                // browse for the supertypes of the event type.
                if (knownTypeId == null)
                {
                    ReferenceDescriptionCollection supertypes = await UAClient.BrowseSuperTypesAsync(session, eventTypeId, false).ConfigureAwait(false);

                    // can't do anything with unknown types.
                    if (supertypes == null)
                    {
                        return null;
                    }

                    // find the first supertype that matches a known event type.
                    for (int i = 0; i < supertypes.Count; i++)
                    {
                        for (int j = 0; j < KnownEventTypes.Length; j++)
                        {
                            if (KnownEventTypes[j] == supertypes[i].NodeId)
                            {
                                knownTypeId = KnownEventTypes[j];
                                eventTypeMappings.Add(eventTypeId, knownTypeId);
                                break;
                            }
                        }

                        if (knownTypeId != null)
                        {
                            break;
                        }
                    }
                }
            }

            if (knownTypeId == null)
            {
                return null;
            }

            // all of the known event types have a UInt32 as identifier.
            uint? id = knownTypeId.Identifier as uint?;

            if (id == null)
            {
                return null;
            }

            // construct the event based on the known event type.
            BaseEventState e = null;

            switch (id.Value)
            {
                case ObjectTypes.ConditionType: { e = new ConditionState(null); break; }
                case ObjectTypes.DialogConditionType: { e = new DialogConditionState(null); break; }
                case ObjectTypes.AlarmConditionType: { e = new AlarmConditionState(null); break; }
                case ObjectTypes.ExclusiveLimitAlarmType: { e = new ExclusiveLimitAlarmState(null); break; }
                case ObjectTypes.NonExclusiveLimitAlarmType: { e = new NonExclusiveLimitAlarmState(null); break; }
                case ObjectTypes.AuditEventType: { e = new AuditEventState(null); break; }
                case ObjectTypes.AuditUpdateMethodEventType: { e = new AuditUpdateMethodEventState(null); break; }
                default: { e = new BaseEventState(null); break; }
            }

            // get the filter which defines the contents of the notification.
            EventFilter filter = monitoredItem.Status.Filter as EventFilter;

            // initialize the event with the values in the notification.
            e.Update(session.SystemContext, filter.SelectClauses, notification);

            // save the original notification.
            e.Handle = notification;

            return e;
        }

        public async void DataChangedNotificationHandlerAsync(MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e)
        {
            try
            {
                if (e == null || e.NotificationValue == null || monitoredItem == null || monitoredItem.Subscription == null || monitoredItem.Subscription.Session == null)
                {
                    return;
                }

                if (!(e.NotificationValue is Opc.Ua.MonitoredItemNotification notification))
                {
                    return;
                }

                if (!(notification.Value is DataValue value))
                {
                    return;
                }

                string dataType = string.Empty;
                VariableNode variable = (VariableNode)await monitoredItem.Subscription.Session.NodeCache.FindAsync(monitoredItem.StartNodeId).ConfigureAwait(false);
                if (variable != null)
                {
                    ISession session = monitoredItem.Subscription.Session;
                    ExpandedNodeId nodeTypeId = variable.DataType;

                    // only custom/complex types (defined in a server-specific namespace) need to be loaded via the
                    // complex type system; built-in base types (namespace 0, e.g. i=11 Double) never do, and attempting
                    // to load them just times out. Also only attempt each type once per session.
                    if (nodeTypeId != null && nodeTypeId.NamespaceIndex != 0)
                    {
                        string typeCacheKey = session.SessionId + "|" + nodeTypeId.ToString();
                        if (_loadedComplexTypes.TryAdd(typeCacheKey, true))
                        {
                            try
                            {
                                // handle complex types using a per-session cached type system
                                ComplexTypeSystem complexTypeSystem = _complexTypeSystems.GetValue(session, s => new ComplexTypeSystem(s));
                                await complexTypeSystem.LoadTypeAsync(nodeTypeId).ConfigureAwait(false);
                            }
                            catch (Exception ex)
                            {
                                _logger.LogWarning("Could not load complex type {typeId}: {message}", nodeTypeId, ex.Message);
                            }
                        }
                    }

                    dataType = NodeId.ToExpandedNodeId(variable.DataType, monitoredItem.Subscription?.Session?.NamespaceUris).ToString();
                }

                if (monitoredItem.Subscription != null)
                {
                    // capture the session locally; it can be torn down concurrently while notifications are still in flight
                    ISession session = monitoredItem.Subscription.Session;
                    if (session == null || session.Endpoint == null || session.Endpoint.Server == null)
                    {
                        return;
                    }

                    string name = monitoredItem.DisplayName;
                    if (Settings.Instance.AddBrowsePathToName)
                    {
                        name = await GetNameWithBrowsePathAsync(session, monitoredItem.ResolvedNodeId, monitoredItem.DisplayName).ConfigureAwait(false);
                    }

                    MessageProcessorModel messageData = new MessageProcessorModel
                    {
                        ExpandedNodeId = NodeId.ToExpandedNodeId(monitoredItem.ResolvedNodeId, session.NamespaceUris).ToString(),
                        ApplicationUri = session.Endpoint.Server.ApplicationUri,
                        MessageContext = (ServiceMessageContext)session.MessageContext,
                        Name = name,
                        Value = value,
                        DataType = dataType
                    };

                    // skip event if needed
                    if (SkipFirst.TryGetValue(messageData.ExpandedNodeId, out bool skip) && skip)
                    {
                        _logger.LogInformation($"Skipping first telemetry event for node '{messageData.ExpandedNodeId}'.");
                        SkipFirst[messageData.ExpandedNodeId] = false;
                    }
                    else
                    {
                        MessageProcessor.Enqueue(messageData);
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing monitored item notification");
            }
        }
    }
}