C# Queuing async Task using BlockingCollection and process queue only after value returned for previous getter task in queue

Question

Recently, i had a requirement to queue async tasks and i was introduced to BlockingCollection in this link Queuing asynchronous task in C# It worked and i'm having a slight change in requirement and need your guidance. I'm using the BlockingCollection as in @Stephen Cleary answer

This is the BlockingCollection from that link

public sealed class ExecutionQueue
{
  //private readonly BlockingCollection<Func<Task>> _queue = new BlockingCollection<Func<Task>>();//commented this
  private readonly BlockingCollection<Task> _queue = new BlockingCollection<Task>();

  public ExecutionQueue() => Complete = Task.Run(() => ProcessQueueAsync());

  public Task Completion { get; }

  public void Complete() => _queue.CompleteAdding();

  private async Task ProcessQueueAsync()
  {
    foreach (var value in _queue.GetConsumingEnumerable())
      await value();
  }
}

//public Task Run(Func<Task> lambda)
public Task Run(<Task> lambda)
{
  var tcs = new TaskCompletionSource<object>();
  _queue.Add(lamda);
  return tcs.Task;
}

1. I need to queue certain DataBase tasks which is within a regular void method. I may not be able to change the signature of this method. How do i do them?

 public static ExecutionQueue taskQueue = new ExecutionQueue();

 private void SaveValesToDB(...)
 {
    var item = GetID(...);
    ...
    taskQueue.Run(Task.Run(() =>
    {
       DBInstance.DBSaveValue1(...); // is it correct to wrap with Task.Run and add to queue? it should be queued and run asynchronously
     });
    ...
 }

2. We save and retrieve data from DB on and off. So, when we queue a DB call that is returning something like a getter, we want to ensure that until we receive the return value we don't process other items that are queued.

private void SaveValesToDB(...)
{
 ...
 taskQueue.Run(Task.Run(() =>
 {
    DBInstance.DBSaveValue1(...); // is this correct? it should be queued and run asynchronously
  });
 ...
 taskQueue.Run(Task.Run(() =>
 {
    var result1 = DBInstance.DBGetValue2(...); // should be queued and run asynchronously; 
    LogData(result1);// not a DB call but believe it should be wrapped in here for the result1, correct?
 });

 /*so in above Task.Run,  i want to ensure that until i receive result1 
 i don't process other items in the queue even 
 if they are added. how can i do that ? 
 The main thread should continue. */
 ...
 var result 2 = DBInstance.DBGetValue3(...); // should be queued and run asynchronously

 UpdateAdvancedLod(result1 +" "+result2);// here, should i block main thread until i get result1 ?
}

3. How to handle errors?

Please, guide me.

Edited:

if using Func<Task> in public Task Run(Func<Task> lambda) then is the below correct?

taskQueue.Run(async () =>
                {
                    await Task.Run(() =>
                    {
                        DBInstance.DBSaveValue1(...);//is this correct
                    });
                }
                );

Answer 1

You could add this method to Stephen Cleary's ExecutionQueue class:

public Task Run(Action action)
{
    return Run(() => Task.Run(action));
}

This is an overload of the existing public Task Run(Func<Task> lambda) method. This one delegates the execution of the supplied action to a ThreadPool thread.

Usage example:

var id = GetID();
var task = taskQueue.Run(() => DBInstance.DBSaveValue1(id));
await task; // Optional

---

Update: To propagate error notifications to the main thread, you could enhance the ExecutionQueue class with an Error event, which would be invoked in the captured context (captured at the time that the instance was created).

private readonly SynchronizationContext _capturedContext;

public event EventHandler<Exception> Error;

public ExecutionQueue() // Constructor
{
    _capturedContext = SynchronizationContext.Current ?? new SynchronizationContext();
    Completion = Task.Run(() => ProcessQueueAsync());
}

private void OnError(Exception ex)
{
    var handler = Error; if (handler == null) return;
    _capturedContext.Post(_ => handler.Invoke(this, ex), null);
}

The OnError should be called from inside the catch (Exception ex) block. This will work with Windows Forms apps and WPF apps, because their UI thread is equipped with a SynchronizationContext. It will not work with a Console app because there is no SynchronizationContext there (the Error event will be raised in a random ThreadPool thread).