使用 Redux-observable 处理刷新令牌

Question

我正在使用 React、Redux 和 Redux-Observable 开发客户端应用程序，每次在初始请求后收到 AccessDenied 异常时，我都想发出刷新请求。

问题是我不知道该怎么做......我花了很多时间在这上面。我唯一发现的是这个旧指南，它不起作用（或者我做错了什么）。

因此，如果您有任何建议，请提供帮助。我并不是要求现成的解决方案。也许有人可以指出我需要走的方向。

现在我的代码如下所示：

史诗

export const getDataEpic: Epic<LogoutAction> = (action$, _, { api }) =>
  action$.pipe(
    ofType(GET_DATA),
    exhaustMap(() => api.Request(getDataRequest).pipe(
      map((data) => {
        ...handle data
      }),
    )),
  );

export const refreshEpic: Epic<RefreshAction> = (action$, _, { api }) =>
  action$.pipe(
    ofType(REFRESH),
    exhaustMap(() => api.Request<RefreshResponse>(refreshRequest).pipe(
      mergeMap(({ data }) => {
        const { success, user, accessToken } = data.auth.refresh;

        const actions: Action[] = [success ? refreshSuccess() : refreshFailure()];
        if (!success)
          return from(actions);

        actions.push(setUser(user));

        api.SetAccessToken(accessToken);

        return from(actions);
      }),
    )),
  );

全球史诗ErrorHandler

export const RootEpic: Epic = (action$, store$, dependencies: EpicDependencies) =>
  combineEpics<any, any, any, any>(
    getDataEpic,
    refreshEpic,
    ...
  )(action$, store$, dependencies).pipe(
    catchError((error, source$) => {
      const api = dependencies.api;

      if (api.IsAccessDeniedError(error)) {
        return source$.pipe(
          ofType(REFRESH_SUCCESS),
          takeUntil(action$.pipe(ofType(REFRESH_FAILURE))),
          take(1),
          mergeMap(() => {
            return source$;
          }),
          mergeWith(of(refresh())),
        );
      }

      console.error(error);
      return source$;
    }),
  );

它当前的行为实际上是在数据请求后发送刷新（如果它得到了

accessDenied

）。但它不会再次调度

getData

操作，并且如果不是 getData，则会阻止所有其他操作

我已经尝试了很多方法来解决这个问题，但我开始怀疑这是否是一个好主意。另外，我尝试通过在数据请求之前发送刷新请求、检查令牌是否过期以及使用 Redux 中间件来实现此目的。

Answer 1

我最近确实在一个应用程序中使用拦截器解决了这个问题。好处是，如果应用程序保持打开状态几个小时，您就不会无缘无故地刷新令牌，因为它是在需要时才懒惰地完成的。我还这样做了，如果同时触发 2 个请求，当刷新令牌时，它们不会同时触发刷新，它们将共享一个等待令牌刷新的缓存。

最后，我使用 grpc 而不是 HTTP，但我们构建了一个

小型库，它真正模仿了 http 拦截器接口，因此您应该很容易将其重新适应 HTTP：

export class AuthenticationGrpcInterceptor implements GrpcInterceptor {
  private readonly authenticationFacade = inject(AuthenticationFacade);
  private readonly logger = inject(Logger).source('AuthGrpcInterceptor');
  private readonly oAuthService = inject(OAuthService);
  private readonly invalidatedTokenResponses = new Set<
    AccessTokenResponse['access_token']
  >();
  private readonly datesService = inject(DatesService);

  public intercept(
    requestData: GrpcRequestData,
    next: GrpcHandler
  ): Observable<unknown> {
    if (requestData.method.service.serviceName !== SomeEndpoint.serviceName) {
      return next.handle(requestData);
    }

    this.logger.debug(
      `Intercepted a gRPC request to "${requestData.method.service.serviceName}/${requestData.method.methodName}" on which bearer token is added`
    );

    return this.authenticationFacade.accessTokenResponseResolved$.pipe(
      switchMap((accessTokenResponseResolved) => {
        const accessToken =
          accessTokenResponseResolved?.accessTokenResponse?.access_token;

        if (!accessToken || this.invalidatedTokenResponses.has(accessToken)) {
          // give it some time to refresh the token but let's not accumulate requests
          // indefinitely it could have unwanted effects if a token happens to resolve way later
          return timer(10_000).pipe(
            switchMap(() =>
              throwError(
                () =>
                  new Error(
                    `Couldn't refresh the access token in a reasonable time`
                  )
              )
            )
          );
        }

        return of({
          accessToken: accessToken,
          accessTokenExp:
            accessTokenResponseResolved.parsedAccessToken.exp ?? null,
          refreshToken:
            accessTokenResponseResolved.accessTokenResponse.refresh_token ??
            null,
        });
      }),
      // we've got a non null token, we shouldn't trigger
      // other requests if the token changes after this point
      take(1),
      mergeMap((accessTokenResponseResolved) => {
        const refresh = (refreshToken: string | null) => {
          if (!refreshToken) {
            throw new Error('No refresh token provided');
          }

          this.invalidatedTokenResponses.add(
            accessTokenResponseResolved.accessToken
          );

          return this.oAuthService.refreshToken(refreshToken).pipe(
            tap((newAccessTokenResponse) => {
              this.logger.debug(`Token refreshed successfully`);
              this.authenticationFacade.setAccessTokenResponse(
                newAccessTokenResponse
              );
              this.invalidatedTokenResponses.delete(
                accessTokenResponseResolved.accessToken
              );
              throw new TokenExpired();
            })
          );
        };

        const currentTimeSeconds = Math.floor(
          this.datesService.getNow() / 1000
        );
        const accessTokenExpirySeconds =
          accessTokenResponseResolved.accessTokenExp ?? Infinity;

        const remainingTimeForAccessTokenSeconds =
          accessTokenExpirySeconds - currentTimeSeconds;

        if (remainingTimeForAccessTokenSeconds < 5) {
          this.logger.debug(
            `Token expiring soon or already expired (${remainingTimeForAccessTokenSeconds}), refreshing`
          );

          return refresh(accessTokenResponseResolved.refreshToken);
        }

        requestData.metadata.set(
          'Authorization',
          `Bearer ${accessTokenResponseResolved.accessToken}`
        );

        return next.handle(requestData).pipe(
          catchError((e: { code: GrpcStatus; message: string }) => {
            if (e.code !== GrpcStatus.UNAUTHENTICATED) {
              throw e;
            }

            this.logger.debug(`Invalid token, refreshing`);

            // we've passed a token but unauthenticated came back so
            // at this stage it means we should try to refresh the token
            return refresh(accessTokenResponseResolved.refreshToken);
          })
        );
      }),
      retry({
        delay: (error: unknown, retryCount: number) => {
          if (retryCount > 3) {
            throw new Error(
              `Number of attempts to refresh the token reached the maximum allowed`
            );
          }

          if (error instanceof TokenExpired) {
            this.logger.debug(
              `Retrying request now that the token has been refreshed`
            );
            return of(true);
          }

          return throwError(() => error);
        },
      })
    );
  }
}

class TokenExpired extends Error {}