Получение результата вызова функции из очереди

        public static BlockingCollection<object>? BlockingCollectionQueue { get; set; }
 
        public static void ProcessQueue()
        {
            object itemNext;
            while (!IsCancelled && BlockingCollectionQueue .TryTake(out itemNext, Timeout.Infinite)) // [OFF] p.s. я знаю что такое CancellationToken, пока не до этого, сбыдлокодил как было =)[/OFF]
            {
                [B]ProcessingFunction(itemNext);[/B] // отсюда хочу вернуть значение каким-то образом, может передавать функцию в очередь? Но тогда всё равно неясно, как получить в вызывающий метод результат выполнения именно того, что было передано.
            }
        }
 
        public static async Task<object> CallOfProcessing(object argObject)
        {
            BlockingCollectionQueue .Add(argObject);
            return Task<object>.Run(() => [B]??????????[/B]); // Как вернуть сюда искомое?
        }

    public static class SomeStaticClass
    {
        private static BlockingCollection<QueueEntry>? BlockingCollectionQueue { get; set; }
 
        private static List<QueueEntry>? ResultsList { get; set; }
 
        public static void ProcessQueue()
        {
            while (!IsCancelled && BlockingCollectionQueue.TryTake(out var objToRecognizeNext, Timeout.Infinite))
            {
                objToRecognizeNext.result =
                    ProcessingFunction(objToRecognizeNext);;
                ResultsList.Add(objToRecognizeNext);
            }
        }
 
        public static object CallOfProcessing(object argObject)
        {
            var entry = new QueueEntry(argObject);
            BlockingCollectionQueue.TryAdd(entry);
            
            var result = ResultsList.FirstOrDefault(x => x._guid == entry._guid);
            while (result._guid == Guid.Empty)
            {
                result = ResultsList.FirstOrDefault(x => x._guid == entry._guid);
                Thread.Sleep(5);
            }
 
            ResultsList.Remove(result);
            return result.result;
        }
 
        public struct QueueEntry
        {
            public Guid _guid;
            public object args;
            public string result;
 
            public QueueEntry(object argsParam)
            {
                _guid = Guid.NewGuid();
                args = argsParam;
                result = string.Empty;
            }
        }
    }

    public static class SomeStaticClass
    {
        /// <summary>
        /// Очередь на обработку
        /// </summary>
        private static BlockingCollection<QueueEntry>? BlockingCollectionQueue { get; set; }
 
        /// <summary>
        /// Результаты выполнения
        /// </summary>
        private static BlockingCollection<QueueEntry>? ResultsList { get; set; }
 
        /// <summary>
        /// Флаг отмены операции
        /// </summary>
        private static bool IsCancelled { get; set; }
 
        /// <summary>
        /// Запуск обработки очереди
        /// </summary>
        private static void ProcessQueue()
        {
            while (!IsCancelled && BlockingCollectionQueue.TryTake(out var objToRecognizeNext, Timeout.Infinite))
            {
                objToRecognizeNext.result =
                    processingObject.ProcessingFunction(objToRecognizeNext);;
                ResultsList.Add(objToRecognizeNext);
            }
        }
 
        /// <summary>
        /// Тот самый объект-обработчик.
        /// Который раньше я инициализировал для каждого вызова.
        /// </summary>
        private static ProcessingObjectType processingObject = new ProcessingObjectType();
        
        /// <summary>
        /// Моя функция для использования этого обработчика
        /// </summary>
        /// <param name="processingObject">Собственно обработчик</param>
        /// <param name="objToRecognizeNext">Обрабатываемая информация</param>
        /// <returns>Результат</returns>
        private static string ProcessingFunction(this ProcessingObjectType processingObject, QueueEntry objToRecognizeNext)
        {
            return processingObject.SomeProcess(objToRecognizeNext);
        }
 
        /// <summary>
        /// Вызываемый в контексте метод
        /// </summary>
        /// <param name="argObject">Обрабатываемые аргументы</param>
        /// <returns>Обработанный объект</returns>
        public static object CallOfProcessing(object argObject)
        {
            var entry = new QueueEntry(argObject);
            BlockingCollectionQueue.TryAdd(entry);
            
            
            var result = ResultsList.FirstOrDefault(x => x._guid == entry._guid);
            while (result._guid == Guid.Empty)
            {
                result = ResultsList.FirstOrDefault(x => x._guid == entry._guid);
                Thread.Sleep(5);
            }
 
            ResultsList.TryTake(out result);
            return result.result;
        }
 
        internal struct QueueEntry
        {
            public Guid _guid;
            public object args;
            public string result;
 
            public QueueEntry(object argsParam)
            {
                _guid = Guid.NewGuid();
                args = argsParam;
                result = string.Empty;
            }
        }
    }
 
    /// <summary>
    /// Внешняя библиотека, которой я пользуюсь
    /// </summary>
    internal class ProcessingObjectType
    {
        /// <summary>
        /// Какой-то её внутренний процесс, который не позволяет запускать обработчик асинхронно
        /// </summary>
        /// <param name="objToRecognizeNext">Аргумент</param>
        /// <returns>Результат</returns>
        public string SomeProcess(SomeStaticClass.QueueEntry objToRecognizeNext)
        {
            // someWork with objToRecognizeNext.args; 
            return string.Empty;
        }
    }

