How to rewrite this code using rxjs observables instead async/await?

Question

I need that createUser function returns Observable<UserEntity> but in this function I also have to make 2 queries to DB and check if this user exists. The code below uses async/await and looks pretty good and clean. But the problem is that I use rxjs everywhere in this project and would like to write it somehow using rxjs. Can it be as clean as now but with Observables?

async create(user: CreateUserDTO): Promise<UserEntity> {
    const userByEmail = await this.getUserByEmail(); 

    const userByLogin = await this.getUserByLogin(); 
    if (userByLogin || userByEmail)
        // retrun error here

    return await this.createUser(user);
}

You should probably rewrite getUserByEmail, getUserByLogin and createUser functions to Observables too, but it's hard to help with that unless we have the code for them. — ShamPooSham
– ShamPooSham, Commented Mar 23, 2022 at 20:07

Saptarsi · Accepted Answer · 2022-04-13 12:15:27Z

I am using RxJs 6.5

forkJoin will emit result when both async functions getUserByEmail & getUserByLogin complete their execution

If getUserByEmail & getUserByLogin returns Promise,for that using from to convert a promise into an observable

mergeMap to subscribe the inner observable.In our case createUser returns observable

create(user: CreateUserDTO): Observable < UserEntity > {

    //If getUserByEmail & getUserByLogin returs Promise

    const getUserByEmail$ = from(this.getUserByEmail());
    const getUserByLogin$ = from(this.getUserByLogin());

    //If Both returns Observable
    //const getUserByEmail$ = this.getUserByEmail();
    //const getUserByLogin$ = this.getUserByLogin();


    return forkJoin({
        userByEmail: this.getUserByEmail(),
        userByLogin: this.getUserByLogin(),
    }).pipe(
        tap((res) => {
            if (res.userByEmail || res.userByLogin) {
                throw 'User exists!';
            }
        }),
        mergeMap(() => {
            return from(this.createUser(user));
            //If createUser returns Observable,then
            //return this.createUser(user);
        })
    );
}

Picci · Accepted Answer · 2022-03-23 20:00:23Z

2

Assuming that this.getUserByEmail(), this.getUserByLogin() and this.createUser(user) return Promises, the code could look like this

create(user: CreateUserDTO): Observable<UserEntity> {
    // with the rxjs from function we turn a Promise into an Observable
    const userByEmail$ = from(this.getUserByEmail()); 
    const userByLogin$ = from(this.getUserByLogin()); 

    // with forkjoin we create an Observable which notifies when all the 
    // Observables which have been passed in as parameters notify
    return forkJoin([userByEmail$, userByLogin$]).pipe(
       // with concatMap you wait for the upstream Observable (i.e. the 
       // Observable created by forkJoin) to notify and complete, and then
       // you return the next Observable in the chain, which is, in this case,
       // the Observable which (when subscribed) creates the user
       concatMap(([userByLogin, userByEmail]) => 
          if (userByLogin || userByEmail) {
              // throw error here
          }
          return from(this.createUser(user))
       })
    )
}

Otherwise, if this.getUserByEmail(), this.getUserByLogin() and this.createUser(user) return Observables you do not need to use the from rxjs function and the code would be slightly simpler, like this

create(user: CreateUserDTO): Observable<UserEntity> {
    return forkJoin([this.getUserByEmail(), this.getUserByLogin()]).pipe(
       concatMap(([userByLogin, userByEmail]) => 
          if (userByLogin || userByEmail) {
              // throw error here
          }
          return from(this.createUser(user))
       })
    )
}

answered Mar 23, 2022 at 20:00

Picci

17.8k16 gold badges80 silver badges125 bronze badges

3 Comments

Rebai Ahmed Over a year ago

Why you used concatMap operator in this case?

Picci Over a year ago

concatMap is the go-to operator any time you have standard sequential async calls to be made. When I say "standard" I mean that you do not have the requirement to kill the request on fly if a new notification comes from upstream (e.g. if you run a query when you click a button and you want to kill that query, if still on fly, if the button is clicked again - in this case the operator to use is switchMap). What may dangerous is mergeMap which keeps all the request and can end up returning not the freshest result. Anyways, it depends on the requirements.

Picci Over a year ago

You may find interesting this article which talks about typical patterns of use of rxjs operators with http.

Collectives™ on Stack Overflow

How to rewrite this code using rxjs observables instead async/await?

2 Answers 2

Comments

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related