将异步函数转换为 RxJs Observable

Question

防止写一定量的代码，比如这个例子：

    ...
    public getDataSet() : Observable<any>
    {
        return new Observable(observer => {
            if (this.dataset === undefined) {
                this.get().subscribe(data => {
                    this.dataset = new DataSet(data) ;
                    observer.next(this.dataset)
                    observer.complete() ;
                }) ;
            }
            else {
                observer.next(this.dataset)
                observer.complete() ;
            }
        }) ;
    }

我想使用 async/await 功能，但仍返回一个 Observable 以在使用异步数据收集服务时保持一致。

因此，根据 RxJs 的文档，我应该使用 from() 运算符。 https://www.learnrxjs.io/learn-rxjs/operators/creation/from

这是我尝试实现的（看起来更简洁）：

import { Observable, from } from 'rxjs' ;
...
        return from(async () => { // ✘ error
            if (this.dataset === undefined) this.dataset = new DataSet(await this.get().toPromise()) ;
            return this.dataset ;
        }) ;

但是，TypeScript 不能将 async () => {} 识别为 ObservableInput，但它确实识别了这个：

import { Observable, from } from 'rxjs' ;
...
        return from(new Promise((resolve) => { // ✔ valid
            if (this.dataset === undefined) {
                this.get().toPromise().then(data => {
                    this.dataset = new DataSet(data) ;
                    resolve(this.dataset) ;
                }) ;
            }
            else resolve(this.dataset) ;
        })) ;

不过，JavaScript 的 async 关键字让函数总是返回一个 Promise。

console.log((async () => {}) ())
// Promise {}

有没有办法让from() RxJs 操作员接受异步承诺？

Answer 1

异步函数在调用时返回一个承诺，在此之前它只是一个函数。

function fromAsyncFunction(): Observable<number> {
    return from((async () => {
      return 1;
    })()); // Call it
}

Answer 2

你要找的是 RxJS toPromise() 函数。

import { map } from 'rxjs/operators';

public getDataSet(): Promise<any> {
  return this.get().pipe(
    map(data => new DataSet(data))  // <-- transform incoming data from subscription
  ).toPromise();
}

然后你可以等待这个函数。

// in some component

public async getDataSet() {
  this.dataSet = await this.someService.getDataSet();
}

但要注意 toPromise() 是来自 RxJS v7 的 deprecated（在撰写本文时为当前版本），并且会在 RxJS v8 中消失。

Answer 3

您还可以执行以下操作，因为不赞成将 promise 与 observable 混合使用 :)：

public getDataSet() : Observable<any> {
  return this.dataset ? of(this.dataset) : this.get().pipe(
    map((dataset) => {
      this.dataset = new DataSet(data);
      return this.dataset;
    })
  );
}

但是，如果第一个请求尚未完成，上述解决方案可能会触发双重请求。

---

感觉就像你试图让它缓存值。在这种情况下，您可以使用 shareReplay() 运算符执行以下操作。这样，get 请求在第一个subscribe 处触发，之后返回响应而不触发请求。它为您省去了使用中介属性的麻烦：

private dataSet$ = this.get().pipe(
  map((data) => new DataSet(data)),
  shareReplay(1)
);

public getDataSet() : Observable<any> {
  return this.dataset$;
}

Answer 4

为什么不单独定义您的可观察对象，并在需要时返回它？ shareReplay 充当数据集的缓存，因此您根本不需要声明 this.dataset。

dataset$ = get().pipe(
  map(data => new DataSet(data)),
  shareReplay(1)
);

public getDataSet() : Observable<any> {
  return this.dataset$.pipe(
    take(1)
  );
}