Как использовать расширение Rx.Nex ForEachAsync с асинхронным действием

Можно ли написать расширение, которое будет принимать асинхронное действие?

Не напрямую.

Подписки Rx обязательно синхронны, потому что Rx — это система, основанная на push-уведомлениях. Когда элемент данных поступает, он проходит через ваш запрос, пока не достигнет окончательной подписки, которая в данном случае должна выполнить Action.

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

Чтобы выполнить синхронно-асинхронный переход в конвейере данных, вам понадобится буфер. Подписка Rx может (синхронно) добавлять в буфер в качестве производителя, в то время как асинхронный потребитель извлекает элементы и обрабатывает их. Итак, вам понадобится очередь производителя/потребителя, которая поддерживает как синхронные, так и асинхронные операции.

Различные типы блоков в потоке данных TPL могут удовлетворить эту потребность. Что-то вроде этого должно быть достаточно:

var obs = StreamDataFromSql().Buffer(BatchSize);
var buffer = new ActionBlock<IList<T>>(batch => WriteDataAsync(batch));
using (var subscription = obs.Subscribe(buffer.AsObserver()))
  await buffer.Completion;

Обратите внимание, что обратного давления нет; как только StreamDataFromSql сможет отправить данные, они будут буферизованы и сохранены во входящей очереди ActionBlock. В зависимости от размера и типа данных это может быстро использовать много памяти.

Для этого правильно использовать Reactive Extensions, поэтому начните с момента создания соединения до тех пор, пока не запишете свои данные.

Вот как:

IObservable<IList<MyData>> query =
    Observable
        .Using(() => new SqlConnection(""), connection =>
            Observable
                .Using(() => new SqlCommand("select * from MyTable", connection), cmd =>
                    Observable
                        .Using(() => cmd.ExecuteReader(), reader =>
                            Observable
                                .While(() => reader.Read(), Observable.Return(GetRow(reader))))))
        .Buffer(BatchSize);

IDisposable subscription =
    query
        .Subscribe(async list => await WriteDataAsync(list));

Я не мог проверить код, но он должен работать. Этот код предполагает, что WriteDataAsync тоже может принимать IList<MyData>. Если он просто не упадет в .ToList().

Вот версия метода ForEachAsync, которая поддерживает асинхронные действия. Он проецирует наблюдаемый источник во вложенный IObservable<IObservable<Unit>>, содержащий асинхронные действия, а затем сглаживает его обратно в IObservable<Unit> с помощью Merge. Полученная наблюдаемая наконец преобразуется в задачу.

По умолчанию действия вызываются последовательно, но их можно вызывать одновременно, настроив необязательный аргумент maximumConcurrency.

Отмена необязательного аргумента cancellationToken приводит к немедленному завершению (отмене) возвращенного Task, возможно, до отмены текущих выполняемых действий.

Любое исключение, которое может возникнуть, распространяется через Task и вызывает отмену всех выполняемых в данный момент действий.

/// <summary>
/// Invokes an asynchronous action for each element in the observable sequence,
/// and returns a 'Task' that represents the completion of the sequence and
/// all the asynchronous actions.
/// </summary>
public static Task ForEachAsync<TSource>(
    this IObservable<TSource> source,
    Func<TSource, CancellationToken, Task> action,
    CancellationToken cancellationToken = default,
    int maximumConcurrency = 1)
{
    // Arguments validation omitted
    return source
        .Select(item => Observable.FromAsync(ct => action(item, ct)))
        .Merge(maximumConcurrency)
        .DefaultIfEmpty()
        .ToTask(cancellationToken);
}

Пример использования:

await StreamDataFromSql()
    .Buffer(BatchSize)
    .ForEachAsync(async (batch, token) => await WriteDataAsync(batch, token));

Вот исходный код ForEachAsync и статья о методах ToEnumerable и AsObservable

Мы можем создать оболочку вокруг ForEachAsync, которая будет ожидать функцию, возвращающую задачу:

public static async Task ForEachAsync<T>( this IObservable<T> t, Func<T, Task> onNext )
{
    foreach ( var x in t.ToEnumerable() )
        await onNext( x );
}

Пример использования:

await ForEachAsync( Observable.Range(0, 10), async x => await Task.FromResult( x ) );