具有 redux-observable 和 RxJS 的非严格序列

Question

我的应用有一个带有微调器的模式，只要发生长时间的阻塞操作，就会显示该模式。 有几个这样的长阻塞动作，每个动作都有一个标记它的开始和结束的动作。

鉴于“动作流”，每当调度一个开始动作时，我想调度showWaitingIndication 动作，直到调度相应的结束动作，然后调度hideWaitingIndication。如果分派了另一个开始操作，然后在第一个阻塞操作正在进行时分派了相应的结束操作，则不应再次调用showWaitingIndication 或hideWaitingIndication。 hideWaitingIndication 也不应在操作仍处于活动状态时被调度。

基本上这个想法是，只要阻塞操作处于活动状态，等待指示就不应隐藏。

例如

StartA -> dispatch(showWaitingIndication) -> 其他事件 -> endA -> dispatch(hideWaitingIndication)

StartA -> dispatch(showWaitingIndication) -> startB -> endB (不应该调用 hide) -> endA -> dispatch(hideWaitingIndication)

还有 StartA -> dispatch(showWaitingIndication) -> startB -> endA (不应该调用 hide！) -> endB -> dispatch(hideWaitingIndication)

我正在努力思考如何使用流来实现这一点（我坚信它非常适合这个问题）。

到目前为止，我已经想出了类似的东西（不起作用）

    let showHideActionPairs = getShowHideActionPairs(); // { "startA": "endA", "startB": "endB"}
    let showActions  = Object.keys(showHideActionPairs);

    return action$ => action$.pipe(
        filter(action => Object.keys(showHideActionPairs).includes(action.type)),
        switchMap(val =>
            {
                let hideAction = showHideActionPairs[val.type];
                return concat(
                    of(waitingIndicationShowAction),
                    empty().pipe(
                            ofType(hideAction),
                            mapTo(waitingIndicationHideAction)
                    ))
            }
        )
    );

这样做的正确方法是什么？

Answer 1

这是一个非常有趣的问题！

我想你可以试试这个：

const showHideActionPairs = getShowHideActionPairs(); // { "startA": "endA", "startB": "endB"}

actions$.pipe(
  windowWhen(() => actions$.pipe(filter(action => action.type === hideWaitingIndication))),
  
  mergeMap(
    window => window.pipe(
      mergeMap(
        action => someAsyncCall().pipe(
          mapTo(showHideActionPairs[action]),
          startWith(showHideActionPairs[action])
        )
      ),

      scan((acc, crtEndAction) => {
        // first time receiving this end action -> the beginning of the async call
        if (!(crtEndAction in acc)) {
          acc[crtEndAction] = true;

          return acc;
        }

        // if the `crtEndAction` exists, it means that the async call has finished
        const {[crtEndAction]: _, ...rest} = acc;

        return rest;
      }, Object.create(null)),

      filter(obj => Object.keys(obj).length === 0),

      mapTo(hideWaitingIndication),

      // a new window marks the beginning of the modal
      startWith(showWaitingIndication),
    )
  )
)

我的第一个想法是我需要找到一种方法来表示一个事件链，这样链子从showWaitingIndication 开始并以@987654325 结束@。链的末端实际上由最后完成的异步调用 (end{N}) 指示。所以我认为这将是windowWhen 的一个很好的用例。

但是 window 是什么？一个窗口是nothing more than a Subject:

/* ... */
const window = this.window = new Subject<T>();
this.destination.next(window);
/* ... */

windowWhen(() => closeNotifier) 的工作方式是将Subject(a window) 作为next 值发送（这就是我们有mergeMap(window => ...) 的原因），它会通过它推送值（例如动作） .我们在window.pipe(...) 中访问这些值。当closeNotifier 发出时，当前的window 将complete 和一个新的window 将被创建并传递，以便后续的动作将通过它发送。值得注意的是，默认创建了一个窗口when the stream is subscribed：

constructor(protected destination: Subscriber<Observable<T>>,
            private closingSelector: () => Observable<any>) {
  super(destination);
  this.openWindow(); // !
}

假设我们正在当前窗口中接收第一个操作。

mergeMap(
  action => someAsyncCall().pipe(
    mapTo(showHideActionPairs[action]),
    startWith(showHideActionPairs[action])
  )
),

一旦该动作被拦截，我们就会发送它的预期结束值，这样它就可以存储在scan 的累加器中。当该操作的异步调用完成时，它将再次发送该结束值，以便可以将其从累加器中删除。
这样，我们可以确定一个窗口的生命周期，当累加器中没有更多的结束值时，该窗口将被关闭。

当这种情况发生时

filter(obj => Object.keys(obj).length === 0),

mapTo(hideWaitingIndication),

我们确保通知所有操作已完成其任务。

Answer 2

我已经接受了 Andrei 的回答，因为他要向我指出正确的方向，他的解决方案涉及 windowWhen 和 accumulator 是解决此问题的正确思维框架。为了完整起见，我还根据他发布了我自己的解决方案，因为我觉得这里的逻辑更明确（而且我个人在寻找解决方案时更容易理解）：

let showHideActionPairs = getShowHideActionPairs();
const relevantActionsTypesArray = Object.keys(showHideActionPairs).concat(Object.values(showHideActionPairs));

actions$ => actions$.pipe(
        // close the "window" when a hide action is received
        windowWhen(() => actions$.pipe(ofType(waitingIndicationHideActionName),)),

        mergeMap(
            window => window.pipe(
                // filter to only look at start/end actions
                ofType.apply(null, relevantActionsTypesArray),
                scan((accumulator, action) => {
                    let waitingForEndAction  = "startAction" in accumulator;
                    // first time we see a start action
                    if (!waitingForEndAction && action.type in showHideActionPairs) {
                        accumulator.startAction = action.type;
                        accumulator.actionable = true;
                    // found the right end action
                    } else if (waitingForEndAction && action.type === showHideActionPairs[accumulator.startAction]) {
                        accumulator.endAction = action.type;
                        accumulator.actionable = true;
                    // any other case is not actionable (will not translate to to an action)
                    }  else {
                        accumulator.actionable = false;
                    }
                    return accumulator;
                }, {}),
                // accumulator spits out stuff for every action but we only care about the actionables
                filter(obj => obj.actionable),
                map(obj => {
                    if (obj.endAction){
                        return waitingIndicationHideAction
                    } else if (obj.startAction) {
                        return waitingIndicationShowAction
                    }
                }),
            )
        )
    )

};