Реактивные расширения Подписаться на вызов ждут

Я хочу выполнить асинхронный вызов, основанный на каждом событии, поднятом Reactive Extensions Observable. Я также пытаюсь синхронизировать все, так как я хочу, чтобы асинхронный вызов завершился до того, как будет обработано следующее событие.

Как можно было бы сделать что-то похожее на следующее? Я говорю так же, как приведенный ниже код не компилируется.

settingsChangedInMemory
    .Subscribe(async _ => {
        var settings = Extract();
        await SaveSettings(settings);
    });

Я не уверен, что это что-то изменит, но мне нужно будет подписать несколько наблюдений. Например, другая подписка вроде этого.

settingsChangedOnDisk
    .Subscribe(async _ => {
        var settings = await ReadSettings(settings);
        Apply(settings);
    });

Как вы могли бы использовать Reactive Extensions?

Ответы

Ответ 1

Как насчет:

settingsChangedInMemory
    .SelectMany(async _ => await SaveSettings(Extract()))
    .Subscribe(x => Apply(x));

Никогда не помещайте async в Subscribe, вы всегда хотите поместить его в SelectMany.

Ответ 2

Вы можете использовать новый метод ForEachAsync, выпущенный в Reactive Extensions (Rx) 2.0, например:

await observable
    .ForEachAsync(async x =>
    {
        Console.WriteLine(x);
        await Task.Delay(1000);
    });

ForEachAsync возвращает a Task<T>, который завершается, когда наблюдаемое завершается. Больше информации в моем сообщении в блоге здесь или этот сообщение в блоге командой реактивных расширений.