[解決済み] Rx Observableで`await`するにはどうしたらいいですか？

質問

observableでawaitできるようにしたい。

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

素朴な試みでは、await は直ちに解決され、実行はブロックされません。

編集する 私の意図する完全なユースケースの疑似コードは、以下の通りです。

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

他のコードを別の別の関数に移動して、それをsubscribeコールバックに渡すことができることは理解していますが、それを避けることができればと思っています。

解決方法は？

にプロミスを渡す必要があります。 await . observableの次のイベントをプロミスに変換して、それを待ちます。

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

編集注：この回答はもともと .take(1) を使っていましたが、値が来る前にストリームが終了した場合に Promise が解決されないという問題を回避するために .first() を使うように変更されました。

RxJS v8時点では。 toPromise は削除されます。その代わりに、上記は以下のように置き換えることができます。 await firstValueFrom(observable)