UA-.NETStandard/Libraries/Opc.Ua.Server/Subscription/MonitoredItem/QueueHandler/DataChangeQueueHandler.cs

/* ========================================================================
 * 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
{
    /// <summary>
    /// Mangages a data value queue for a data change monitoredItem
    /// </summary>
    public interface IDataChangeQueueHandler : IDisposable
    {
        /// <summary>
        /// Sets the queue size.
        /// </summary>
        /// <param name="queueSize">The new queue size.</param>
        /// <param name="discardOldest">Whether to discard the oldest values if the queue overflows.</param>
        /// <param name="diagnosticsMasks">Specifies which diagnostics which should be kept in the queue.</param>
        void SetQueueSize(uint queueSize, bool discardOldest, DiagnosticsMasks diagnosticsMasks);

        /// <summary>
        /// Set the sampling interval of the queue
        /// </summary>
        /// <param name="samplingInterval">the sampling interval</param>
        void SetSamplingInterval(double samplingInterval);

        /// <summary>
        /// Number of DataValues in the queue
        /// </summary>
        int ItemsInQueue { get; }

        /// <summary>
        /// Queues a value
        /// </summary>
        /// <param name="value">the dataValue</param>
        /// <param name="error">the error</param>
        /// <returns>true of overflow occured</returns>
        bool QueueValue(in DataValue value, ServiceResult error);

        /// <summary>
        /// Dequeues the last item
        /// </summary>
        /// <returns>true if an item was dequeued</returns>
        bool PublishSingleValue(
            out DataValue value,
            out ServiceResult error,
            bool noEventLog = false);
    }

    /// <summary>
    /// Mangages a data value queue for a data change monitoredItem
    /// </summary>
    public class DataChangeQueueHandler : IDataChangeQueueHandler
    {
        /// <summary>
        /// Creates a new Queue
        /// </summary>
        /// <param name="monitoredItemId">the id of the monitored item</param>
        /// <param name="createDurable">true if a durable queue shall be created</param>
        /// <param name="queueFactory">the factory for <see cref="IDataChangeMonitoredItemQueue"/></param>
        /// <param name="telemetry">The telemetry context to use to create obvservability instruments</param>
        /// <param name="discardedValueHandler">Handler for discarded values</param>
        public DataChangeQueueHandler(
            uint monitoredItemId,
            bool createDurable,
            IMonitoredItemQueueFactory queueFactory,
            ITelemetryContext telemetry,
            Action? discardedValueHandler = null)
            : this(monitoredItemId, createDurable, queueFactory, telemetry, discardedValueHandler, null)
        {
        }

        /// <summary>
        /// Creates a new Queue with an explicit <see cref="TimeProvider"/> so
        /// the per-sample timestamp used for sampling-interval throttling can
        /// be mocked in tests.
        /// </summary>
        public DataChangeQueueHandler(
            uint monitoredItemId,
            bool createDurable,
            IMonitoredItemQueueFactory queueFactory,
            ITelemetryContext telemetry,
            Action? discardedValueHandler,
            TimeProvider? timeProvider)
        {
            m_logger = telemetry.CreateLogger<DataChangeQueueHandler>();
            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;
        }

        /// <summary>
        /// Create a DatachangeQueueHandler from an existing queue
        /// Used for restore after a server restart
        /// </summary>
        public DataChangeQueueHandler(
            IDataChangeMonitoredItemQueue dataValueQueue,
            bool discardOldest,
            double samplingInterval,
            ITelemetryContext telemetry,
            Action? discardedValueHandler = null)
            : this(dataValueQueue, discardOldest, samplingInterval, telemetry, discardedValueHandler, null)
        {
        }

        /// <summary>
        /// Create a DatachangeQueueHandler from an existing queue with an
        /// explicit <see cref="TimeProvider"/>.
        /// </summary>
        public DataChangeQueueHandler(
            IDataChangeMonitoredItemQueue dataValueQueue,
            bool discardOldest,
            double samplingInterval,
            ITelemetryContext telemetry,
            Action? discardedValueHandler,
            TimeProvider? timeProvider)
        {
            m_logger = telemetry.CreateLogger<DataChangeQueueHandler>();

            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);
        }

        /// <summary>
        /// Sets the queue size.
        /// </summary>
        /// <param name="queueSize">The new queue size.</param>
        /// <param name="discardOldest">Whether to discard the oldest values if the queue overflows.</param>
        /// <param name="diagnosticsMasks">Specifies which diagnostics which should be kept in the queue.</param>
        public void SetQueueSize(
            uint queueSize,
            bool discardOldest,
            DiagnosticsMasks diagnosticsMasks)
        {
            bool queueErrors = (diagnosticsMasks & DiagnosticsMasks.OperationAll) != 0;

            m_discardOldest = discardOldest;

            // copy existing values.
            List<DataValue>? existingValues = null;
            List<ServiceResult>? existingErrors = null;

            if (ItemsInQueue > 0)
            {
                existingValues = new List<DataValue>((int)queueSize);
                existingErrors = new List<ServiceResult>((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]);
                }
            }
        }

        /// <summary>
        /// Set the sampling interval of the queue
        /// </summary>
        /// <param name="samplingInterval">the sampling interval</param>
        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;
            }
        }

        /// <summary>
        /// Number of DataValues in the queue
        /// </summary>
        public int ItemsInQueue => m_dataValueQueue.ItemsInQueue;

        /// <summary>
        /// Queues a value
        /// </summary>
        /// <param name="value">the dataValue</param>
        /// <param name="error">the error</param>
        /// <returns>true of overflow occured</returns>
        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);
        }

        /// <summary>
        /// Deques the last item
        /// </summary>
        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;
        }

        /// <summary>
        /// Enque value
        /// </summary>
        /// <returns>true of overflow occured</returns>
        /// <exception cref="ServiceResultException"></exception>
        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;
        }

        /// <summary>
        /// Sets the overflow bit in the value and error.
        /// </summary>
        /// <param name="value">The value to update.</param>
        /// <param name="error">The error to update.</param>
        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);
            }
        }

        /// <inheritdoc/>
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Overridable method to dispose of resources.
        /// </summary>
        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;
    }
}