如何在RxJS和redux-observable中正确排序和响应动作？

Question

就RxJS而言，我还在学习绳索。我在项目中使用redux-observable，我真的很喜欢。但是，我觉得这里没有掌握一些基本知识，因此我想向您寻求帮助。

这是示例代码：

const requestDelete = action$ => {
  const actionSuccess$ = action$.pipe(
    ofType(types.AN_ACTION_THAT_MAY_FAIL_SUCCESS),
    take(1),
    mergeMap(_ => of(actions.deleteSuccess()))
  )

  const actionFailed$ = action$.pipe(
    ofType(types.AN_ACTION_THAT_MAY_FAIL_FAILED),
    take(1),
    mergeMap(_ => of(actions.deleteFailed()))
  )

  return action$.pipe(
    ofType(types.REQUEST_DELETE),
    mergeMap(action =>
      from(SomeService.delete(action.payload)).pipe(
        mergeMap(_ =>
          merge(
            actionSuccess$,
            actionFailed$,
            of(actions.requestAnActionThatMayFail())
          )
        ),
        catchError(err => of(actions.deleteFailed(err)))
      )
    ),
    catchError(err =>
      of({ type: 'CRITICAL_REQUEST_DELETE_ERROR', err })
    )
  )
}

这是我想到的流程：

1. 用户请求删除某些资源。
2. 调用SomeService来做到这一点。
3. 如果解决，我想打电话给requestAnActionThatMayFail。现在，我想依靠该操作的结果并等待其完成。
4. 如果可以，我想选择actions.deleteSuccess())。
5. 如果失败，我想使用actions.deleteFailed()触发错误等。

现在，问题是-此代码works，但是merge对我来说很不好。我认为我应该使用concat并将of(actions.requestAnActionThatMayFail())用作第一个参数。如果更改流的顺序或使用concat，则整个史诗将不起作用。任何想法可能是什么问题？

Answer 1

如果我更改流的顺序

我怀疑它不会这样工作，因为requestAnActionThatMayFail是同步操作。这解释了为什么，因为merge将顺序地]订阅提供的可观察对象，这意味着如果您有

merge(
 actionFailed$,
 of(actions.requestAnActionThatMayFail()),
 actionSuccess$,
)
并且requestAnActionThatMayFail成功，由于actionSuccess$尚未订阅（actionFailed$已订阅），它将无法工作。如果requestAnActionThatMayFail为asynchronous

，则不应遇到此问题。

或使用concat

concat是merge，concurrent设置为1。虽然merge可以有多个活动内部观测值，但concat只能有一个。设置concurrent时，merge将缓冲超出的值，并且当一个内部可观察值completes

时，它将采用最早的缓冲值并从中创建一个内部可观察值。解释与上一节中的解释相同：其他可观察的对象[[尚未]订阅。所以，这就是我的方法：

const requestDelete = action$ => {
  const actionSuccess$ = action$.pipe(
    ofType(types.AN_ACTION_THAT_MAY_FAIL_SUCCESS),
    take(1),
    mergeMap(_ => of(actions.deleteSuccess()))
  )

  const actionFailed$ = action$.pipe(
    ofType(types.AN_ACTION_THAT_MAY_FAIL_FAILED),
    take(1),
    mergeMap(_ => of(actions.deleteFailed()))
  )

  const requestDelete$ = action$.pipe(
    ofType(types.REQUEST_DELETE),
    mergeMap(action =>
      from(SomeService.delete(action.payload)),
      mergeMap(_ => of(actions.requestAnActionThatMayFail())),
      catchError(err => of(actions.deleteFailed(err)))
    ),

    // This seems a little redundant because of the previous `catchError`
    // catchError(err =>
    //   of({ type: 'CRITICAL_REQUEST_DELETE_ERROR', err })
    // )
  )

  // Make sure all the obs. are susbcribed
  return merge(actionSuccess$, actionFailed$, requestDelete$);
}