使用 redux-observable 去抖动和取消

Question

我正在尝试创建一个简单的 redux-observable 史诗，它可以去抖并且可以取消。我的代码：

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()]))
    ));
};

该代码仅在某些时候有效。根据服务器的响应速度，我相信可能会发生 2 种情况中的 1 种。

场景 1（有效）：

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

场景 2（损坏）

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

有没有一种方法可以编写我的史诗，使得只有 validateCancelAction 可以绕过 debounceTime 并取消 ajax 调用而无需等待？

谢谢！

Answer 1

您实际上只是对 validateRequestAction 的匹配进行去抖动，但您的 .takeUntil(action$.ofType(validateCancelAction)) 没有任何去抖动。我可能是错的，但如果取消动作有可能在之前被调度，动作已经通过去抖动，那么它打算取消的动作将未被取消，因为 ajax 请求尚未开始，takeUntil 也未开始。在您的副作用（ajax 在这种情况下）实际开始并且 takeUntil 正在监听可能的取消之前不允许取消，可以避免这场比赛。

在你的 UI 中，你不会给用户取消的能力，直到 redux 中的某些状态被设置。由于我们的史诗需要告诉 redux 何时翻转它，我们需要发出一个我们将在 reducer 中监听的动作。

最简单的方法是使用 startWith 运算符：

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
    ));
};

所以在这个例子中，一些 reducer 会监听 validateRequestActionStarted 并改变一些状态，然后 UI 就会知道我们应该给他们取消的能力。

一种完全不同的防止竞争的方法——但在大多数情况下我不推荐这种方法——是完全在 top-level 流上 takeUntil 然后只是 "restart" 史诗使用 repeat 如果它被取消。因此，当我们取消时，这将关闭所有内容；任何待处理的 ajax 和任何待处理的去抖动。

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();
};

值得注意的是，我使用术语 epic 和 restart 来帮助概念化我们的特定领域，但这主要是普通的 RxJS，因此它通常适用于 redux-observable 之外。 "epic" 只是我们函数模式的一个词，它采用一系列动作（输入）和 returns 一系列动作（输出）。

Answer 2

我假设您可能希望有两种情况：

场景一：

您想在收到取消操作时立即取消油门。这意味着您可能想要重置第二个流。很好，但可能不是你想要的。

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
    )
  )

场景二：

您发送一个取消信号，同时告诉每个待处理的请求："Hey, you are not allow to dispatch"。有两种方法可以做到这一点：

首先，以与请求操作相同的延迟限制取消操作，使其与请求操作流竞争。

代码：

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 })
  )
)

第二个也是正确的解决方案是，在调度取消操作时，将状态设置为正在取消。在允许发出任何请求之前，每个受限操作都必须检查此条件：

实际上，这只是你想将取消状态存储在你的流中还是在 redux 中。我打赌你会选择第一个。代码：

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
    )
  )

编辑：

您还需要 distinctUntilChanged() allow 流，否则 debounceTime 将无效。

使用 redux-observable 去抖动和取消

Debouncing and cancelling with redux-observable

rxjs

redux-observable