在 RxJS NextObserver 中使用异步方法

Question

我正在尝试在 NestJS 拦截器中使用异步函数。这些拦截器使用 RxJS Observables，如下所示：

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

这可行，但是如果我希望方法是tap异步的怎么办？方法签名是：

(value: T) => void

我可以在那里放置一个异步方法吗？或者我应该采取不同的方法？

Answer 1

如果您希望它是异步的并且应该捕获它的错误，您需要使用mergeMap或任何其他合适的运算符来在流上下文内部处理它，因为tap会在流外部产生副作用。

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

如果您不关心它的错误 - 只需致电.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),
                ),
            );
    }
}