在 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 中的方法是异步的怎么办？ method signature 是：

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