撤消 takeUntil 对触发 rxjs 中另一个动作的影响

Posted 2023-02-19

【中文标题】撤消 takeUntil 对触发 rxjs 中另一个动作的影响

【英文标题】：Undo the effect of takeUntil on trigger of another action in rxjs

【发布时间】：2021-10-12 17:32:13

【问题描述】：

我正在使用 redux-observables 来处理我的 react/redux 应用程序的效果。当我触发一个 redux 操作时，我正面临一种情况，我想撤消 takeUntil 的效果。

这是一个示例代码

const productOrderPlace = (action$, state$) => 
  return action$.pipe(
    ofType(action.PRODUCT_ORDER_PLACE),
    mergeMap(() => 
      return api.productOrderPlace(state$.value).pipe(mergeMap((res) => 
        return merge(of(action.getStatus()).pipe(delay(res.beginAt)), of(action.getStatusExpired()).pipe(delay(15000)))
      ));
    )
  );
;

const getStatus = (action$, state$) => 
  return action$.pipe(
    ofType(action.GET_STATUS),
    takeUntil(merge(
      action$.pipe(ofType(actions.GET_STATUS_EXPIRED)),
      action$.pipe(ofType(actions.GET_STATUS_DONE))
    )),
    mergeMap(() => 
      return api.getStatus(state$.value).pipe(mergeMap(() => 
       // this is the polling call here i would be triggering GET_STATUS again after a specific interval from the current polling response  or call GET_STATUS_DONE if response contains success status
      ));
    )
  );

所以，基本上我正在做的是当用户点击订单按钮时​​，productOrderPlace API 将被调用，然后通过调用getStatus 开始轮询订单状态，这里用于轮询我没有使用来自的默认配方rxjs 带有间隔或计时器，因为轮询间隔不一样，下一个轮询间隔将取自之前的轮询响应。并且应该在几秒钟后根据beginAt 在productOrderPlace 响应中调用第一个轮询调用。用户可以等待的最长时间是 15 秒，所以我一收到productOrderPlace 响应就启动一个计时器

所以，现在我面临一个问题，当触发过期事件时，会向用户显示错误。但是，用户可以再次单击订单按钮并再次下订单，在这种情况下getStatus不会被执行，因为我已经使用了takeUntil并且takeUntil里面的动作已经被触发了，所以我想撤销效果takeUntil 当用户再次点击 Order 按钮（即 PRODUCT_ORDER_PLACE 再次触发时）

【问题讨论】：

投票从哪里开始？我不明白你的两个函数是如何相互使用的

为什么 productOrderPlace 调用自己但只有一个参数？

@Jensen 实际上是 api.productOrderPlace ，它不是同一个功能，我在这里发布时做了一些更改，因为我不想在公共论坛中留下我的公司代码，可能会发生违反，但这段代码是我正在尝试的总体要点。已经编辑了上面的代码。并且轮询不是一个明确的，我在评论中提到我根据 API 的状态再次调用 GET_STATUS 。谢谢

【参考方案1】：

我认为这可能是一种解决方法：

const getStatus = (action$, state$) => 
  return action$.pipe(
    ofType(action.GET_STATUS),

    mergeMap(() => 
      return api.getStatus(state$.value).pipe(mergeMap(() => /* ... */));
    ),

    // Placing the `takeUntil` operator here since the `mergeMap` operator
    // won't pass along the *complete* notification unless all of the active
    // inner observables complete too.
    takeUntil(
      action$.pipe(
        ofType(action.PRODUCT_ORDER_PLACE),
        switchMap(
          () => merge(
            action$.pipe(ofType(actions.GET_STATUS_EXPIRED)),
            action$.pipe(ofType(actions.GET_STATUS_DONE))
          )
        )
      )
    ),
  );

通过使用switchMap，我们在每次发出PRODUCT_ORDER_PLACE 操作时重新订阅。