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

Question

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

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

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

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

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

Как вы можете использовать реактивные расширения для этого?

Answer 1

Как насчет:

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

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

Answer 2

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

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

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