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 : 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 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 BuildFetchQuery(IQueryable 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 BuildFetchOrder(IQueryable query) { return query.OrderBy(x => x.Priority).ThenBy(x => x.StartFrom); } protected override void OnDisposing() { } } }