/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Diagnostics; using System.Threading; using System.Threading.Tasks; namespace Opc.Ua.Server { /// /// An object that manages all events raised within the server. /// public class EventManager : IDisposable { /// /// Creates a new instance of a sampling group. /// public EventManager(IServerInternal server, uint maxQueueSize, uint maxDurableQueueSize) { m_server = server ?? throw new ArgumentNullException(nameof(server)); m_monitoredItems = []; m_maxEventQueueSize = maxQueueSize; m_maxDurableEventQueueSize = maxDurableQueueSize; } /// /// Frees any unmanaged resources. /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// An overrideable version of the Dispose. /// protected virtual void Dispose(bool disposing) { if (disposing) { List? monitoredItems; lock (m_lock) { monitoredItems = [.. m_monitoredItems.Values]; m_monitoredItems.Clear(); } foreach (IEventMonitoredItem monitoredItem in monitoredItems) { monitoredItem?.Dispose(); } } } /// /// Reports an event to the supplied monitored items WITHOUT /// validating . Use /// instead so per-item permission /// checks on the event's EventTypeId and SourceNode /// are honored (Part 3 §8.55). /// /// is null. [Obsolete("Use ReportEventAsync so PermissionType.ReceiveEvents is validated on EventTypeId and SourceNode for each monitored item.")] public static void ReportEvent(IFilterTarget e, IList monitoredItems) { if (e == null) { throw new ArgumentNullException(nameof(e)); } foreach (IEventMonitoredItem monitoredItem in monitoredItems) { monitoredItem.QueueEvent(e); } } /// /// Reports an event to the supplied monitored items, validating /// on both the /// EventTypeId and SourceNode of the event for each /// item before queueing (per Part 3 §8.55). /// /// The event payload. /// /// The node manager that owns the source. Used to evaluate role /// permissions for the event's EventTypeId and /// SourceNode. /// /// The candidate receivers. /// The cancellation token. /// /// , , or /// is null. /// public static async ValueTask ReportEventAsync( IFilterTarget e, IAsyncNodeManager nodeManager, IList monitoredItems, CancellationToken cancellationToken = default) { if (e == null) { throw new ArgumentNullException(nameof(e)); } if (nodeManager == null) { throw new ArgumentNullException(nameof(nodeManager)); } if (monitoredItems == null) { throw new ArgumentNullException(nameof(monitoredItems)); } foreach (IEventMonitoredItem monitoredItem in monitoredItems) { if (monitoredItem == null) { continue; } ServiceResult result = await nodeManager .ValidateEventRolePermissionsAsync(monitoredItem, e, cancellationToken) .ConfigureAwait(false); if (ServiceResult.IsBad(result)) { continue; } monitoredItem.QueueEvent(e); } } /// /// Creates a set of monitored items. /// public IEventMonitoredItem CreateMonitoredItem( OperationContext context, IAsyncNodeManager nodeManager, object handle, uint subscriptionId, MonitoredItemIdFactory monitoredItemIdFactory, TimestampsToReturn timestampsToReturn, double publishingInterval, MonitoredItemCreateRequest itemToCreate, EventFilter filter, bool createDurable) { lock (m_lock) { // calculate sampling interval. double samplingInterval = itemToCreate.RequestedParameters.SamplingInterval; if (samplingInterval < 0) { samplingInterval = publishingInterval; } // limit the queue size. uint revisedQueueSize = CalculateRevisedQueueSize( createDurable, itemToCreate.RequestedParameters.QueueSize); // Allocate the monitored item id uint monitoredItemId; do { monitoredItemId = monitoredItemIdFactory.GetNextId(); } while (!m_monitoredItems.TryAdd(monitoredItemId, null!)); // create the monitored item. IEventMonitoredItem monitoredItem = new MonitoredItem( m_server, nodeManager, handle, subscriptionId, monitoredItemId, itemToCreate.ItemToMonitor, context.DiagnosticsMask, timestampsToReturn, itemToCreate.MonitoringMode, itemToCreate.RequestedParameters.ClientHandle, filter, filter, null, samplingInterval, revisedQueueSize, itemToCreate.RequestedParameters.DiscardOldest, MinimumSamplingIntervals.Continuous, createDurable); // now save the monitored item. Debug.Assert(m_monitoredItems[monitoredItemId] == null); m_monitoredItems[monitoredItemId] = monitoredItem; return monitoredItem; } } /// /// Restore a MonitoredItem after a restart /// public IEventMonitoredItem RestoreMonitoredItem( IAsyncNodeManager nodeManager, object handle, IStoredMonitoredItem storedMonitoredItem) { lock (m_lock) { // limit the queue size. storedMonitoredItem.QueueSize = CalculateRevisedQueueSize( storedMonitoredItem.IsDurable, storedMonitoredItem.QueueSize); // create the monitored item. var monitoredItem = new MonitoredItem( m_server, nodeManager, handle, storedMonitoredItem); // save the monitored item. m_monitoredItems.Add(monitoredItem.Id, monitoredItem); return monitoredItem; } } /// /// calculates a revised queue size based on the application confiugration limits /// private uint CalculateRevisedQueueSize(bool isDurable, uint queueSize) { if (queueSize > m_maxEventQueueSize && !isDurable) { queueSize = m_maxEventQueueSize; } if (queueSize > m_maxDurableEventQueueSize && isDurable) { queueSize = m_maxDurableEventQueueSize; } return queueSize; } /// /// Modifies a monitored item. /// public void ModifyMonitoredItem( OperationContext context, IEventMonitoredItem monitoredItem, TimestampsToReturn timestampsToReturn, MonitoredItemModifyRequest itemToModify, EventFilter filter) { lock (m_lock) { // should never be called with items that it does not own. if (!m_monitoredItems.ContainsKey(monitoredItem.Id)) { return; } // limit the queue size. uint revisedQueueSize = CalculateRevisedQueueSize( monitoredItem.IsDurable, itemToModify.RequestedParameters.QueueSize); // modify the attributes. monitoredItem.ModifyAttributes( context.DiagnosticsMask, timestampsToReturn, itemToModify.RequestedParameters.ClientHandle, filter, filter, null!, itemToModify.RequestedParameters.SamplingInterval, revisedQueueSize, itemToModify.RequestedParameters.DiscardOldest); } } /// /// Deletes a monitored item. /// public void DeleteMonitoredItem(uint monitoredItemId) { lock (m_lock) { m_monitoredItems.Remove(monitoredItemId); } } /// /// Returns the currently active monitored items. /// public IList GetMonitoredItems() { lock (m_lock) { return [.. m_monitoredItems.Values]; } } private readonly Lock m_lock = new(); private readonly IServerInternal m_server; private readonly Dictionary m_monitoredItems; private readonly uint m_maxEventQueueSize; private readonly uint m_maxDurableEventQueueSize; } }