CODEWITHSUNDEEP
c#multithreadingtask-parallel-librarysystem.reactiveproducer-consumer
nesh_s
nesh_s

Reputation: 399

How to continue with other tasks in the queue if current fails?

I have some code that executes a queue like below

Queue
    .GetConsumingEnumerable()
    .ToObservable()
    .Select(x => x.ObserveOn(NewThreadScheduler.Default))
    .SubscribeOn(NewThreadScheduler.Default)
    .Subscribe(
        grp => grp.ForEachAsync(b => b.Execute())
                  .ContinueWith(ExecuteOnTaskFailure, TaskContinuationOptions.OnlyOnFaulted));

the method ExecuteOnTaskFailure is defined as follows

private static void ExecuteOnTaskFailure(Task previousTask)
{
    if (!previousTask.IsFaulted)
        return;

    if (previousTask.Exception != null && previousTask.Exception.InnerExceptions != null)
        foreach (var exception in previousTask.Exception.InnerExceptions)
        {
            Logger.Error("Task failed continued to the next task : " + exception.Message, exception);
        }
}

This doesnt work. I cant seem to figure out HOW to make tasks in my queue continue executing even if one of them fails to execute. Also is there a way for me to requeue this failed task at the end of my queue?

Any help here is much appreciated.

Upvotes: 4

Views: 668

Answers (1)

svick
svick

Reputation: 244908

It sounds like what you want is a try-catch in your loop:

grp.ForEachAsync(async b =>
{
    try
    {
        await b.Execute();
    }
    catch (Exception ex)
    {
        Logger.Error(ex);
        Queue.Add(b);
    }
})

But BlockingCollection is not very async-friendly, because it blocks (as its name suggests) when it's empty. You might want to consider a different approach.

Upvotes: 1

Related Questions