Debouncing and cancelling with redux-observable

问题:

I am trying to create a simple redux-observable epic which debounces and is cancelable. My code:

export const apiValidate = action$ => {
    return action$.ofType(validateRequestAction)
        .debounceTime(250)
        .switchMap((action) => (
            Observable.ajax({
                url: url,
                method: 'GET',
                crossDomain: true,
                headers: {
                    "Content-Type": 'application/json'
                },
                responseType: 'json'
            })
           .map(payload => (new APISuccessValidate()))
           .takeUntil(action$.ofType(validateCancelAction))
           .catch(payload => ([new APIFailureValidate()]))
    ));
};

The code only works sometimes. Depending on the speed of the response from the server, I believe 1 of 2 scenarios can occur.

Scenario 1 (works):

Time 0ms   - Fire validateRequestAction
Time 250ms - Ajax request occurs
Time 251ms - Fire validateCancelAction
Time 501ms - validateCancelAction passes debounce and cancels properly
Nothing else occurs

Scenario 2 (broken)

Time 0ms   - Fire validateRequestAction
Time 250ms - Ajax request occurs
Time 251ms - Fire validateCancelAction
Time 400ms - Ajax returns, APISuccessValidate action fired
Time 501ms - validateCancelAction passes debounce and there is nothing to cancel

Is there a way I can write my epic such that only the validateCancelAction can bypass the debounceTime and cancel the ajax call without waiting?

Thanks!

回答1:

You're actually only debouncing your matching of validateRequestAction, but your .takeUntil(action$.ofType(validateCancelAction)) does not have any debouncing. I may be wrong, but if it's possible for the cancel action to be dispatched before the action has made it past the debounce, then the action it was meant to cancel will not be cancelled because the ajax request hasn't even started yet, nor the takeUntil. This race can be avoided by not allowing a cancellation until your side effect (ajax in this case) has actually started and the takeUntil is listening for the possible cancellation.

In your UI you would not give the user the ability to cancel until some state in redux is set. Since our epic needs to tell redux when to flip that, we'll need to emit an action that we will listen for in the reducers.

The easiest way is use the startWith operator:

export const apiValidate = action$ => {
    return action$.ofType(validateRequestAction)
        .debounceTime(250)
        .switchMap((action) => (
            Observable.ajax({
                url: url,
                method: 'GET',
                crossDomain: true,
                headers: {
                    "Content-Type": 'application/json'
                },
                responseType: 'json'
            })
          .map(payload => (new APISuccessValidate()))
          .takeUntil(action$.ofType(validateCancelAction))
          .catch(payload => ([new APIFailureValidate()]))
          .startWith({ type: 'validateRequestActionStarted' }) // <-- here
    ));
};

So in this example, some reducer would listen for validateRequestActionStarted and change some state that the UI will then know we should give them the ability to cancel.

---

A totally different way of preventing that race--but one I wouldn't recommend in most cases--would be to takeUntil on the top-level stream entirely and then just "restart" the epic using repeat if it gets cancelled. So this would shut down everything when we cancel; any pending ajaxs and any pending debounces.

export const apiValidate = action$ => {
    return action$.ofType(validateRequestAction)
        .debounceTime(250)
        .switchMap((action) => (
            Observable.ajax({
                url: url,
                method: 'GET',
                crossDomain: true,
                headers: {
                    "Content-Type": 'application/json'
                },
                responseType: 'json'
            })
          .map(payload => (new APISuccessValidate()))
          .catch(payload => ([new APIFailureValidate()]))
        ))
        .takeUntil(action$.ofType(validateCancelAction))
        .repeat();
};

---

It's worth noting that I used the terms epic and restart to help conceptualize our specific domain, but this is mostly just normal RxJS so it's generally applicable outside of redux-observable. An "epic" is just a word for our pattern of a function which takes a stream of actions (input) and returns a stream of actions (output).

回答2:

I assume that there're two scenarios that you may want it to be:

Scenario 1:

You want to cancel the throttle immediately when cancel action is received. This means that you may want to reset the second stream. It is good but may be not what you want.

action$ => {
  const requestAction$ = action$.pipe(
    ofType(validateRequestAction),
    share()
  )
  return merge(
    action$.pipe(
      ofType(validateCancelAction),
      mapTo(false)
    ),
    requestAction$.pipe(mapTo(true))
  ).pipe(
    distinctUntilChanged(),
    switchMap(
      condition => 
        condition ? 
          requestAction$.pipe(
            debounceTime(250),
            switchMap(query => sendRequest(query)
          ) : EMPTY
    )
  )

Scenario 2:

You send a cancel signal and at the same time, tell every pending requests that: "Hey, you are not allow to dispatch". There're two way to do this:

• The first, throttle the cancel action with the same latency with the request action so that it race against request action stream.

Code:

merge(
  action$.pipe(
    ofType(validateCancelAction),
    debounceTime(250),
    mapTo(undefined)
  ),
  action$.pipe(
    ofType(validateRequestAction),
    debounceTime(250),
    pluck('payload')
  )
).pipe(
  switchMap(
    query => query ? sendRequest(query) : of({ type: validateCancelDone })
  )
)

• The second and the correct solution is, when a cancel action is dispatched, set the state to being cancelled. Every throttled actions have to check this condition before it is allowed to make any request:

Actually, this is just whether you want to store the cancelled state inside your stream or inside redux. I bet you choose the first one. Code:

export default action$ => 
  combineLatest(
    action$.pipe(
      ofType(validateRequestAction),
      debounceTime(250),
      pluck('payload')
    ),
    merge(
      action$.pipe(
        ofType(validateCancelAction),
        mapTo(false)
      ),
      action$.pipe(
        ofType(validateRequestAction),
        mapTo(true)
      )
    ).pipe(
      distinctUntilChanged()
    )
  ).pipe(
    switchMap(
      ([query, allow]) =>
        allow
          ? sendRequest(query)
          : EMPTY
    )
  )

Edit:

You also need to distinctUntilChanged() the allow stream or debounceTime will take no effect.