如何等待其他史诗完成然后调用回调？

Question

我试图在第一个史诗中映射多个动作，在第二个史诗中处理它们，然后在第一个史诗中调用一个回调。

这是我看到的流程：

1. fetchMultipleOperationsEpic侦听OPERATIONS_FETCH_ALL并创建fetchOperations操作流。
2. 然后fetchOperationsEpic启动，为每个调度的操作使用fetchOperationData获取数据。
3. 获取每个操作的数据后，fetchMultipleOperationsEpic应调用回调。

const fetchMultipleOperationsEpic = action$ => action$.pipe(
    ofType(OPERATIONS_FETCH_ALL),
    switchMap(({ payload }) => {
        const stream = of(...payload.ids.map(processId => fetchOperations({ processId })));

        // PROBLEM HERE: 'done' is logged before actions from fetchOperationsEpic complete
        return stream.pipe(finalize(() => console.log('done')))
    })
)

const fetchOperationsEpic = action$ => action$.pipe(
    ofType(OPERATIONS_FETCH),
    mergeMap(({ payload }) => fetchOperationData(payload.processId))
)

const fetchOperationData = processId => {
    api.fetchOperationsByProcess(processId).pipe(
        mergeMap(operations => of(fetchSuccess(), fooAction(), barAction())),
        catchError(error => of(fetchOperationsFailure({ processId, error })))
    )
}

也尝试使用forkJoin但没有成功：

const fetchMultipleOperationsEpic = action$ => action$.pipe(
    ofType(OPERATIONS_FETCH_ALL),
    switchMap(({ payload }) => {
        const observables = [...payload.ids.map(processId => of(fetchOperations({ processId })))];

        // Doesn't wait as well
        return forkJoin(observables).pipe(
            mergeMap(o => o),
            finalize(() => console.log('done'))
        )
    })
)

实际上，我使其工作的唯一方法是避免调用fetchOperations操作（从而避免调用fetchOperationsEpic），并直接调用fetchOperationData：

const fetchMultipleOperationsEpic = action$ => action$.pipe(
    ofType(OPERATIONS_FETCH_ALL),
    mergeMap(({ payload }) => {
        const observables = [...payload.ids.map(id => fetchOperationData(id))];
        return merge(...observables)
            .pipe(finalize(() => console.log('done')))
    })
)

哪个好。我只是想知道我的初步方法是否真的可行。

Answer 1

你认为你有“问题”的地方：

return stream.pipe(finalize(() => console.log('done')))

这实际上是正确的行为。 finalize将在其正上方的of返回的操作列表被发送到商店后立即运行。那是因为你已经明确地将finalize链接到那个内部可观察者。

您正在考虑的方法的问题是，一旦某个操作进入商店，它就不再属于该特定可观察范围。你不能简单地等待处理其他史诗中的特定动作，因为最终，史诗永远不会真正结束。在严格的RxJS意义上，可观察者从不发出完整的通知。

一般来说，我建议寻找一个使用单个史诗的解决方案（可能与你上一个例子一致）。这允许您简单地链接运算符，例如finalize，而无需跳过太多的箍。您仍然可以重构您的代码，以便史诗可以使用可重用的函数，以防您想要在不同的史诗之间共享/复制业务逻辑。

您还可以查看依赖于通过不同操作进行史诗间通信的解决方案。但是，根据工作流程的复杂程度，这可能会变得棘手。您需要每个操作来唯一标识子流程。这是一个想法：

const fetchMultipleOperationsEpic = action$ => action$.pipe(
  ofType(OPERATIONS_FETCH_ALL),
  switchMap(({ payload }) =>
    merge(...payload.ids.map(processId =>
      concat(
        of(fetchOperations({ processId })),
        // Note: This keeps the observable running until either
        // the corresponding "success" or "failure" action is
        // dispatched:
        action$.pipe(
          filter(action => (action.type === FETCH_SUCCESS || action.type === FETCH_FAILURE) && action.payload.processId === processId),
          first(),
          ignoreElements(),
        ),
      )
    )).pipe(
      finalize(() => console.log('done')),
    )
  )
)

const fetchOperationsEpic = action$ => action$.pipe(
    ofType(OPERATIONS_FETCH),
    mergeMap(({ payload }) => fetchOperationData(payload.processId))
)

const fetchOperationData = processId => {
    api.fetchOperationsByProcess(processId).pipe(
        // Note: We have to pass along the `processId` in the
        // "success" action so that the inner observables in
        // `fetchMultipleOperationsEpic` can complete:
        mergeMap(operations => of(fetchSuccess({ processId }), fooAction(), barAction())),
        catchError(error => of(fetchOperationsFailure({ processId, error })))
    )
}