使用 RxJS switchMap 仅取消订阅具有相同请求 URL/action 负载 (redux-observable epics) 的流

Question

我有一个界面，用户可以在其中触发对同一端点但具有不同参数（在本例中为 UUID）的调用。到目前为止，我一直享受 switchMap 每当我发送相同类型的新 redux 操作时取消我的 in-flight http 请求的行为，在这种情况下我仍然想要这种行为，但是仅如果新操作请求的 UUID（操作的一部分 object）与已经在进行的操作相同。我不太确定正确的方法。

例如，在一次分派多个动作后，我希望所有具有唯一 ID 的动作都完成，但那些重复现有但尚未完成的 ID 会取消之前的请求并取而代之。

例如：

store.dispatch({ type: "GET_SOME_DATA", uuid: "1" })
store.dispatch({ type: "GET_SOME_DATA", uuid: "2" })
store.dispatch({ type: "GET_SOME_DATA", uuid: "2" })
store.dispatch({ type: "GET_SOME_DATA", uuid: "3" })
store.dispatch({ type: "GET_SOME_DATA", uuid: "2" })
// Get results back for '1', then '3', then '2' assuming equal response times.
// Only the duplicate uuid calls were cancelled, even though all have the same 'type'

我已经尝试使用 .distinctUntilChanged((a, b) => a.uuid === b.uuid) 将流的输入过滤到 .switchMap 中，只是为了看看会发生什么，但这仅仅限制了到达 switchMap 的操作以及取消所有操作的行为但最近的 GET_SOME_DATA 与操作相关的 API 调用仍然发生。

const getDataEpic = (action$) =>
  action$.ofType(GET_SOME_DATA)
    .switchMap(({ uuid }) => // would be great if the switchMap would only cancel existing streams with same uuid
      ajax.getJSON(`/api/datastuff/${uuid}`)
        .map((data) => successAction(uuid, data.values))
        .catch((err) => Observable.of(
          errorAction(uuid),
          setNotificationAction((err.xhr.response && err.xhr.response.message) || 'That went wrong'),
        ))

目前，我正在使用 mergeMap，但我担心这可能会导致问题，例如我已经运行进行实时搜索，其中较旧的请求可能会在最长时间后得到解决最近的一个，导致我的 redux 存储用旧数据更新，因为 mergeMap 不像 switchMap 那样取消 Observable 流...有没有办法让我查看当前的 RxJS Ajax 请求并取消那些与新操作的 url，还是我显然缺少的更好的解决方案？

干杯！

编辑：我想知道将 switchMap 更改为 mergeMap，然后链接 takeUntil 并取消其他 GET_SOME_DATA 操作是否是一种正确的方法，或者如果那只会让所有请求立即取消吗？例如

const getDataEpic = (action$) =>
  action$.ofType(GET_SOME_DATA)
    .mergeMap(({ uuid }) =>
      ajax.getJSON(`/api/datastuff/${uuid}`)
        .takeUntil(
          action$.ofType(GET_SOME_DATA).filter(laterAction => laterAction.uuid === uuid)
        )
        .map((data) => successAction(uuid, data.values))
        .catch((err) => Observable.of(
          errorAction(uuid),
          setNotificationAction((err.xhr.response && err.xhr.response.message) || 'That went wrong'),
    ))

Edit2：显然 takeUntil 添加似乎有效！我不确定它是否 100% 是合适的，但我希望得到一些反馈。我也想支持手动取消选项，讨论的方法是正确的实现吗？

Edit3：我认为这是我的最终版本。删除了 mergeMap 中 Redux 操作的解构，以防 redux-observables 的新手遇到这个：

const getDataEpic = (action$) =>
  action$.ofType(GET_SOME_DATA)
    .mergeMap((action) =>
      ajax.getJSON(`/api/datastuff/${action.uuid}`)
        .takeUntil(Observable.merge(
          action$.ofType(MANUALLY_CANCEL_GETTING_DATA)
            .filter((cancelAction) => cancelAction.uuid === action.uuid),
          action$.ofType(GET_SOME_DATA)
            .filter((laterAction) => laterAction.uuid === action.uuid),
        ))
        .map((data) => successAction(action.uuid, data.values))
        .catch((err) => Observable.of(
          errorAction(action.uuid),
          setNotificationAction((err.xhr.response && err.xhr.response.message) || 'That went wrong'),
    ))

以及通过快速单击视线中的所有内容观察到的网络行为。只有 non-duplicated 个 ID 请求通过了！

Answer 1

您还可以使用 groupBy 运算符来处理具有相同 uuid 的流，并对每个 uuid 操作流应用有用的 switchMap 行为:

action$.ofType(GET_SOME_DATA)
.groupBy(
    ({ uuid }) => uuid, // group all the actions by uuid
    x => x,
    group$ => group$.switchMap(_ => Observable.timer(5000)) // close existing streams if no event of a grouped action is emitted 5 seconds in a row (prevents memory leaks)
)
.mergeMap(actionsGroupedByUuid$ => 
    actionsGroupedByUuid$.switchMap(({ uuid }) => 
        ajax.getJSON(`/api/datastuff/${uuid}`)
            .map((data) => successAction(uuid, data.values))
            .catch((err) => Observable.of(
                errorAction(uuid),
                setNotificationAction((err.xhr.response && err.xhr.response.message) || 'That went wrong'),
            )) 
    )
);

使用 RxJS switchMap 仅取消订阅具有相同请求 URL/action 负载 (redux-observable epics) 的流

Using RxJS switchMap to only unsubscribe from streams with the same request URL/action payload (redux-observable epics)

javascript

rxjs

redux

redux-observable