qliberace/Common/qdr.fnd.core.business/ProcessQueue/Queue.cs

using System;
using System.Collections.Generic;
using System.Linq;
using Quadarax.Foundation.Core.Data.Extensions;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Object;
using Quadarax.Foundation.Core.Value.Extensions;

namespace Quadarax.Foundation.Core.Business.ProcessQueue
{
    public class Queue<TProvider> : DisposableObject where TProvider : IQueueProvider
    {
        private TProvider _provider;
        private IQLogger _logger;
        private IQLog _log;
        private string _queueName;
        private TimeSpan? _timeout;
        
        protected TProvider Provider => _provider;
        protected IQLogger Logger => _logger;
        public string Name => _queueName;
        protected IQLog Log => _log;

        

        public Queue(TProvider provider,string queueName, TimeSpan? initialTimeout, IQLogger logger)
        {
            _provider = provider ?? throw new ArgumentNullException(nameof(provider));
            if (string.IsNullOrEmpty(queueName)) throw new ArgumentNullException(nameof(queueName));
            _queueName = queueName;
            _timeout = initialTimeout;

            if (logger != null)
            {
                _logger = logger;
                _log = logger.GetLogger(GetType());
            }
        }

        public IEnumerable<IQueueItem> Fetch(int itemsCount)
        {
            if (!Provider.IsOpen)
                Provider.Open();
            
            var items = BuildFetchOrder(BuildFetchQuery(Provider.Query())).FirstPage(itemsCount).ToArray();

            foreach (var item in items) item.Assign(Name);
            Provider.UpdateAndCommit(items);
            return items;
        }


        protected virtual IQueryable<IQueueItem> BuildFetchQuery(IQueryable<IQueueItem> query)
        {
            var now = DateTime.Now;
            return query.Where(x => x.State.In(QueueItemStateEnum.New, QueueItemStateEnum.Pending) &&
                                    x.StartFrom > now &&
                                    x.StartTo.GetValueOrDefault(now.AddHours(1)) < now && x.Affinity == null);
        }
        protected virtual IQueryable<IQueueItem> BuildFetchOrder(IQueryable<IQueueItem> query)
        {
            return query.OrderBy(x => x.Priority).ThenBy(x => x.StartFrom);
        }

        protected override void OnDisposing()
        {
        }
    }
}