UA-.NETStandard/samples/Quickstarts.Servers/DurableSubscription/DurableEventMonitoredItemQueue.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 System.Threading;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;

namespace Quickstarts.Servers
{
    public class DurableEventMonitoredItemQueue : IEventMonitoredItemQueue
    {
        private const uint kMaxNoOfEntriesCheckedForDuplicateEvents = 1000;
        private const uint kBatchSize = 1000;

        /// <summary>
        /// Invoked when the queue is disposed
        /// </summary>
        public event EventHandler? Disposed;

        /// <summary>
        /// Creates an empty queue.
        /// </summary>
        public DurableEventMonitoredItemQueue(
            bool createDurable,
            uint monitoredItemId,
            IBatchPersistor batchPersistor,
            ITelemetryContext telemetry)
        {
            IsDurable = createDurable;
            m_logger = telemetry.CreateLogger<DurableEventMonitoredItemQueue>();
            m_batchPersistor = batchPersistor;
            MonitoredItemId = monitoredItemId;
            QueueSize = 0;
            ItemsInQueue = 0;
            m_enqueueBatch = new EventBatch([], kBatchSize, monitoredItemId);
            m_dequeueBatch = m_enqueueBatch;
        }

        /// <summary>
        /// Creates a queue from a template
        /// </summary>
        public DurableEventMonitoredItemQueue(
            StorableEventQueue queue,
            IBatchPersistor batchPersistor)
        {
            IsDurable = queue.IsDurable;
            MonitoredItemId = queue.MonitoredItemId;
            QueueSize = queue.QueueSize;
            m_batchPersistor = batchPersistor;
            m_logger = NullLogger<DurableEventMonitoredItemQueue>.Instance;

            m_enqueueBatch = queue.EnqueueBatch ?? new EventBatch([], kBatchSize, MonitoredItemId);
            // a null dequeue batch means it was the enqueue batch when the queue was stored.
            m_dequeueBatch = queue.DequeueBatch ?? m_enqueueBatch;

            // batches that come back without events hold nothing to deliver; dropping them
            // keeps Dequeue from promoting an empty batch while events remain queued.
            if (queue.EventBatches != null)
            {
                foreach (EventBatch batch in queue.EventBatches)
                {
                    if (batch?.Events != null && batch.Events.Count > 0)
                    {
                        m_eventBatches.Add(batch);
                    }
                }
            }

            if (m_dequeueBatch != m_enqueueBatch && m_dequeueBatch.Events.Count == 0)
            {
                if (m_eventBatches.Count > 0)
                {
                    m_dequeueBatch = m_eventBatches[0];
                    m_eventBatches.RemoveAt(0);
                }
                else
                {
                    m_dequeueBatch = m_enqueueBatch;
                }
            }

            // the stored format has no count: derive it from the restored events so that
            // Dequeue, the duplicate check and the queue size handling see them.
            int itemsInQueue = m_enqueueBatch.Events.Count;
            if (m_dequeueBatch != m_enqueueBatch)
            {
                itemsInQueue += m_dequeueBatch.Events.Count;
            }
            foreach (EventBatch batch in m_eventBatches)
            {
                itemsInQueue += batch.Events.Count;
            }
            ItemsInQueue = itemsInQueue;
        }

        /// <inheritdoc/>
        public bool IsDurable { get; }

        /// <inheritdoc/>
        public uint MonitoredItemId { get; }

        /// <inheritdoc/>
        public uint QueueSize { get; protected set; }

        /// <inheritdoc/>
        public int ItemsInQueue { get; private set; }

        /// <inheritdoc/>
        public bool Dequeue(out EventFieldList value)
        {
            value = null!;
            if (ItemsInQueue > 0)
            {
                if (m_dequeueBatch.PersistingInProgress || m_dequeueBatch.IsPersisted)
                {
                    m_logger.DequeueRequestedBeforeRestore(MonitoredItemId);
                    m_batchPersistor.RequestBatchRestore(m_dequeueBatch);

                    if (!SpinWait.SpinUntil(
                        () => !m_dequeueBatch.PersistingInProgress && !m_dequeueBatch.IsPersisted, 10))
                    {
                        m_logger.DequeueFailedBeforeRestore(MonitoredItemId);
                        // Dequeue failed as queue could not be restored in time
                        return false;
                    }
                }

                value = m_dequeueBatch.Events[0];
                m_dequeueBatch.Events.RemoveAt(0);
                ItemsInQueue--;
                HandleDequeBatching();
                return true;
            }
            return false;
        }

        /// <inheritdoc/>
        public void Enqueue(EventFieldList value)
        {
            if (QueueSize == 0)
            {
                throw new ServiceResultException(
                    StatusCodes.BadInternalError,
                    "Error queueing Event. Queue size is set to 0");
            }

            //Discard oldest
            if (ItemsInQueue == QueueSize)
            {
                Dequeue(out EventFieldList _);
            }

            m_enqueueBatch.Events.Add(value);
            ItemsInQueue++;
            HandleEnqueueBatching();
        }

        /// <summary>
        /// persists batches if needed
        /// </summary>
        private void HandleEnqueueBatching()
        {
            // Store the batch if it is full
            if (m_enqueueBatch.Events.Count >= kBatchSize)
            {
                // Special case: if the enqueue and dequeue batch are the same only one batch exists, so no storing is needed
                if (m_dequeueBatch == m_enqueueBatch)
                {
                    m_dequeueBatch = new EventBatch(
                        m_enqueueBatch.Events,
                        kBatchSize,
                        MonitoredItemId);
                    m_enqueueBatch = new EventBatch([], kBatchSize, MonitoredItemId);
                }
                // persist the batch
                else
                {
                    m_logger.StoringBatch(MonitoredItemId);

                    var batchToStore = new EventBatch(
                        m_enqueueBatch.Events,
                        kBatchSize,
                        MonitoredItemId);
                    m_eventBatches.Add(batchToStore);
                    //only persist second batch in list, as the first could be needed, for duplicate event check
                    if (m_eventBatches.Count > 1)
                    {
                        m_batchPersistor.RequestBatchPersist(m_eventBatches[^2]);
                    }

                    m_enqueueBatch = new EventBatch([], kBatchSize, MonitoredItemId);
                }
            }
        }

        /// <summary>
        /// Restores batches if needed
        /// </summary>
        private void HandleDequeBatching()
        {
            // request a restore if the dequeue batch is half empty
            if (m_dequeueBatch.Events.Count <= kBatchSize / 2 && m_eventBatches.Count > 0)
            {
                m_batchPersistor.RequestBatchRestore(m_eventBatches[0]);
            }

            // if the dequeue batch is empty and there are stored batches, set the dequeue batch to the first stored batch
            if (m_dequeueBatch.Events.Count == 0)
            {
                if (m_eventBatches.Count > 0 && m_dequeueBatch != m_enqueueBatch)
                {
                    m_dequeueBatch = m_eventBatches[0];
                    m_eventBatches.RemoveAt(0);

                    // Request a restore for the next batch if there is one
                    if (m_eventBatches.Count > 0)
                    {
                        m_batchPersistor.RequestBatchRestore(m_eventBatches[0]);
                    }
                }
                else
                {
                    //only one batch exists
                    m_dequeueBatch = m_enqueueBatch;
                }
            }
        }

        /// <inheritdoc/>
        public bool IsEventContainedInQueue(IFilterTarget instance)
        {
            // a duplicate is the same instance reported again through another notifier
            // path, so it sits among the most recently queued events: scan from the newest
            // (the enqueue batch, then the stored batches from the latest backwards, then
            // the dequeue batch, which holds the oldest events).
            int remaining =
                ItemsInQueue > kMaxNoOfEntriesCheckedForDuplicateEvents
                    ? (int)kMaxNoOfEntriesCheckedForDuplicateEvents
                    : ItemsInQueue;

            if (ContainsHandle(m_enqueueBatch.Events, instance, ref remaining))
            {
                return true;
            }

            for (int batch = m_eventBatches.Count - 1; batch >= 0 && remaining > 0; batch--)
            {
                // older batches are persisted (no events in memory) and hold no recent duplicate.
                List<EventFieldList>? events = m_eventBatches[batch].Events;
                if (events == null)
                {
                    return false;
                }
                if (ContainsHandle(events, instance, ref remaining))
                {
                    return true;
                }
            }

            return m_dequeueBatch != m_enqueueBatch &&
                m_dequeueBatch.Events != null &&
                ContainsHandle(m_dequeueBatch.Events, instance, ref remaining);
        }

