CombineLatest 导致多个流提交答案

【问题标题】：CombineLatest causes multiple stream submissionsCombineLatest 导致多个流提交
【发布时间】：2021-02-09 00:51:46
【问题描述】：

我想合并三个流，它们分别代表数组的初始状态、添加的元素和编辑的元素。预计一次只执行添加或编辑排放。但是，combineLatest 会触发两次，这会导致向数组添加两次新行程。

private initialTripsSubject = new BehaviorSubject<Trip[]>([]);
  initialTrips$ = this.initialTripsSubject.asObservable();
  private newTripSubject = new BehaviorSubject<Trip>(null);
  addedTrip$ = this.newTripSubject.asObservable();
  private updatedTripSubject = new BehaviorSubject<Trip>(null);
  editedTrip$ = this.updatedTripSubject.asObservable();
  updatedTrips$ = combineLatest([
    this.initialTrips$,
    this.addedTrip$,
    this.editedTrip$
  ]).pipe(
    map(([existingTrips, newTrip, editedTrip]) => {
        if (editedTrip) {
          existingTrips.map(item => {
            if (editedTrip && item.id === editedTrip.id) {
              item.country = editedTrip.country;
              item.departureDate = editedTrip.departureDate;
              item.arrivalDate = editedTrip.arrivalDate;
              item.notes = editedTrip.notes;
            }
            return item;
          });
        }
        if (newTrip) {
          existingTrips.push(newTrip);
        }
        return existingTrips;
      }
    )
);

事件从这个函数被发送到 addedTrip$

 addTrip(newTrip: Trip): void {
    this.newTripSubject.next(newTrip);
  }

在模板中，流通过| async使用

 <table *ngIf="(trips$ | async) as trips; else emptyTrips" class="table table-striped">

我的假设是当addedTrip$ 被处理时，它会改变initialTrips$ 的状态，从而导致combineLatest 的第二次触发。

尝试将流的合并拆分为两部分：merge(editedTips, addedTrip) 和 combineLatest([initialTrips, resultOfMerge])，但没有帮助。

如何避免合并流的双重触发？

更新

Trip$ 在组件中初始化

 trips$ = this.tripService.updatedTrips$$.pipe(
    catchError(err => {
        this.errorMessageSubject.next(err.message);
        return EMPTY;
      }
    )
  );

更新 2

链接到 StarBlitz 的问题

https://stackblitz.com/edit/angular-combinelatest-repeat-pf7q2g?file=src/app/app.component.html

【问题讨论】：

trips$ 是如何定义的？
combineLatest 将在所有 3 个源 observables 发射后第一次发射，然后它会在任何 observables 发射时发射一次。在您的情况下，它将发出初始状态，然后在您调用addTrip() 后发出。你是什么意思“避免双重射击”？你想省略哪个发射？
@BizzyBob 更新了 trip$ 初始化。 “零”发射对我来说很好，因为所有值都是空/空的。当我通过addTrip 函数使用Subject 提交事件时，我观察到updatedTrips$ 的两次发射（但addTrip 方法仅被调用一次）
有趣。每次调用 addTrip 时，我只期望一个发射。你能在 StackBlitz 中重现吗？
@ryzhman 我建立了一个非常相似的简化应用程序stackblitz.com/edit/angular-combinelatest-repeat。如您所见，每次添加行程时，combineLatest 仅被调用一次。请分享一个完整的可重现示例。

标签： angular rxjs

【解决方案1】：

updatedTrips$ combineLatest 被调用两次的原因是由于您的代码中的以下流程：

通过分配this.newTripSubject.next(newTrip)，第一次调用addTrip 触发updatedTrips$ combineLatest。 updatedTrips$'s 依赖的 Observable 之一是 newTripSubject
updatedTrips$ 触发 allTripsWithCountries$ combineLatest 因为它是其依赖的 Observable 之一。这会在列表中添加一趟行程。
maxTripId$ 也可以工作，因为它也依赖于 updatedTrips$。
allTripsWithCountries$ combineLatest 再次触发，因为它依赖于步骤 3 中的 maxTripId$，并再次添加行程。 这就是问题所在

查看我稍作修改的版本here。我删除了allTripsWithCountries$ combineLatest 对maxTripId$ 的依赖。它实际上不是必需的，因为您要遍历每次行程并增加其 id，因此结果是具有顺序 id 的列表。

我相信您可以进一步增强代码并使其更优雅。但我希望过度调用的原因是明确的。

【讨论】：

谢谢。建议的修复工作。但我想了解流程：我的期望是流从上到下排序：updatedTrips$ 的操作将是第一个，然后maxTrip$ 将被计算（因为它在下面）并且只有然后allTripsWithCountries$ 将运行。你的解释是不是意味着同一个流不能被引用两次，否则会导致循环排放？

【解决方案2】：

正如@Benny 指出的那样，您的问题本质上是由于将combinelatest() 与同一源可观察对象的多个实例一起使用。

来自您的 StackBlitz：

  maxTripId$ = this.updatedTrips$.pipe(
    map(trips => Math.max(...trips.map(trip => (trip.id ? trip.id : 0)), 0))
  );
  
  allTripsWithCountries$ = combineLatest([
    this.updatedTrips$,
    this.maxTripId$
  ]).pipe(
    map(...)
  );

这与以下内容基本相同：

  allTripsWithCountries$ = combineLatest([
    this.updatedTrips$,
    this.updatedTrips$.pipe(
      map(trips => Math.max(...trips.map(trip => (trip.id ? trip.id : 0)), 0))
    )
  ]).pipe(
    map(...)
  );

所以实际上，allTripsWithCountries$ 只依赖于一个updatedTrips$，所以你不需要将maxTripId$ 定义为单独的源，也不需要combineLatest：

  allTripsWithCountries$ = this.updatedTrips$.pipe(
      map(trips => {
        if (trips.length > 0) {
          const maxExistingId = Math.max(...trips.map(trip => (trip.id ? trip.id : 0)), 0);
          return trips.map(
            trip => ({
              ...trip,
              country: new Country(name),
              id: maxExistingId + 1
            } as Trip)
        );
      }
    })

更新StackBlitz

【讨论】：