RxJs：延迟的机制分享（关于令牌刷新示例）

Question

我有以下代码片段：

@Injectable()
class RefreshTokenInterceptor implements HttpInterceptor {
  private readonly renewTokens$: Observable<string | undefined> = defer(() =>
    this.tokens.accessToken$.pipe(
      take(1),
      switchMap((accessToken: string | undefined) =>
        accessToken
          ? this.auth.getAccessTokenSilently()
          : this.auth.loginWithRedirect().pipe(switchMap(() => this.auth.getAccessTokenSilently()))
      ),
      tap((accessToken: string) => this.tokens.renewAccessToken(accessToken)),
      share()
    )
  );

  constructor(private readonly auth: AuthService, private readonly tokens: TokenService) {}

  intercept(request: HttpRequest<unknown>, next: HttpHandler): Observable<HttpEvent<unknown>> {
    return next.handle(request).pipe(
      catchError((error: unknown) =>
        error instanceof HttpErrorResponse && error.status === 401
          ? this.renewTokens$.pipe(
              switchMap(
                (accessToken: string | undefined) => next.handle(withBearerTokenInAuth(request, accessToken))
              )
            )
          : throwError(() => error))
    );
  }
}

我好奇的部分是可观察的 renewTokens$ 及其工作方式。 让我们想象一下这样的情况：我们的 accessToken 已过期，并且我们执行了 3 个需要授权的并行请求。 上面代码片段的网络选项卡中的结果如下：

接下来发生的事情：

1. 所有 3 个请求均失败，出现 401 刷新
2. accessToken 的请求已被触发
3. 使用新令牌重试所有 3 个先前失败的请求，结果,成功了。

我不太清楚为什么我们只有一个令牌请求（尽管这是一种期望的行为）。经过一些调试后，我发现传递给 defer 的工厂函数被调用了 3 次（因为 3 个请求），据我了解，每次返回新的共享可观察值时。为什么所有请求都等待同一个可观察对象发出值？我认为每个请求都应该接收从 defer 的工厂函数返回的单独的可观察值（意味着每个未经授权的请求应该触发单独的令牌请求）

UPD

的实现>TokenService：

@Injectable({
  providedIn: 'root'
})
export class TokenService {
  private readonly accessTokenSubj: BehaviorSubject<string | undefined> =
    new BehaviorSubject<string | undefined>(undefined);

  constructor(private readonly auth: AuthService) {
    auth.isAuthenticated$.pipe(
      switchMap(
        (isAuthenticated: boolean) => isAuthenticated ? auth.getAccessTokenSilently() : of(undefined)
      )
    ).subscribe((accessToken: string | undefined) => this.accessTokenSubj.next(accessToken))
  }

  renewAccessToken(accessToken: string) {
    this.accessTokenSubj.next(accessToken);
  }

  get accessToken$(): Observable<string | undefined> {
    return this.accessTokenSubj.asObservable();
  }
}

AuthService由@auth0/auth0-Angular

Answer 1

您需要将 share() 放在 defer() 之外。

@Injectable()
class RefreshTokenInterceptor implements HttpInterceptor {
  private readonly renewTokens$: Observable<string | undefined> = defer(
    () => this.tokens.accessToken$.pipe(
      /* ... */
    )
  ).pipe(share())
}

从 defer(fn) 获取的 observable 始终在订阅时调用 fn，因此，虽然从技术上讲，您获取的是共享 observable，但每次都会获取不同的 observable。

此外，我猜您会希望获得订阅时最后发出的令牌，在这种情况下，您可以将 share() 替换为 shareReplay(1)。