避免使用 @ngrx/effects 来取消相同的操作

Question

有没有办法避免取消前一个的影响？

我需要去做：

this.tagsStoreService.initTagsForFamilyId(this.tagFamily.id)

我有这样的效果：

@Effect() initTagsForFamilyId$: Observable<Action> = this.actions$
    .pipe(
        ofType<InitTagsForFamilyIdAction>(TagsStateActionTypes.INIT_TAGS_FOR_FAMILY_ID_ACTION),
        switchMap(params => {
            // this.loadingService.showLoading();
            return this.tagsService.initTagsForFamilyId(params.payload)
                .pipe(
                    exhaustMap((data) => {
                        this.loadingService.hideLoading();

                        return [
                            new InitTagsForFamilyIdSuccessAction({ ...data }),
                        ];
                    }),
                    catchError(error => {
                        // this.loadingService.hideLoading();
                        return of(new TagsFailureAction({ error }));
                    }),
                );
        }),
    );

预先感谢您的帮助;-)

Answer 1

你应该mergeMap使用switchMap

let {
  interval,
  of
} = rxjs
let {
take,
tap,
switchMap,
delay,
finalize
} = rxjs.operators

const switchMapCase = interval(500).pipe(
  take(5),
  tap(() => console.log("new request")),
  switchMap(() =>
    of(null).pipe(
      delay(800),
      tap(() => console.log("end")),
      finalize(() => console.log("request stream complete"))
    )
  )
);

switchMapCase.subscribe();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.6/rxjs.umd.min.js"></script>

如果您运行上面的代码片段（带有switchMap），您会注意到每500毫秒都会发出一个新事件（新请求），该事件将进入switchMap并返回一个新的可观察量，该可观察量将在800ms 后解析，因为800>500在内部之前observable 已解析，新的 observable 将来自源流，这将取消当前的内部流（具有 ms 延迟的流），并且将返回800一个也具有 ms 延迟的新流。800这个事情会发生 5 次（由于take(5)操作符的原因，这相当于调度该TagsStateActionTypes.INIT_TAGS_FOR_FAMILY_ID_ACTION动作 5 次，500每次 dispathc 之间有 ms 的延迟）。在第 5 次迭代中，内部可观察量将完成，因为不会有新的调度来取消它。

let {
  interval,
  of
} = rxjs
let {
take,
tap,
mergeMap,
delay,
finalize
} = rxjs.operators

const mergeMapCase = interval(500).pipe(
  take(5),
  tap(() => console.log("new request")),
  mergeMap(() =>
    of(null).pipe(
      delay(800),
      tap(() => console.log("server response recieved")),
      finalize(() => console.log("request stream complete"))
    )
  )
);

mergeMapCase.subscribe();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.6/rxjs.umd.min.js"></script>

在上面的代码片段（）中mergeMap，我们几乎有相同的结构，唯一的区别是我们使用的不同之处mergeMap在于switchMap，不像switchMap（取消前一个流）mergeMap将新的“innerStream”添加到源可观察中而不取消旧的。

*旁注：*实际上在效果中被取消的东西不是可观察的，action或者effect是可this.tagsService.initTagsForFamilyId观察的，完成后将触发另一个action。