rxjs - How to retry after catching and processing

问题:

I'm using rxjs v5.4.3, redux-observable v0.16.0.

in my application, I'd like to achieve below:

• an user has auth token, and refresh token to regenerate auth token.
• the user requests with auth token. (by emitting REQUEST action)
• if it failed, request regenerating auth token with refresh token.
  • if refreshed, emit TOKEN_REFRESHED action to update auth token, and do not emit REQUEST_FAILURE.
  • if refreshing failed, emit REQUEST_FAILURE
• after refreshing(and updating auth token reducer), retry requesting using the refreshed auth token.
  • if request succeeded, emit REQUEST_SUCCESS, and if failed, emit REQUEST_FAILURE.

I'd like to achieve like:

const fetchEpic = (action$: ActionsObservable<Action>, store: Store<IRootState>) => action$
  .ofAction(actions.fetchPost)
  .mergeMap(({ payload: { postId } })) => {
    const { authToken, refreshToken } = store.getState().auth;

    return api.fetchPost({ postId, authToken }) // this returns Observable<ResponseJSON>
      .map(res => actions.fetchSuccess({ res })) // if success, just emit success-action with the response
      .catch(err => {
        if (isAuthTokenExpiredError(err) {
          return api.reAuthenticate({ refreshToken })
            .map(res => actions.refreshTokenSuccess({ authToken: res.authToken });
            .catch(actions.fetchFailure({ err }))
            // and retry fetchPost after re-authenticate!
        }

        return Observable.of(actions.fetchFailure({ err }))
      })
  }

is there any solution?

回答1:

There are many ways to do it, but I would recommend splitting off the reauthentication into its own epic to make it easier to maintain/test/reuse.

Here's what that might look like:

const reAuthenticateEpic = (action$, store) =>
  action$.ofType(actions.reAuthenticate)
    .switchMap(() => {
      const { refreshToken } = store.getState().auth;

      return api.reAuthenticate({ refreshToken })
        .map(res => actions.refreshTokenSuccess({ authToken: res.authToken }))
        .catch(err => Observable.of(
          actions.refreshTokenFailure({ err })
        ));
    });

We'll also want to use something like Observable.defer so that each time we retry, we look up the latest version of the authToken:

Observable.defer(() => {
  const { authToken } = store.getState().auth;
    return api.fetchPost({ postId, authToken });
})

When we catch errors in fetchEpic and detect isAuthTokenExpiredError we return an Observable chain that:

1. Starts listening for a single refreshTokenSuccess, signalling we can retry
2. Just in case the reauthing itself fails, we listen for it with .takeUntil(action$.ofType(refreshTokenFailure)) so that we aren't waiting around forever--you might want to handle this case differently, your call.
3. mergeMap it to the original source, which is the second argument of the catch callback. The "source" is the Observable chain before the catch, and since Observables are lazy, when we receive the refreshTokenSuccess action it it will resubscribe to that chain again, effectively be a "retrying"
4. Merge the above chain with an Observable of an reAuthenticate action. This is used to kick off the actual reauth.

To summarize: the Observable chain we return from catch will first starting listening for refreshTokenSuccess, then it emits reAuthenticate, then when (and if) we receive refreshTokenSuccess we will then "retry" the source, our api.fetchPost() chain above the catch that we wrapped in an Observable.defer. If refreshTokenFailure is emitted before we receive our refreshTokenSuccess, we give up entirely.

const fetchEpic = (action$, store) =>
  action$.ofType(actions.fetchPost)
    .mergeMap(({ payload: { postId } })) =>
      Observable.defer(() => {
        const { authToken } = store.getState().auth;
        return api.fetchPost({ postId, authToken });
      })
        .map(res => actions.fetchSuccess({ res }))
        .catch((err, source) => {
          if (isAuthTokenExpiredError(err)) {
            // Start listening for refreshTokenSuccess, then kick off the reauth
            return action$.ofType(actions.refreshTokenSuccess)
              .takeUntil(action$.ofType(refreshTokenFailure))
              .take(1)
              .mergeMapTo(source) // same as .mergeMap(() => source)
              .merge(
                Observable.of(action.reAuthenticate())
              );
          } else {
            return Observable.of(actions.fetchFailure({ err }));
          }
        });
    );

These examples are untested, so I may have some minor issues but you hopefully get the gist. There's also probably a more elegant way to do this, but this will at least unblock you. (Others are more than welcome to edit this answer if they can decrease the complexity)

---

Side notes

• This creates the slight potential for infinite retries, which can cause nasty issues both in the person's browser or your servers. It might be a good idea to only retry a set number of times, and/or put some sort of delay in between your retries. In practice this might not be worth worrying about, you'll know best.

• You (or someone else reading this later) may be tempted to use .startWith(action.reAuthenticate()) instead of the merge, but be mindful that a startWith is just shorthand for a concat, not a merge, which means it would synchronously emit the action before we have started to listen for a success one. Usually that isn't an issue since http requests are async, but it's caused people bugs before.