Using an async method in a RxJS NextObserver

user_15349 0
Viewed 2954

I'm trying to use an async function in a NestJS interceptor. These interceptors use RxJS Observables like this:

@Injectable()
export class MyInterceptor implements NestInterceptor {
    async intercept<T>(context: ExecutionContext, next: CallHandler): Promise<Observable<T>> {
        await doBegin();
        return next
            .handle()
            .pipe(
                tap(
                    () => console.log("Done"),
                    (e) => console.error(e)
                )
            );
    }
}

This works, but what if I want the methods in tap to be async? The method signature is:

(value: T) => void

Can I just put an async method in there? Or should I take a different approach?

rxjs async-await nestjs
1 Answers

If you want it to be async and its error should be caught you need to use mergeMap or any other suitable operator to handle it inside of the stream context, because tap causes side effects outside of the stream.

const myAsyncFunction = () => {
  // a sample of promise.
  return new Promise(resolve => {
    setTimeout(() => {
      console.log('Promise!');
      resolve();
    }, 1000);
  });
}

@Injectable()
export class MyInterceptor implements NestInterceptor {
    async intercept<T>(context: ExecutionContext, next: CallHandler): Promise<Observable<T>> {
        await doBegin();
        return next
            .handle()
            .pipe(
                mergeMap(value => from(myAsyncFunction()).pipe(
                  ignoreElements(),
                  // catchError(() => EMPTY), // catching all errors.
                  endWith(value),
                )),
                tap(
                  () => {}, // nothing to do here, we need error.
                  (e) => console.error(e), // or catchError if you wan't to handle it.
                ),
            );
    }
}

if you don't care about its errors - simply call .then.

@Injectable()
export class MyInterceptor implements NestInterceptor {
    async intercept<T>(context: ExecutionContext, next: CallHandler): Promise<Observable<T>> {
        await doBegin();
        return next
            .handle()
            .pipe(
                tap(
                  myAsyncFunction, // if it returns `new Promise` - it will work.
                  (e) => console.error(e),
                ),
            );
    }
}
user_131120180
Related