How can I `await` on an Rx Observable?

Question

I'd like to be able to await on an observable, e.g.

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

A naive attempt results in the await resolving immediately and not blocking execution

Edit: The pseudocode for my full intended use case is:

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

I understand that I can move the other code into another separate function and pass it into the subscribe callback, but I'm hoping to be able to avoid that.

Answer 1

You have to pass a promise to await. Convert the observable's next event to a promise and await that.

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

Edit note: This answer originally used .take(1) but was changed to use .first() which avoids the issue of the Promise never resolving if the stream ends before a value comes through.

As of RxJS v8, toPromise will be removed. Instead, the above can be replaced with await firstValueFrom(observable)