/* ======================================================================== * 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 Microsoft.Extensions.Logging; namespace Opc.Ua.Server { /// /// Mangages an event queue for usage by a MonitoredItem /// public interface IEventQueueHandler : IDisposable { /// /// Sets the queue size. /// /// The new queue size. /// Whether to discard the oldest values if the queue overflows. void SetQueueSize(uint queueSize, bool discardOldest); /// /// The number of Items in the queue /// int ItemsInQueue { get; } /// /// True if the queue is overflowing /// bool Overflow { get; } /// /// Checks the last 1k queue entries if the event is already in there /// bool IsEventContainedInQueue(IFilterTarget instance); /// /// true if queue is already full and discarding is not allowed /// bool SetQueueOverflowIfFull(); /// /// Adds an event to the queue. /// void QueueEvent(EventFieldList fields); /// /// Publish Events /// /// Context for the operation /// Notifications to publish /// the maximum number of notifications to enqueue per call /// the number of events that were added to the notification queue uint Publish( OperationContext context, Queue notifications, uint maxNotificationsPerPublish); } /// /// Mangages an event queue for usage by a MonitoredItem /// public class EventQueueHandler : IEventQueueHandler { /// /// Creates a new Queue handler /// /// create a durable queue /// the factory for creating the factory for /// the id of the monitoredItem associated with the queue /// The telemetry context to use to create obvservability instruments public EventQueueHandler( bool createDurable, IMonitoredItemQueueFactory queueFactory, uint monitoredItemId, ITelemetryContext telemetry) { m_logger = telemetry.CreateLogger(); m_eventQueue = queueFactory.CreateEventQueue(createDurable, monitoredItemId); m_discardOldest = false; Overflow = false; } /// /// Create an EventQueueHandler from an existing queue /// Used for restore after a server restart /// public EventQueueHandler( IEventMonitoredItemQueue eventQueue, bool discardOldest, ITelemetryContext telemetry) { m_logger = telemetry.CreateLogger(); m_eventQueue = eventQueue; m_discardOldest = discardOldest; Overflow = false; } /// /// Sets the queue size. /// /// The new queue size. /// Whether to discard the oldest values if the queue overflows. public void SetQueueSize(uint queueSize, bool discardOldest) { m_discardOldest = discardOldest; m_eventQueue.SetQueueSize(queueSize, discardOldest); } /// /// The number of Items in the queue /// public int ItemsInQueue => m_eventQueue.ItemsInQueue; /// /// True if the queue is overflowing /// public bool Overflow { get; private set; } /// /// Checks the last 1k queue entries if the event is already in there /// public bool IsEventContainedInQueue(IFilterTarget instance) { return m_eventQueue.IsEventContainedInQueue(instance); } /// /// true if queue is already full and discarding is not allowed /// public bool SetQueueOverflowIfFull() { if (m_eventQueue.ItemsInQueue >= m_eventQueue.QueueSize && !m_discardOldest) { Overflow = true; return true; } return false; } /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// An overrideable version of the Dispose. /// protected virtual void Dispose(bool disposing) { if (disposing) { m_eventQueue?.Dispose(); } } /// /// Adds an event to the queue. /// /// public virtual void QueueEvent(EventFieldList fields) { // make space in the queue. if (m_eventQueue.ItemsInQueue >= m_eventQueue.QueueSize) { Overflow = true; if (!m_discardOldest) { throw new InvalidOperationException( "Queue is full and no discarding of old values is allowed"); } m_eventQueue.Dequeue(out _); } // queue the event. m_eventQueue.Enqueue(fields); } /// /// Publish Events /// /// System context /// Notifications /// the maximum number of notifications to enqueue per call public uint Publish( OperationContext context, Queue notifications, uint maxNotificationsPerPublish) { uint notificationCount = 0; while (notificationCount < maxNotificationsPerPublish && m_eventQueue.Dequeue(out EventFieldList fields)) { foreach (Variant field in fields.EventFields) { StatusResult tmpStatus; bool gotStatus = field.TryGetStructure(out tmpStatus!); StatusResult? statusResult = gotStatus ? tmpStatus : null; if (statusResult is { } status) { status.ApplyDiagnosticMasks( context.DiagnosticsMask, context.StringTable, m_logger); } } notifications.Enqueue(fields); notificationCount++; } // if overflow event is placed at the end of the queue only set overflow to false if the overflow event // still fits into the publish Overflow = Overflow && notificationCount == maxNotificationsPerPublish && !m_discardOldest; return notificationCount; } private bool m_discardOldest; private readonly IEventMonitoredItemQueue m_eventQueue; private readonly ILogger m_logger; } }