如何在Rx Observable上"等待"？

Question

我希望能够等待一个可观察的,例如

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

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

编辑:我的完整用途的伪代码是:

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

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

Answer 1

你必须通过承诺await.将observable的下一个事件转换为promise并等待它.

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

编辑注释:这个答案最初使用.take(1)但更改为使用.first(),这避免了如果流在值到来之前结束,Promise永远不会解决的问题.

Answer 2

它一定要是

await observable.first().toPromise();

正如之前的评论中所指出的那样,当存在空的完整观察值时take(1),first()运算符与运算符之间存在很大差异.

Observable.empty().first().toPromise()将导致拒绝,EmptyError因此可以相应地处理,这通常是理想的行为.

并且Observable.empty().take(1).toPromise()将导致未决的承诺,这是可取的......几乎从不.

Answer 3

使用新的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 4

如果toPromise不推荐使用，您可以使用，.pipe(take(1)).toPromise但正如您在此处看到的那样，它没有被弃用。

所以请使用toPromise(RxJs 6) 说：

//return basic observable
const sample = val => Rx.Observable.of(val).delay(5000);
//convert basic observable to promise
const example = sample('First Example')
  .toPromise()
  //output: 'First Example'
  .then(result => {
    console.log('From Promise:', result);
  });

异步/等待示例：

//return basic observable
const sample = val => Rx.Observable.of(val).delay(5000);
//convert basic observable to promise
const example = await sample('First Example').toPromise()
// output: 'First Example'
console.log('From Promise:', result);

在这里阅读更多。

并请删除这个错误的声明说toPromise已弃用。

Answer 5

你需要await一个承诺,所以你会想要使用toPromise().见这对更多的细节toPromise().

Answer 6

toPromise()不建议使用，因为它从 RxJs 7 开始逐渐贬值。您可以使用 RxJs 7lastValueFrom()和中存在的两个新运算符firstValueFrom()。更多详情可在这找到

const result = await lastValueFrom(myObservable$);

Beta 版本的实现可在此处获取：

• 第一个值来自
• 最后值来自