EVCB_OCPP.OCPPServer.16J/EVCB_OCPP.WSServer/Helper/QueueHandler.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using static System.Runtime.InteropServices.JavaScript.JSType;

namespace EVCB_OCPP.WSServer.Helper;

public class QueueHandler<T>
{
    public QueueHandler(Func<T,Task> handler = null,int maxConcurrency = 1)
    {
        _queue = new ();
        this._handler = handler;
        this._semaphore = new (maxConcurrency);
        this._handlerPair = new();
    }

    public void Enqueue(T data)
    {
        _queue.Enqueue(data);
        _ = TrtStartHandler();
    }
    public void Enqueue(T data, Func<T, Task> handler)
    {
        _queue.Enqueue(data);
        _handlerPair.Add(data, handler);
        _ = TrtStartHandler();
    }

    private readonly ConcurrentQueue<T> _queue;
    private readonly Func<T, Task> _handler;
    private readonly SemaphoreSlim _semaphore;
    private readonly Dictionary<T, Func<T, Task>> _handlerPair;

    private async Task TrtStartHandler()
    {
        if(_semaphore.Wait(0))
        {
            return;
        }
        if(!_queue.TryDequeue(out var data))
        {
            return;
        }
        var handler = GetHandler(data);
        await handler(data);
        _semaphore.Release();
        _ = TrtStartHandler();
    }

    private Func<T, Task> GetHandler(T data)
    {
        if (_handlerPair.TryGetValue(data, out Func<T, Task> value))
        {
            return value;
        }

        if (_handler is not null)
        {
            return _handler;
        }

        return (data) => Task.CompletedTask;
    }
}