// Объект из сторонней сбоки, выполняющий всю непосредственную работу. В одном экземпляре.
private readonly ProcessingObjectType processor = new();
 
// Обработчик запросов со встроенным буфером/очередью.
// Добавьте using System.Threading.Tasks.Dataflow
private readonly ActionBlock<QueueEntry> Requests = new(
    ProcessQueueEntry, // Ссылка на метод, обрабатывающий каждый элемент очереди (см. ниже)
    new ExecutionDataflowBlockOptions 
    { 
        BoundedCapacity = 100, // Максимальное количество элементов в очереди. Если очередь полна, то SendAsync (см. ниже) будет асинхронно ждать освобождения.
        MaxDegreeOfParallelism = 1, // Количество элементов, обрабатываемых одновременно. У вас ограничение на один, потому единица.
    });
 
// Пользователи вызывают этот метод: string result = await instance.CallOfProcessing(arg, token);
public async Task<string> CallOfProcessing(object args, CancellationToken token = default)
{
    using QueueEntry entry = new(args, token);
 
    // Добавление элемента в очередь на обработку.
    // Если очередь полна (InputCount == BoundedCapacity), то выполнение будет асинхронно ожидать свободного места.
    await Requests.SendAsync(entry, token).ConfigureAwait(false);
 
    Console.WriteLine($"Enqueued {args} (current queue: {Requests.InputCount})");
 
    // Ожидание окончания работы: пока до элемент дойдет очередь и обработчик не произведет результат.
    return await entry.Completion.ConfigureAwait(false);
}
 
// Этот метод вызывается для каждого элемента очереди.
// Может быть асинхронным — зависит от реализации метода из сторонней сборки.
private void ProcessQueueEntry(QueueEntry entry)
{
    string result = processor.SomeProcess(entry.Args, entry.CancellationToken);
 
    // Уведомление о том, что выполнение закончено: в этот момент код с string result = await CallOfProcessing(arg, token) завершится.
    entry.Complete(result);
}
 
private readonly struct QueueEntry : IDisposable
{
    private readonly TaskCompletionSource<string> _tcs;
    private readonly CancellationTokenSource _cancellation;
    private readonly CancellationTokenRegistration _cancellationRegistration;
 
    public QueueEntry(object argsParam, CancellationToken token = default) : this()
    {
        Args = argsParam;
        _tcs = new();
 
        // Обработка отмены
        _cancellation = CancellationTokenSource.CreateLinkedTokenSource(token);
        _cancellationRegistration = _cancellation.Token.Register(Cancel);
    }
    public object Args { get; }
    public CancellationToken CancellationToken => _cancellation.Token;
    public Task<string> Completion => _tcs.Task;
 
    public void Dispose()
    {
        _cancellation.Dispose();
        _cancellationRegistration.Dispose();
    }
    internal void Cancel()
    {
        _tcs.TrySetCanceled(CancellationToken);
    }
    internal void Complete(string result) => _tcs.TrySetResult(result);
}
 
private class ProcessingObjectType
{
    static readonly Random rand = new();
    public string SomeProcess(object arg, CancellationToken cancellationToken)
    {
        // Имитируем бурную деятельность.
        TimeSpan timeToComplete = TimeSpan.FromSeconds(rand.Next(15));
        Console.WriteLine($"Work on {arg} will complete in {timeToComplete}");
        Task.Delay(timeToComplete, cancellationToken).Wait();
 
        return $"Work on {arg} completed in {timeToComplete}";
    }
}

ProcessorWrapper processor = new(); // Класс, инкапсулирующий всю вышеприведенную логику.
CancellationTokenSource massCancellation = new();
while (true)
{
    string arg = Console.ReadLine();
    if (string.IsNullOrEmpty(arg)) 
        break;
 
    // Ожидаем окончания в отдельном потоке из пула — чтобы не блокировать ввод консоли для следующей задачи и для имитации одновременных запросов разными пользователями
    _ = Task.Run(async () => Console.WriteLine(await processor.CallOfProcessing(arg, massCancellation.Token)));
}
 
// Отменяем все висящие и активные запросы из очереди.
massCancellation.Cancel();

    public static class ProcessingObjectTypeSync
    {
        // Единственный экземпляр обработчика
        private static readonly ProcessingObjectType processing = new();
 
        // Синхронный локирующий метод
        // Если было несколько вызовов, то блокируется выполнени
        // до завершение предыдущих вызово
        public static string SomeProcess(QueueEntry objToRecognizeNext)
        {
            lock (processing)
                return processing.SomeProcess(objToRecognizeNext);
        }
 
        // Перегрузка для Фабрики Task, чтобы не создавать лямбд
        private static string SomeProcess(object obj)
            => SomeProcess((QueueEntry)obj);
 
        // Асинхронный метод не блокирующий вызывающий поток,
        // если одновременно несколько вызовов.
        public static async Task<string> SomeProcessAsync(QueueEntry objToRecognizeNext)
            => await Task.Factory.StartNew(SomeProcess, objToRecognizeNext);
    }