|
|
|
|
using System.Collections.Concurrent;
|
|
|
|
|
using Connected;
|
|
|
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
|
|
|
|
|
|
|
|
namespace Connected.Collections.Concurrent;
|
|
|
|
|
|
|
|
|
|
public abstract class Dispatcher<TArgs, TJob> : IDispatcher<TArgs, TJob>
|
|
|
|
|
where TJob : IDispatcherJob<TArgs>
|
|
|
|
|
{
|
|
|
|
|
private CancellationTokenSource _tokenSource;
|
|
|
|
|
protected Dispatcher(int size)
|
|
|
|
|
{
|
|
|
|
|
WorkerSize = size;
|
|
|
|
|
|
|
|
|
|
_tokenSource = new();
|
|
|
|
|
|
|
|
|
|
Queue = new();
|
|
|
|
|
Jobs = new();
|
|
|
|
|
QueuedDispatchers = new();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public CancellationToken CancellationToken => _tokenSource.Token;
|
|
|
|
|
private ConcurrentQueue<TArgs> Queue { get; set; }
|
|
|
|
|
private List<DispatcherJob<TArgs>> Jobs { get; set; }
|
|
|
|
|
protected bool IsDisposed { get; private set; }
|
|
|
|
|
private int WorkerSize { get; }
|
|
|
|
|
public int Available => Math.Max(0, WorkerSize * 4 - Queue.Count - QueuedDispatchers.Sum(f => f.Value.Count));
|
|
|
|
|
private ConcurrentDictionary<string, QueuedDispatcher<TArgs, TJob>> QueuedDispatchers { get; set; }
|
|
|
|
|
public DispatcherProcessBehavior Behavior => DispatcherProcessBehavior.Parallel;
|
|
|
|
|
|
|
|
|
|
public void Cancel()
|
|
|
|
|
{
|
|
|
|
|
_tokenSource?.Cancel();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public bool Dequeue(out TArgs? item)
|
|
|
|
|
{
|
|
|
|
|
return Queue.TryDequeue(out item);
|
|
|
|
|
}
|
|
|
|
|
public bool Enqueue(string queue, TArgs item)
|
|
|
|
|
{
|
|
|
|
|
if (EnsureDispatcher(queue) is not QueuedDispatcher<TArgs, TJob> dispatcher)
|
|
|
|
|
throw new SysException(this, $"{SR.ErrCreateQueuedDispatcher} ({queue})");
|
|
|
|
|
|
|
|
|
|
return dispatcher.Enqueue(item);
|
|
|
|
|
}
|
|
|
|
|
public bool Enqueue(TArgs item)
|
|
|
|
|
{
|
|
|
|
|
Queue.Enqueue(item);
|
|
|
|
|
|
|
|
|
|
if (Jobs.Count < WorkerSize)
|
|
|
|
|
{
|
|
|
|
|
/*
|
|
|
|
|
* Dispatcher jobs should be transient so it's safe to request a service from the root collection.
|
|
|
|
|
*/
|
|
|
|
|
if (CollectionsStartup.Application.Services.GetService<TJob>() is not DispatcherJob<TArgs> job)
|
|
|
|
|
throw new NullReferenceException($"{SR.ErrCreateService} ({typeof(DispatcherJob<TArgs>).Name})");
|
|
|
|
|
|
|
|
|
|
job.Completed += OnCompleted;
|
|
|
|
|
|
|
|
|
|
lock (Jobs)
|
|
|
|
|
{
|
|
|
|
|
Jobs.Add(job);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
job.Run(Queue, CancellationToken);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return true;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void OnCompleted(object? sender, EventArgs e)
|
|
|
|
|
{
|
|
|
|
|
if (sender is not DispatcherJob<TArgs> job)
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
|
|
if (Queue.IsEmpty)
|
|
|
|
|
{
|
|
|
|
|
lock (Jobs)
|
|
|
|
|
{
|
|
|
|
|
Jobs.Remove(job);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
job.Dispose();
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
job.Run(Queue, CancellationToken);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private QueuedDispatcher<TArgs, TJob>? EnsureDispatcher(string queueName)
|
|
|
|
|
{
|
|
|
|
|
if (QueuedDispatchers.TryGetValue(queueName, out QueuedDispatcher<TArgs, TJob>? result))
|
|
|
|
|
return result;
|
|
|
|
|
|
|
|
|
|
result = new QueuedDispatcher<TArgs, TJob>(this, queueName);
|
|
|
|
|
|
|
|
|
|
result.Completed += OnQueuedCompleted;
|
|
|
|
|
|
|
|
|
|
if (!QueuedDispatchers.TryAdd(queueName, result))
|
|
|
|
|
{
|
|
|
|
|
result.Completed -= OnQueuedCompleted;
|
|
|
|
|
|
|
|
|
|
if (QueuedDispatchers.TryGetValue(queueName, out QueuedDispatcher<TArgs, TJob>? retryResult))
|
|
|
|
|
return retryResult;
|
|
|
|
|
else
|
|
|
|
|
return default;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return result;
|
|
|
|
|
}
|
|
|
|
|
private void OnQueuedCompleted(object? sender, EventArgs e)
|
|
|
|
|
{
|
|
|
|
|
if (sender is not QueuedDispatcher<TArgs, TJob> dispatcher)
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
|
|
if (dispatcher.Count > 0)
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
|
|
QueuedDispatchers.Remove(dispatcher.QueueName, out _);
|
|
|
|
|
|
|
|
|
|
dispatcher.Dispose();
|
|
|
|
|
}
|
|
|
|
|
private void Dispose(bool disposing)
|
|
|
|
|
{
|
|
|
|
|
if (!IsDisposed)
|
|
|
|
|
{
|
|
|
|
|
if (disposing)
|
|
|
|
|
{
|
|
|
|
|
if (_tokenSource is not null)
|
|
|
|
|
{
|
|
|
|
|
if (!_tokenSource.IsCancellationRequested)
|
|
|
|
|
_tokenSource.Cancel();
|
|
|
|
|
|
|
|
|
|
_tokenSource.Dispose();
|
|
|
|
|
_tokenSource = null;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (Queue is not null)
|
|
|
|
|
{
|
|
|
|
|
Queue.Clear();
|
|
|
|
|
Queue = null;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (Jobs is not null)
|
|
|
|
|
{
|
|
|
|
|
foreach (var job in Jobs)
|
|
|
|
|
job.Dispose();
|
|
|
|
|
|
|
|
|
|
Jobs.Clear();
|
|
|
|
|
Jobs = null;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (QueuedDispatchers is not null)
|
|
|
|
|
{
|
|
|
|
|
foreach (var dispatcher in QueuedDispatchers)
|
|
|
|
|
dispatcher.Value.Dispose();
|
|
|
|
|
|
|
|
|
|
QueuedDispatchers.Clear();
|
|
|
|
|
Queue = null;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
IsDisposed = true;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
protected virtual void OnDisposing()
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
public void Dispose()
|
|
|
|
|
{
|
|
|
|
|
Dispose(true);
|
|
|
|
|
GC.SuppressFinalize(this);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|