        /// <summary>
        /// Scans the events from the newest to the oldest for an event raised for the
        /// instance, checking at most <paramref name="remaining"/> events.
        /// </summary>
        private static bool ContainsHandle(
            List<EventFieldList> events,
            IFilterTarget instance,
            ref int remaining)
        {
            for (int i = events.Count - 1; i >= 0 && remaining > 0; i--, remaining--)
            {
                if (events[i] is EventFieldList queuedEvent &&
                    EventMonitoredItemQueue.IsEventRaisedFor(queuedEvent, instance))
                {
                    return true;
                }
            }
            return false;
        }

        /// <inheritdoc/>
        public void SetQueueSize(uint queueSize, bool discardOldest)
        {
            QueueSize = queueSize;

            if (ItemsInQueue > QueueSize)
            {
                //ToDo: Remove files

                int itemsToRemove = (int)(ItemsInQueue - QueueSize);

                if (discardOldest)
                {
                    // Remove from output batch
                    while (itemsToRemove > 0 && m_dequeueBatch.Events.Count > 0)
                    {
                        // Remove from output batch
                        int removeCount = Math.Min(itemsToRemove, m_dequeueBatch.Events.Count);
                        m_dequeueBatch.Events.RemoveRange(0, removeCount);
                        ItemsInQueue -= removeCount;
                        itemsToRemove -= removeCount;
                    }

                    // Remove from stored batches if needed
                    while (itemsToRemove > 0 && m_eventBatches.Count > 0)
                    {
                        EventBatch batch = m_eventBatches[0];
                        m_batchPersistor.RestoreSynchronously(batch);
                        int batchCount = batch.Events.Count;

                        if (itemsToRemove >= batchCount)
                        {
                            m_eventBatches.RemoveAt(0);
                            ItemsInQueue -= batchCount;
                            itemsToRemove -= batchCount;
                        }
                        else
                        {
                            batch.Events.RemoveRange(0, itemsToRemove);
                            ItemsInQueue -= itemsToRemove;
                            itemsToRemove = 0;
                        }
                    }

                    // Remove from output batch
                    while (itemsToRemove > 0 && m_enqueueBatch.Events.Count > 0)
                    {
                        // Remove from output batch
                        int removeCount = Math.Min(itemsToRemove, m_enqueueBatch.Events.Count);
                        m_enqueueBatch.Events.RemoveRange(0, removeCount);
                        ItemsInQueue -= removeCount;
                        itemsToRemove -= removeCount;
                    }
                }
                else
                {
                    // Remove from input batch
                    while (itemsToRemove > 0 && m_enqueueBatch.Events.Count > 0)
                    {
                        int removeCount = Math.Min(itemsToRemove, m_enqueueBatch.Events.Count);
                        m_enqueueBatch.Events
                            .RemoveRange(m_enqueueBatch.Events.Count - removeCount, removeCount);
                        ItemsInQueue -= removeCount;
                        itemsToRemove -= removeCount;
                    }

                    // Remove from stored batches if needed
                    while (itemsToRemove > 0 && m_eventBatches.Count > 0)
                    {
                        EventBatch batch = m_eventBatches[^1];
                        m_batchPersistor.RestoreSynchronously(batch);
                        int batchCount = batch.Events.Count;

                        if (itemsToRemove >= batchCount)
                        {
                            m_eventBatches.RemoveAt(m_eventBatches.Count - 1);
                            ItemsInQueue -= batchCount;
                            itemsToRemove -= batchCount;
                        }
                        else
                        {
                            batch.Events
                                .RemoveRange(batch.Events.Count - itemsToRemove, itemsToRemove);
                            ItemsInQueue -= itemsToRemove;
                            itemsToRemove = 0;
                        }
                    }

                    // Remove from output batch
                    while (itemsToRemove > 0 && m_dequeueBatch.Events.Count > 0)
                    {
                        int removeCount = Math.Min(itemsToRemove, m_dequeueBatch.Events.Count);
                        m_dequeueBatch.Events
                            .RemoveRange(m_dequeueBatch.Events.Count - removeCount, removeCount);
                        ItemsInQueue -= removeCount;
                        itemsToRemove -= removeCount;
                    }
                }
            }
        }

