我如何“等待”可观测到的Rx？

Question

我希望能够等待一个可观察的时间，例如。

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

天真的尝试会导致等待立即解析，而不会阻止执行

编辑：我的全部预期用例的伪代码是：

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

我知道我可以将其他代码移动到另一个单独的函数中，并将其传递到subscribe回调中，但我希望能够避免这种情况。

Answer 1

很可能是这样

await observable.first().toPromise();

正如在前面的评论中所指出的，当存在空的完成可观测值时，take（1）和first（）运算符之间存在实质性差异。

Observable.empty（）.first（）.toPromise（）将导致拒绝EmptyError，因为实际上没有值，因此可以相应地处理该错误。

和Observable.empty（）.take（1）.toPromise（）将使用未定义的值进行解析。

Answer 2

使用新的firstValueFrom（）或lastValueFrom（）而不是toPromise（），正如这里所指出的，它在RxJS 7中被弃用，并将在RxJS 8中删除。

import { firstValueFrom} from 'rxjs';
import { lastValueFrom } from 'rxjs';

this.myProp = await firstValueFrom(myObservable$);
this.myProp = await lastValueFrom(myObservable$);

这在RxJS 7中可用

见：https://indepth.dev/rxjs-heads-up-topromise-is-being-deprecated/

Answer 3

您必须向wait传递promise。将可观察对象的下一个事件转化为promise，然后等待。

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

编辑说明：此答案最初使用.take（1），但改为使用.first（），这避免了如果流在值通过之前结束，则promise永远不会解决的问题。

从RxJS v8开始，toPromise将被删除。相反，可以用wait firstValueFrom（observable）