Convert async function to RxJs Observable

Question

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 ?

Roberto Zvjerković · Accepted Answer · 2021-07-27 14:46:15Z

11

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
}

answered Jul 27, 2021 at 14:46

Roberto Zvjerković

10.2k4 gold badges31 silver badges49 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Tot Over a year ago

Damn I just realized it when I received answers. I just needed to call it. Thanks !

Timothy Over a year ago

It's certainly one of the most useful answer. I didn't now that.

Barremian · Accepted Answer · 2021-07-27 14:48:05Z

2

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.

edited Jul 27, 2021 at 14:48

answered Jul 27, 2021 at 14:39

Barremian

31.2k4 gold badges37 silver badges67 bronze badges

4 Comments

Roberto Zvjerković Over a year ago

That's the other way around? OP is looking to return an Observable, not a Promise.

Barremian Over a year ago

@RobertoZvjerković: I went through the question again and am still don't quite understand it. So OP wants to return an observable and use async/await on it?

Tot Over a year ago

I already was using toPromise(), but differently. The issue in your sample is that you don't check is this.dataset is already available. Your solution would work, but the async/await looks way better. Thanks for the proposition though !

Barremian Over a year ago

@Tot: Sorry I overlooked the if check in the question. It looks like you're trying to have a caching mechanism for the value from the observable. You could use the shareReplay(1) with buffer 1 as shown by Poul Kruijt in his answer.

Poul Kruijt · Accepted Answer · 2021-07-27 17:11:51Z

2

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

edited Jul 27, 2021 at 17:11

answered Jul 27, 2021 at 14:46

Poul Kruijt

72.3k12 gold badges152 silver badges152 bronze badges

2 Comments

Tot Over a year ago

I think your solution is the best for my real case. However, it doesn't quite solve the original issue (to use async/await with RxJS's from()). I will accept another answer, but I hope yours will get the most upvotes !

Poul Kruijt Over a year ago

@Tot Yep, it's a typical XY Problem

Mrk Sef · Accepted Answer · 2021-07-27 15:00:16Z

0

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

answered Jul 27, 2021 at 15:00

Mrk Sef

8,0721 gold badge14 silver badges25 bronze badges

3 Comments

Tot Over a year ago

Same answer as above (from @Poul_Kruijt)... oh in fact better as it uses the DataSet type ! Though sometimes I'd like to use await ...toPromise(), which would not benefit the shareReplay(1) operator. I've just tested it and of course it completes the observable. Sometimes observables are very practicle and fluent, sometimes it's a mess compared to async/await. I'll have to see what's best.

Mrk Sef Over a year ago

async/await is syntactic sugar for promises. Since it has support in the language spec it is going to be nicer as some of the complexity is abstracted into the language itself. That being said, I would avoid mixing Observables and Promises except at the boundaries between libraries that use one or the other.

Ben Racicot Over a year ago

We're looking for the answer to the question. If you're willing to go this direction then just next a BehaviorSubject in the promise's .then, it would be simpler.

Collectives™ on Stack Overflow

Convert async function to RxJs Observable

4 Answers 4

2 Comments

4 Comments

2 Comments

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

4 Answers 4

2 Comments

4 Comments

2 Comments

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related