angular - 如何在@ngrx/effects 中等待 2 个 Action

Question

effect 可以像 Promise.all 一样等待两个 action 吗？示例:

@Effect()
pulic addUser() {
   return this.actions$.ofType(user.ADD)
      .switchMap(() => {
         return this.userService.add();
      })
      .map(() => {
         return new user.AddSuccessAction();
      });
}

@Effect()
pulic addUserOptions() {
   return this.actions$.ofType(userOptions.ADD)
      .switchMap(() => {
         return this.userOptionsService.add();
      })
      .map(() => {
         return new userOptions.AddSuccessAction();
      });
}

@Effect()
public complete() {
   return this.actions$.ofType(user.ADD_SUCCESS, userOptions.ADD_SUCCESS)
      // how to make it works like Promise.all ?
      .switchMap(() => {
         return this.statisticService.add();
      })
      .map(() => {
         return new account.CompleteAction();
      });
}

已更新我想要实现的是与 Promise.all 类似的行为。如何并行调度两个效果，等到所有效果都解决了，然后调度第三个 Action 。类似于 https://redux-saga.js.org/docs/advanced/RunningTasksInParallel.html有了 promise ，这是很明显的:

Promise.all([fetch1, fetch2]).then(fetch3);

在 ngrx/effects 中可以吗？还是ngrx/effects的方式不对？

回答

您可以使用几个选项:

1) 不要使用通用操作。

遵循 Myke Ryan 演讲中的这些规则:https://youtu.be/JmnsEvoy-gY

优点:更容易调试

缺点:大量样板和操作

2) 对嵌套操作使用复杂流。

查看这篇文章:https://bertrandg.github.io/ngrx-effects-complex-stream-with-nested-actions/

这是两个 Action 的简单示例:

@Effect()
public someAction(): Observable<Action> {
    return this.actions$.pipe(
        ofType(actions.SOME_ACTION),
        map((action: actions.SomeAction) => action.payload),
        mergeMap((payload) => {
            const firstActionSuccess$ = this.actions$.pipe(
                ofType(actions.FIRST_ACTION_SUCCESS),
                takeUntil(this.actions$.pipe(ofType(actions.FIRST_ACTION_FAIL))),
                first(),
            );

            const secondActionsSuccess$ = this.actions$.pipe(
                ofType(actions.SECOND_ACTION_SUCCESS),
                takeUntil(this.actions$.pipe(ofType(actions.SECOND_ACTION_FAIL))),
                first(),
            );

            const result$ = forkJoin(firstActionSuccess$, secondActionsSuccess$).pipe(
                first(),
            )
                .subscribe(() => {
                    // do something
                });

            return [
                new actions.FirstAction(),
                new actions.SecondAction(),
            ];
        }),
    );
}

优点:你可以实现你想要的

缺点:复杂流太复杂而无法支持 :) 看起来很难看并且可能很快就会变得糟糕，可观察对象在操作成功或失败之前不会取消订阅，这意味着理论上任何第三方 Action 可以向这些可观察对象发出信号。

3) 使用聚合器模式。

查看 Victor Savkin 关于 NgRx 状态管理模式和最佳实践的演讲:https://www.youtube.com/watch?v=vX2vG0o-rpM

这是一个简单的例子:

首先，您需要使用 correlationId 参数创建操作。 CorrelationId 应该是 uniq，例如它可能是一些 guid。您将在您的操作链中使用此 ID 来标识您的操作。

export class SomeAction implements Action {
    public readonly type = SOME_ACTION;

    constructor(public readonly correlationId?: string | number) { }
    // if you need payload, then make correlationId as a second argument
    // constructor(public readonly payload: any, public readonly correlationId?: string | number) { }
}

export class SomeActionSuccess implements Action {
    public readonly type = SOME_ACTION_SUCCESS;

    constructor(public readonly correlationId?: string | number) { }
}

export class FirstAction implements Action {
    public readonly type = FIRST_ACTION;

    constructor(public readonly correlationId?: string | number) { }
}

export class FirstActionSuccess implements Action {
    public readonly type = FIRST_ACTION_SUCCESS;

    constructor(public readonly correlationId?: string | number) { }
}

// the same actions for SecondAction and ResultAction

然后我们的效果:

@Effect()
public someAction(): Observable<Action> {
    return this.actions$.pipe(
        ofType(actions.SOME_ACTION),
        mergeMap((action: actions.SomeAction) => {
            return [
                new actions.FirstAction(action.corelationId),
                new actions.SecondAction(action.corelationId),
            ];
        }),
    );
}

@Effect()
public firstAction(): Observable<Action> {
    return this.actions$.pipe(
        ofType(actions.FIRST_ACTION),
        switchMap((action: actions.FirstAction) => {
            // something
            ...map(() => new actions.FirstActionSuccess(action.correlationId));
        }),
    );
}
// the same for secondAction

@Effect()
public resultAction(): Observable<Action> {
    return this.actions$.pipe(
        ofType(actions.SOME_ACTION),
        switchMap((action: actions.SomeAction) => {
            const firstActionSuccess$ = this.actions$.pipe(
                ofType(actions.FIRST_ACTION_SUCCESS),
                filter((t: actions.FirstActionSuccess) => t.correlationId === action.correlationId),
                first(),
            );

            const secondActionsSuccess$ = this.actions$.pipe(
                ofType(actions.SECOND_ACTION_SUCCESS),
                filter((t: actions.SecondActionSuccess) => t.correlationId === action.correlationId),
                first(),
            );

            return zip(firstActionSuccess$, secondActionsSuccess$).pipe(
                map(() => new actions.resultSuccessAction()),
            )
        }),
    );
}

优点:与第 2 点相同，但没有第三方操作。

缺点:与第 1 点和第 2 点相同

4) 不要对 API 使用效果。使用模拟效果但返回 Observable 的良好旧服务。

为您服务:

public dispatchFirstAction(): Observable<void> {
    this.store.dispatch(new actions.FirstAction(filter));

    return this.service.someCoolMethod().pipe(
        map((data) => this.store.dispatch(new actions.FirstActionSuccess(data))),
        catchError((error) => {
            this.store.dispatch(new actions.FirstActionFail());

            return Observable.throw(error);
        }),
    );
}

所以你以后可以在任何地方组合它，比如:

const result1$ = this.service.dispatchFirstAction();
const result2$ = this.service.dispatchSecondAction();

forkJoin(result1$, result2$).subscribe();

5) 使用 ngxs:https://github.com/ngxs/store

优点:更少的样板，这感觉像是棱 Angular 分明的东西，它增长很快

缺点:功能不如 ngrx

Answer 1

我是 RXJS 的新手，但是这个呢。

如果将 tap 更改为 switchMap，则可以删除 {dispatch: false}。

@Effect({dispatch: false})
public waitForActions(): Observable<any> {
    const waitFor: string[] = [
        SomeAction.EVENT_1,
        SomeAction.EVENT_2,
        SomeAction.EVENT_3,
    ];

    return this._actions$
        .pipe(
            ofType(...waitFor),
            distinct((action: IAction<any>) => action.type),
            bufferCount(waitFor.length),
            tap(console.log),
        );
}