        /// <summary>
        /// Brings the queue with contents into a storable format
        /// </summary>
        public StorableEventQueue ToStorableQueue()
        {
            // the stored queue gets detached copies of resident batches: a batch spilled to
            // disk is loaded back (its file is keyed by a batch id that does not survive a
            // restart) and a batch whose spill is in flight is copied before the persistor
            // can clear its events, so the copies stay stable while they are encoded.
            EventBatch enqueueBatch = DetachBatch(m_enqueueBatch);
            EventBatch dequeueBatch = ReferenceEquals(m_dequeueBatch, m_enqueueBatch)
                ? enqueueBatch
                : DetachBatch(m_dequeueBatch);

            return new StorableEventQueue
            {
                IsDurable = IsDurable,
                MonitoredItemId = MonitoredItemId,
                DequeueBatch = dequeueBatch,
                EnqueueBatch = enqueueBatch,
                EventBatches = m_eventBatches.ConvertAll(DetachBatch),
                QueueSize = QueueSize
            };
        }

        /// <summary>
        /// Returns a resident copy of the batch that the persistor cannot change.
        /// </summary>
        private EventBatch DetachBatch(EventBatch batch)
        {
            for (int attempt = 0; attempt < kMaxDetachAttempts; attempt++)
            {
                // the persistor clears the events of a spilled batch under this lock.
                lock (batch)
                {
                    if (!batch.IsPersisted && batch.Events != null)
                    {
                        return new EventBatch([.. batch.Events], batch.BatchSize, batch.MonitoredItemId);
                    }
                }

                m_batchPersistor.RestoreSynchronously(batch);
            }

            m_logger.SpilledBatchNotLoadedForStore(MonitoredItemId);
            return new EventBatch([], batch.BatchSize, batch.MonitoredItemId);
        }

        private const int kMaxDetachAttempts = 3;

        /// <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)
            {
                Disposed?.Invoke(this, EventArgs.Empty);
            }
        }

        /// <summary>
        /// the contained in the queue
        /// </summary>
        private EventBatch m_enqueueBatch;
        private readonly List<EventBatch> m_eventBatches = [];
        private EventBatch m_dequeueBatch;
        private readonly IBatchPersistor m_batchPersistor;
        private readonly ILogger m_logger;
    }

    /// <summary>
    /// Batch of events
    /// </summary>
    public class EventBatch : BatchBase
    {
        public EventBatch(List<EventFieldList> events, uint batchSize, uint monitoredItemId)
            : base(batchSize, monitoredItemId)
        {
            Events = events;
        }

        public List<EventFieldList> Events { get; private set; }

        public override void SetPersisted()
        {
            Events = null!;
            IsPersisted = true;
            PersistingInProgress = false;
        }

        public void Restore(List<EventFieldList>? events)
        {
            Events = events ?? [];
            IsPersisted = false;
            RestoreInProgress = false;
        }
    }

    public class StorableEventQueue
    {
        public bool IsDurable { get; set; }

        public uint MonitoredItemId { get; set; }

        public EventBatch? EnqueueBatch { get; set; }

        public List<EventBatch> EventBatches { get; set; } = null!;

        public EventBatch? DequeueBatch { get; set; }

        public uint QueueSize { get; set; }
    }
}