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

Question

我有一个界面,用户可以触发对相同端点的调用但具有不同的参数(在本例中为UUID).到目前为止,switchMap每当我发送一个具有相同类型的新redux动作时,我一直很享受取消我的飞行中http请求的行为,在这种情况下我仍然想要这种行为,但只有当新动作请求UUID时(部分)动作对象)与正在进行的动作对象相同.我不确定正确的做法.

例如,在一次调度几个动作后,我希望所有与唯一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,但我担心这可能会导致问题,如我碰到的现场搜索过程中一个较旧的请求可能最近的一个后解决,导致我的终极版商店以来旧数据进行更新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%在适当的方面上升,但我喜欢一些反馈.我也想支持手动取消选项,这里讨论的方法是否正确实施？

编辑3:我认为这是我最后的工作版本.删除了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'),
    ))

并且通过快速点击所看到的一切来观察网络行为.只有非重复的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'),
            )) 
    )
);