/* ======================================================================== * 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 a data value queue for a data change monitoredItem /// public interface IDataChangeQueueHandler : IDisposable { /// /// Sets the queue size. /// /// The new queue size. /// Whether to discard the oldest values if the queue overflows. /// Specifies which diagnostics which should be kept in the queue. void SetQueueSize(uint queueSize, bool discardOldest, DiagnosticsMasks diagnosticsMasks); /// /// Set the sampling interval of the queue /// /// the sampling interval void SetSamplingInterval(double samplingInterval); /// /// Number of DataValues in the queue /// int ItemsInQueue { get; } /// /// Queues a value /// /// the dataValue /// the error /// true of overflow occured bool QueueValue(in DataValue value, ServiceResult error); /// /// Dequeues the last item /// /// true if an item was dequeued bool PublishSingleValue( out DataValue value, out ServiceResult error, bool noEventLog = false); } /// /// Mangages a data value queue for a data change monitoredItem /// public class DataChangeQueueHandler : IDataChangeQueueHandler { /// /// Creates a new Queue /// /// the id of the monitored item /// true if a durable queue shall be created /// the factory for /// The telemetry context to use to create obvservability instruments /// Handler for discarded values public DataChangeQueueHandler( uint monitoredItemId, bool createDurable, IMonitoredItemQueueFactory queueFactory, ITelemetryContext telemetry, Action? discardedValueHandler = null) : this(monitoredItemId, createDurable, queueFactory, telemetry, discardedValueHandler, null) { } /// /// Creates a new Queue with an explicit so /// the per-sample timestamp used for sampling-interval throttling can /// be mocked in tests. /// public DataChangeQueueHandler( uint monitoredItemId, bool createDurable, IMonitoredItemQueueFactory queueFactory, ITelemetryContext telemetry, Action? discardedValueHandler, TimeProvider? timeProvider) { m_logger = telemetry.CreateLogger(); m_dataValueQueue = queueFactory.CreateDataChangeQueue(createDurable, monitoredItemId); m_timeProvider = timeProvider ?? TimeProvider.System; m_discardedValueHandler = discardedValueHandler!; m_monitoredItemId = monitoredItemId; m_discardOldest = false; m_overflow = default; m_overflowPending = false; m_nextSampleTime = 0; m_samplingInterval = 0; } /// /// Create a DatachangeQueueHandler from an existing queue /// Used for restore after a server restart /// public DataChangeQueueHandler( IDataChangeMonitoredItemQueue dataValueQueue, bool discardOldest, double samplingInterval, ITelemetryContext telemetry, Action? discardedValueHandler = null) : this(dataValueQueue, discardOldest, samplingInterval, telemetry, discardedValueHandler, null) { } /// /// Create a DatachangeQueueHandler from an existing queue with an /// explicit . /// public DataChangeQueueHandler( IDataChangeMonitoredItemQueue dataValueQueue, bool discardOldest, double samplingInterval, ITelemetryContext telemetry, Action? discardedValueHandler, TimeProvider? timeProvider) { m_logger = telemetry.CreateLogger(); m_dataValueQueue = dataValueQueue; m_monitoredItemId = dataValueQueue.QueueSize; m_discardOldest = discardOldest; m_discardedValueHandler = discardedValueHandler!; m_nextSampleTime = 0; m_overflow = default; m_overflowPending = false; m_timeProvider = timeProvider ?? TimeProvider.System; SetSamplingInterval(samplingInterval); } /// /// Sets the queue size. /// /// The new queue size. /// Whether to discard the oldest values if the queue overflows. /// Specifies which diagnostics which should be kept in the queue. public void SetQueueSize( uint queueSize, bool discardOldest, DiagnosticsMasks diagnosticsMasks) { bool queueErrors = (diagnosticsMasks & DiagnosticsMasks.OperationAll) != 0; m_discardOldest = discardOldest; // copy existing values. List? existingValues = null; List? existingErrors = null; if (ItemsInQueue > 0) { existingValues = new List((int)queueSize); existingErrors = new List((int)queueSize); while (PublishSingleValue(out DataValue value, out ServiceResult error, true)) { existingValues.Add(value); existingErrors.Add(error); } } m_dataValueQueue.ResetQueue(queueSize, queueErrors); m_overflow = default; m_overflowPending = false; // requeue the data. if (existingValues != null) { for (int ii = 0; ii < existingValues.Count; ii++) { Enqueue(existingValues[ii], existingErrors![ii]); } } } /// /// Set the sampling interval of the queue /// /// the sampling interval public void SetSamplingInterval(double samplingInterval) { // substract the previous sampling interval. if (m_samplingInterval < m_nextSampleTime) { m_nextSampleTime -= m_samplingInterval; } // calculate the next sampling interval. m_samplingInterval = (long)samplingInterval; if (m_samplingInterval > 0) { m_nextSampleTime += m_samplingInterval; } else { m_nextSampleTime = 0; } } /// /// Number of DataValues in the queue /// public int ItemsInQueue => m_dataValueQueue.ItemsInQueue; /// /// Queues a value /// /// the dataValue /// the error /// true of overflow occured public bool QueueValue(in DataValue value, ServiceResult error) { long now = m_timeProvider.GetTimestampMilliseconds(); if (m_dataValueQueue.ItemsInQueue > 0) { // check if too soon for another sample. if (now < m_nextSampleTime) { if (m_logger.IsEnabled(LogLevel.Trace)) { m_dataValueQueue.TryPeekLastValue(out DataValue overwrittenValue); m_logger.LogTrace( "OVERWRITTEN VALUE (TOO SOON FOR ANOTHER SAMPLE): Value={Value} CODE={Code}<{Code:X8}> SamplingInterval={SamplingInterval}" + "QueueValueCall {Now} NextSampleTime {NextSampleTime}", overwrittenValue.WrappedValue, overwrittenValue.StatusCode.Code, value.StatusCode.Code, m_samplingInterval, now, m_nextSampleTime); } m_dataValueQueue.OverwriteLastValue(value, error); m_discardedValueHandler?.Invoke(); return false; } } // update next sample time. if (m_nextSampleTime > 0) { long delta = now - m_nextSampleTime; if (m_samplingInterval > 0 && delta >= 0) { m_nextSampleTime += ((delta / m_samplingInterval) + 1) * m_samplingInterval; } } else { m_nextSampleTime = now + m_samplingInterval; } // queue next value. return Enqueue(value, error); } /// /// Deques the last item /// public bool PublishSingleValue( out DataValue value, out ServiceResult error, bool noEventLog = false) { if (m_dataValueQueue.Dequeue(out value, out error)) { if (m_overflowPending && m_overflow == value) { SetOverflowBit(ref value, ref error); m_overflow = default; m_overflowPending = false; } if (!noEventLog && m_logger.IsEnabled(LogLevel.Trace)) { m_logger.LogTrace( "DEQUEUE VALUE: Value={Value} CODE={Code}<{Code:X8}> OVERFLOW={Overflow}", value.WrappedValue, value.StatusCode.Code, value.StatusCode.Code, value.StatusCode.Overflow); } return true; } return false; } /// /// Enque value /// /// true of overflow occured /// private bool Enqueue(DataValue value, ServiceResult error) { // check for empty queue. if (m_dataValueQueue.ItemsInQueue == 0) { if (m_logger.IsEnabled(LogLevel.Trace)) { m_logger.LogTrace("ENQUEUE VALUE: Value={Value}", value.WrappedValue); } m_dataValueQueue.Enqueue(value, error); return false; } // check if the latest value has initial dummy data if (m_dataValueQueue.TryPeekLastValue(out DataValue lastValue) && lastValue.StatusCode == StatusCodes.BadWaitingForInitialData) { // overwrite the last value m_dataValueQueue.OverwriteLastValue(value, error); return false; } // check if queue is full. if (m_dataValueQueue.ItemsInQueue == m_dataValueQueue.QueueSize) { m_discardedValueHandler?.Invoke(); if (!m_discardOldest) { if (m_dataValueQueue.TryPeekLastValue(out DataValue peekedLast)) { ServerUtils.ReportDiscardedValue( default, m_monitoredItemId, peekedLast); } //set overflow bit in newest value m_overflow = value; m_overflowPending = true; // overwrite last value m_dataValueQueue.OverwriteLastValue(value, error); return true; } // remove oldest value. if (m_dataValueQueue.Dequeue(out DataValue discardedValue, out _)) { ServerUtils.ReportDiscardedValue(default, m_monitoredItemId, discardedValue); } else { throw new ServiceResultException( StatusCodes.BadInternalError, "Error queueing DataValue. DataValueQueue was full but it was not possible to discard the oldest value."); } //set overflow bit in oldest value if (m_dataValueQueue.TryPeekOldestValue(out DataValue oldestValue)) { m_overflow = oldestValue; m_overflowPending = true; } m_dataValueQueue.Enqueue(value, error); return true; } else if (m_logger.IsEnabled(LogLevel.Trace)) { m_logger.LogTrace("ENQUEUE VALUE: Value={Value}", value.WrappedValue); } m_dataValueQueue.Enqueue(value, error); return false; } /// /// Sets the overflow bit in the value and error. /// /// The value to update. /// The error to update. private static void SetOverflowBit(ref DataValue value, ref ServiceResult error) { value = value.WithStatus(value.StatusCode.SetOverflow(true)); if (error != null) { // have to copy before updating because the ServiceResult is invariant. error = new ServiceResult( error.NamespaceUri, error.StatusCode.SetOverflow(true), error.LocalizedText, error.AdditionalInfo, error.InnerResult); } } /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// Overridable method to dispose of resources. /// protected virtual void Dispose(bool disposing) { if (disposing) { m_dataValueQueue?.Dispose(); } } private readonly IDataChangeMonitoredItemQueue m_dataValueQueue; private readonly ILogger m_logger; private readonly TimeProvider m_timeProvider; private readonly uint m_monitoredItemId; private bool m_discardOldest; private long m_nextSampleTime; private long m_samplingInterval; private readonly Action m_discardedValueHandler; private DataValue m_overflow; private bool m_overflowPending; } }