node.js - 如何将 AsyncLocalStorage 用于 Observable？

Question

我想使用 AsyncLocalStorage在 NestJs Interceptor :

export interface CallHandler<T = any> {
    handle(): Observable<T>;
}
export interface NestInterceptor<T = any, R = any> {
    intercept(context: ExecutionContext, next: CallHandler<T>): Observable<R> | Promise<Observable<R>>;
}

拦截器函数得到一个 next CallHandler返回 Observable .
我无法使用 run在这种情况下(运行回调将在 callHandler.handle() observable 完成之前立即退出):

  intercept(context: ExecutionContext, callHandler: CallHandler): Observable<any> | Promise<Observable<any>> {
    const asyncLocalStorage = new AsyncLocalStorage();
    const myStore = {  some: 'data'};
    return asyncLocalStorage.run(myStore, () => callHandler.handle());
  }

见 broken replit-example
我想出的解决方案是这样的:

const localStorage = new AsyncLocalStorage();

export class MyInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, callHandler: CallHandler): Observable<any> | Promise<Observable<any>> {
    const resource = new AsyncResource('AsyncLocalStorage', { requireManualDestroy: true });
    const myStore = { some: 'data' };

    localStorage.enterWith(myStore);
    return callHandler.handle().pipe(
      finalize(() => resource.emitDestroy())
    );
  }
}

见 working replit example
这似乎工作正常，但我不确定这是否真的正确 - 它看起来凌乱且容易出错。所以我想知道:

这是正确的吗？

有没有更好/更清洁的方法来处理这个问题？

Answer 1

我的使用方式 cls-hooks ，我找到的解决方案是:

return new Observable(observer => {
  namespace.runAndReturn(async () => {
    namespace.set("some", "data")   
    next.handle()
        .subscribe(
          res => observer.next(res), 
          error => observer.error(error),
          () => observer.complete()
        )
  })
})