Как я могу "ждать" на Rx Observable?

Я хотел бы быть в состоянии ждать на наблюдаемом, например,

const source = Rx.Observable.create(/* ... */)
//...
await source;

Наивная попытка приводит к немедленному разрешению и не блокирует выполнение

Изменить: Псевдокод для моей полной цели использования:

if (condition) {
  await observable;
}
// a bunch of other code

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

Ответы

Ответ 1

Вы должны передать обещание await. Преобразуйте наблюдаемое следующее событие в обещание и ожидайте этого.

if (condition) {
  await observable.first().toPromise();
}

Редактировать примечание: этот ответ первоначально использовал .take(1), но был изменен для использования .first(), который позволяет избежать проблемы с обещанием, которое никогда не будет разрешаться, если поток закончится до того, как будет достигнуто значение.

Ответ 2

Это должно быть

await observable.first().toPromise();

Как отмечалось ранее в комментариях, существует существенная разница между операторами take(1) и first(), когда пустые завершенные наблюдаемые.

Observable.empty().first().toPromise() приведет к отказу с EmptyError, который может быть обработан соответствующим образом, что обычно является желательным поведением.

И Observable.empty().take(1).toPromise() приведет к ожидающему обещания, что желательно... почти никогда.

Ответ 3

Вам понадобится await обещание, поэтому вы захотите использовать toPromise(). Подробнее о toPromise() см. .