Использование Polly с потоком данных TPL

Конвейеры обработки данных и обработка временных сбоев, кажется, идут рука об руку, поэтому мне интересно посмотреть, смогу ли я получить 2 лучшие библиотеки для них — TPL Dataflow и Polly, соответственно - для приятной игры вместе.

В качестве отправной точки я хотел бы применить политику обработки ошибок к ActionBlock. В идеале я хотел бы инкапсулировать его в методе создания блоков с такой подписью:

ITargetBlock<T> CreatePollyBlock<T>(
    Action<T> act, ExecutionDataflowBlockOptions opts, Polly.Policy policy)

Было бы достаточно просто policy.Execute выполнить действие внутри ActionBlock, но у меня есть два требования:

1. В случае повторной попытки я не хочу, чтобы повторная попытка элемента имела приоритет над другими элементами в очереди. Другими словами, когда вы терпите неудачу, вы идете в конец очереди.
2. Что еще более важно, если перед повторной попыткой есть период ожидания, я не хочу, чтобы это блокировало получение новых элементов. И если установлено ExecutionDataflowBlockOptions.MaxDegreeOfParallelism, я не хочу, чтобы элементы, ожидающие повторной попытки, «учитывались» в этом макс.

Чтобы удовлетворить эти требования, я полагаю, мне нужен «внутренний» ActionBlock с примененным пользователем ExecutionDataflowBlockOptions и некоторый «внешний» блок, который отправляет элементы во внутренний блок и применяет любую логику ожидания и повторной попытки (или что-то другое, что диктует политика). ) вне контекста внутреннего блока. Вот моя первая попытка:

// wrapper that provides a data item with mechanism to await completion
public class WorkItem<T>
{
    private readonly TaskCompletionSource<byte> _tcs = new TaskCompletionSource<byte>();

    public T Data { get; set; }
    public Task Completion => _tcs.Task;

    public void SetCompleted() => _tcs.SetResult(0);
    public void SetFailed(Exception ex) => _tcs.SetException(ex);
}

ITargetBlock<T> CreatePollyBlock<T>(Action<T> act, Policy policy, ExecutionDataflowBlockOptions opts) {
    // create a block that marks WorkItems completed, and allows
    // items to fault without faulting the entire block.
    var innerBlock = new ActionBlock<WorkItem<T>>(wi => {
        try {
            act(wi.Data);
            wi.SetCompleted();
        }
        catch (Exception ex) {
            wi.SetFailed(ex);
        }
    }, opts);

    return new ActionBlock<T>(async x => {
        await policy.ExecuteAsync(async () => {
            var workItem = new WorkItem<T> { Data = x };
            await innerBlock.SendAsync(workItem);
            await workItem.Completion;
        });
    });
}

Чтобы проверить это, я создал блок с политикой ожидания и повторения и фиктивный метод, который выдает исключение при первых трех вызовах (для всего приложения). Затем я передал ему некоторые данные:

"a", "b", "c", "d", "e", "f"

"a", "a", "a", "a", "b", "c", "d", "e", "f"

return new ActionBlock<T>(async x => {
    await policy.ExecuteAsync(async () => {
        var workItem = new WorkItem<T> { Data = x };
        await innerBlock.SendAsync(workItem);
        await workItem.Completion;
    });
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 100 });

"a", "e", "b", "c", "d", "a", "e", "b"

"a", "b", "c", "d", "e", "a", "b", "c"