Convert async function to RxJs Observable

user_2204717 0
Viewed 2421

To prevent writing a certain amount of code, such as this example:

    ...
    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() ;
            }
        }) ;
    }

I would like to use async/await feature but still return an Observable to remain consistent in using asynchronous data gathering services.

Thus according to RxJs's documentation, I should be using from() operator. https://www.learnrxjs.io/learn-rxjs/operators/creation/from

Here is what I try to implement (which looks much more concise) :

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

However, TypeScript doesn't recognize async () => {} as an ObservableInput, but it does recognize this one :

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) ;
        })) ;

Though, JavaScript's async keyword makes a function always return a Promise.

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

Is there a way to make the from() RxJs operator accepting async promises ?

observable rxjs async-await angular typescript
4 Answers

Async function returns a promise when called, before that it's just a function.

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

What you're looking for is RxJS toPromise() function.

import { map } from 'rxjs/operators';

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

You could then await this function.

// in some component

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

But beware toPromise() is deprecated from RxJS v7 (current as of this writing) and would be gone in RxJS v8.

user_65139210

You can also do the following, because mixing promise with observable is frowned upon :):

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

However, this above solution might trigger double requests, if the first one hasn't finished yet.

It feels like you are trying to get it to cache the value. In that case you can do the following with the shareReplay() operator. This way the get request is triggered at the first subscribe, and after that the response is returned without triggering a request. It saves you the hassle of using an intermediary property:

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

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

Why not just define your observable separately, and return it when you need it? shareReplay acts a cache for your dataset, so you shouldn't need to declare this.dataset at all.

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

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