NgRx Effect中tap操作符无法分发importStatusUpdated action问题

你在tap里调用MigrationActions.importStatusUpdated()只是创建了一个action对象，但并没有把它分发到NgRx的action流中。tap操作符仅用于执行副作用（比如打印日志），它不会修改Observable的数据流，也不会自动触发NgRx的action分发——只有Effect返回的Observable中emit的action才会被NgRx处理并分发。

要实现每次轮询都更新Store中的状态，你需要把importStatusUpdated action加入到Observable的数据流里，让Effect能够emit它。这里提供两种简洁的实现方式：

方式一：直接在Effect中分发更新action #

把状态更新逻辑移到Effect里，直接调用store.dispatch()分发action，代码逻辑更直观：

pollForImportProcessStatusEffect = createEffect(() =>
      this.actions$.pipe(
        ofType(MigrationActions.importInProcess),
        switchMap((payload) =>
          timer(0, 2000).pipe(
            concatMap(() => this.apiClient.getImportProcessStatus(payload.importedFileId)),
            // 直接通过store.dispatch分发更新action
            tap((status) => this.store.dispatch(MigrationActions.importStatusUpdated(status))),
            takeWhile(status => status.status === FileStatus.IN_PROGRESS, true),
            map(data => MigrationActions.importProcessFinished(data.status)),
            catchError(() => of(MigrationActions.importInProcess(payload.importedFileId)))
          )
        )
      )
  );

方式二：让私有方法同时返回状态和action #

如果希望把轮询逻辑封装在私有方法里，可以用mergeMap让方法同时发出状态数据和更新action，再在Effect中拆分处理：

调整私有方法 #

private pollForProcessStatus(importedFileId: number): Observable<ImportFileProcessStatus | Action> {
    return timer(0, 2000).pipe(
      concatMap(() => this.apiClient.getImportProcessStatus(importedFileId)),
      // 同时发出状态数据和更新action
      mergeMap((status: ImportFileProcessStatus) => [
        status,
        MigrationActions.importStatusUpdated(status)
      ]),
      takeWhile((value) => {
        // 仅对状态数据判断是否继续轮询，action直接放行
        if (value instanceof ImportFileProcessStatus) {
          return value.status === FileStatus.IN_PROGRESS;
        }
        return true;
      }, true),
      takeLast(1)
    );
  }

调整Effect逻辑 #

pollForImportProcessStatusEffect = createEffect(() =>
      this.actions$.pipe(
        ofType(MigrationActions.importInProcess),
        switchMap((payload) =>
          this.pollForProcessStatus(payload.importedFileId).pipe(
            mergeMap((value) => {
              // 区分数据流中的内容，分别处理
              if (value instanceof ImportFileProcessStatus) {
                return of(MigrationActions.importProcessFinished(value.status));
              }
              return of(value); // 分发更新action
            }),
            catchError(() => of(MigrationActions.importInProcess(payload.importedFileId)))
          )
        )
      )
  );

内容的提问来源于stack exchange，提问作者Tom Carmi