CODEWITHSUNDEEP
angulartypescriptasync-awaitrxjsobservable
Tot
Tot

Reputation: 915

Convert async function to RxJs Observable

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 ?

Upvotes: 3

Views: 10421

Answers (4)

Poul Kruijt
Poul Kruijt

Reputation: 71961

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

Upvotes: 2

Mrk Sef
Mrk Sef

Reputation: 8062

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

Upvotes: 0

Barremian
Barremian

Reputation: 31125

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.

Upvotes: 2

Roberto Zvjerković
Roberto Zvjerković

Reputation: 10157

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
}

Upvotes: 11

Related Questions