RXJS 一个接一个地进行可观察的调度（不是 concat）

Question

我正在使用 redux，redux-observable 在 react 中。

我有以下

const goToItemEpic = (action$, state$): Observable<any> => action$.pipe(
    ofType(ItemsDetailsActions.goToItem),
    concatMap((action: {payload: {projectId: string; datastoreId: string; itemId: string}}) =>
        of(ProjectActions.setCurrentProjectId({projectId: action.payload.projectId})).pipe(
            delay(1000),
            switchMap(() => {
                return [
                    DatastoreActions.setCurrentDatastoreId({datastoreId: action.payload.datastoreId}),
                    ItemsActions.setCurrentItemId({ itemId: action.payload.itemId })
                ];
            })
        )
    )
);

基本上我想派发setCurrentProjectId然后等一段时间再派发 setCurrentDatastoreId 和 setCurrentItemId

我面临两个问题：

1）当前代码延迟setCurrentDatastoreId和setCurrentItemId但不执行setCurrentProjectId，为什么？

2) 这里的延迟是硬编码的 1000 毫秒。但我想根据另一个可观察到的执行来延迟 getDatastoresSuccess

所以它会是

execute => setCurrentProjectId
waitFor => getDatastoresSuccess
execute => setCurrentDatastoreId
execute => setCurrentItemId

我如何在 redux-observable 中做到这一点？

编辑：

const goToItemEpic = (action$, state$): Observable<any> => action$.pipe(
    ofType(ItemsDetailsActions.goToItem),
    concatMap((action: {payload: {projectId: string; datastoreId: string; itemId: string}}) =>
        concat(
            ProjectActions.setCurrentProjectId({projectId: action.payload.projectId}),
            action$.pipe(
                ofType(DatastoreActions.getDatastoresSuccess),
                first(),
                switchMap(() => of(
                    DatastoreActions.setCurrentDatastoreId({datastoreId: action.payload.datastoreId}),
                    ItemsActions.setCurrentItemId({ itemId: action.payload.itemId })
                ))
            )
        )
    )
);

subscribeTo.js:23 Uncaught TypeError: You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Iterable.

Answer 1

您可以通过再次使用 action$.pipe 来等待所需的操作：

const goToItemEpic = (action$, state$): Observable<any> => action$.pipe(
    ofType(ItemsDetailsActions.goToItem),
    concatMap((action: {payload: {projectId: string; datastoreId: string; itemId: string}}) =>
        concat(
            of(ProjectActions.setCurrentProjectId({projectId: action.payload.projectId})),
            action$.pipe(
                ofType(ItemsDetailsActions.getDatastoresSuccess),
                first(),
                switchMap(() => of(
                    DatastoreActions.setCurrentDatastoreId({datastoreId: action.payload.datastoreId}),
                    ItemsActions.setCurrentItemId({ itemId: action.payload.itemId })
                )
            })

因此，在 concat 中，您首先发出 ProjectActions.setCurrentProjectId 操作，然后等待 getDatastoresSuccess 类型的 first() 操作并将其映射到 setCurrentDatastoreId 和 setCurrentItemId

问题 1) 的答案是 setCurrentProjectId 用作 switchMap 的（忽略的）输入，然后映射到 setCurrentDatastoreId 和 setCurrentItemId

RXJS 一个接一个地进行可观察的调度（不是 concat）

RXJS make an observable dispatch after another (not concat)

observable

rxjs

redux

redux-observable