using System.Collections.Concurrent; using Microsoft.Extensions.DependencyInjection; namespace Connected.Collections.Concurrent; internal sealed class QueuedDispatcher : IDispatcher where TJob : IDispatcherJob { public event EventHandler? Completed; public QueuedDispatcher(IDispatcher dispatcher, string queueName) { Dispatcher = dispatcher; Queue = new(); QueueName = queueName; /* * Dispatcher jobs should be transient so it's safe to request a service from the root collection. */ if (CollectionsStartup.Application?.Services.GetService>() is not DispatcherJob job) throw new SysException(this, $"{SR.ErrCreateService} ({typeof(DispatcherJob).Name})"); job.Completed += OnCompleted; Job = job; } public CancellationToken CancellationToken => Dispatcher.CancellationToken; public bool IsDisposed { get; private set; } public DispatcherProcessBehavior Behavior => DispatcherProcessBehavior.Queued; public string QueueName { get; } private DispatcherJob Job { get; set; } private IDispatcher Dispatcher { get; set; } public int Count => Queue.Count; private ConcurrentQueue Queue { get; set; } public void Cancel() { } public bool Dequeue(out TArgs? item) { return Queue.TryDequeue(out item); } public bool Enqueue(TArgs item) { if (IsDisposed) return false; Queue.Enqueue(item); if (!Job.IsRunning) Job.Run(Queue, CancellationToken); return true; } private void OnCompleted(object? sender, EventArgs e) { if (sender is not DispatcherJob job) return; if (!Queue.IsEmpty) { job.Run(Queue, CancellationToken); return; } Completed?.Invoke(this, EventArgs.Empty); } public bool Enqueue(string queue, TArgs args) { return Dispatcher.Enqueue(queue, args); } private void Dispose(bool disposing) { if (!IsDisposed) { if (disposing) { if (Job is not null) { Job.Dispose(); Job = null; } } IsDisposed = true; } } public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } }