EventQueue.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Waher.Persistence;
 
namespace Waher.Events.Queue
{
    public class EventQueue : EventSink
    {
        private Timer timer;
        private int eventLifetimeDays;
        private readonly string queueName;
        private IPersistedQueue queue = null;
        private bool errorState = false;
 
        public EventQueue(string ObjectID, string QueueName)
            : this(ObjectID, QueueName, 0, null)
        {
        }
 
        public EventQueue(string ObjectID, string QueueName, int EventLifetimeDays,
            TimeSpan? CleanupTime)
            : base(ObjectID)
        {
            this.queueName = QueueName;
 
            if (EventLifetimeDays <= 0 || !CleanupTime.HasValue)
            {
                this.timer = null;
                this.eventLifetimeDays = 0;
            }
            else
            {
                if (CleanupTime.Value < TimeSpan.Zero || CleanupTime.Value.TotalDays >= 1.0)
                    throw new ArgumentOutOfRangeException("Invalid time.", nameof(CleanupTime));
 
                int MillisecondsPerDay = 1000 * 60 * 60 * 24;
                int MsUntilNext = (int)((DateTime.Today.AddDays(1).Add(CleanupTime.Value) - DateTime.Now).TotalMilliseconds + 0.5);
 
                this.eventLifetimeDays = EventLifetimeDays;
                this.timer = new Timer(this.PurgeOldItems, null, MsUntilNext, MillisecondsPerDay);
            }
        }
 
        public override async Task DisposeAsync()
        {
            if (!(this.queue is null))
            {
                await this.queue.DisposeAsync();
                this.queue = null;
            }
 
            this.timer?.Dispose();
            this.timer = null;
 
            await base.DisposeAsync();
        }
 
        public override async Task Queue(Event Event)
        {
            if (this.queue is null)
                this.queue = await Database.GetQueue(this.queueName);
 
            QueuedEvent Item = new QueuedEvent(Event);
 
            if (await this.queue.Enqueue(Item, this.errorState ? 0 : 10000))
                this.errorState = false;
            else if (!this.errorState)
            {
                this.errorState = true;
 
                Event Error = new Event(EventType.Emergency, "Event Queue " + this.queueName +
                    " is full and new events cannot be queued.", this.ObjectID, string.Empty,
                    string.Empty, EventLevel.Major, string.Empty,
                    typeof(EventQueue).Namespace, string.Empty);
 
                Error.Avoid(this);
 
                Log.Event(Error);
            }
        }
 
        public async Task ClearQueue()
        {
            if (this.queue is null)
                this.queue = await Database.GetQueue(this.queueName);
 
            await this.queue.Clear();
        }
 
        private async void PurgeOldItems(object P)
        {
            try
            {
                await this.PurgeOldItems(DateTime.UtcNow.AddDays(-this.eventLifetimeDays));
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
        public async Task<int> PurgeOldItems(DateTime Limit)
        {
            int NrEvents = 0;
            object Item;
 
            Item = await this.queue.Peek();
 
            while (!(Item is null))
            {
                if (Item is QueuedEvent QueuedEvent && QueuedEvent.Timestamp > Limit)
                    break;
                    
                await this.queue.Dequeue();
                NrEvents++;
 
                Item = await this.queue.Peek();
            }
 
            if (NrEvents > 0)
            {
                KeyValuePair<string, object>[] Tags = new KeyValuePair<string, object>[]
                {
                    new KeyValuePair<string, object>("Limit", Limit),
                    new KeyValuePair<string, object>("NrEvents", NrEvents)
                };
 
                if (NrEvents == 1)
                    Log.Informational("Purging 1 event from the queue.", this.ObjectID, Tags);
                else
                    Log.Informational("Purging " + NrEvents.ToString() + " events from the queue.", this.ObjectID, Tags);
            }
 
            return NrEvents;
        }
 
    }
}