带有 ngrx 存储的 Angular Rxjs 可观察链

Question

我将 Angular 7 与 ngrx 商店一起使用。该商店包含应用程序状态，我在 OnInit 上的应用程序组件中订阅了它。存储中有几个变量，它们是可互换的(用按钮交换)。

这是我在组件中的示例代码。

this.appState.getGasStationSushi().pipe(switchMap((sushi) => {
  this.gsSushi = sushi;
  return this.appState.getStarbucksSushi();
}), switchMap((sushi) => {
  this.sbSushi = sushi;
  return this.service.compare(this.gsSushi, this.sbSushi);
}).subscribe((result)=>{
  console.log(result);
}));

在 View 中单击按钮，用户可以更改两个 sushi 值，这会导致最后一次订阅调用两次，这是有道理的 (RxJS)。我可以删除 switchMap 并编写类似

的内容

-- gas station sushi susbcription 
   -- star bucks sushi susbcription 
      -- compare

我不是很喜欢这个，我相信肯定有一个 rxjs/operator 之类的。有人可以提出建议吗？

此外，尝试了 forkjoin，但对于 ngrx store 来说，似乎需要调用 first 或 last，如下所示。上面的说法可以引用forkjoinWithstore

const $sushiObs = [
  this.appState.getGasStationSushi().pipe(first()),
  this.appState.getStarbucksSushi().pipe(first())
];
forkjoin($sushiObs).subscribe((result) => {
  console.log(result);
});

如果我使用上述模式，订阅会在第一次触发但之后就不会触发。

Answer 1

首先，这是一个working example在 stackblitz 上。

我没有使用商店，而是创建了一个返回可观察对象的模拟类 SushiState。

class SushiState {
  private _gas = new BehaviorSubject(1);
  private _starbucks = new BehaviorSubject(1);

  public get gas() {
    return this._gas.asObservable();
  }
  public get starbucks() {
    return this._gas.asObservable();
  }

  public increaseSushi(n = 1) {
    this._gas.next(this._gas.value + n);
    this._starbucks.next(this._starbucks.value + n);
  }

  public static compareSushi(g: number, s: number): string {
    return `gas is ${g}, starbucks is ${s}`;
  }
}

至于组件，这里是代码。

export class AppComponent implements OnInit {
  state: SushiState;

  gasSushi: Observable<number>;
  sbSushi: Observable<number>;

  combined: string;
  combinedTimes = 0;
  zipped: string;
  zippedTimes = 0;

  ngOnInit() {
    this.state = new SushiState;
    this.gasSushi = this.state.gas;
    this.sbSushi = this.state.gas;

    const combined = combineLatest(
      this.gasSushi,
      this.sbSushi,
    ).pipe(
      tap((sushi) => {
        console.log('combined', sushi);
        this.combinedTimes++;
      }),
      map((sushi) => SushiState.compareSushi(sushi[0], sushi[1])),
    );
    combined.subscribe(result => this.combined = result);

    const zipped = zip(
      this.gasSushi,
      this.sbSushi,
    ).pipe(
      tap((sushi) => {
        console.log('zipped', sushi);
        this.zippedTimes++;
      }),
      map((sushi) => SushiState.compareSushi(sushi[0], sushi[1])),
    );
    zipped.subscribe(result => this.zipped = result);
  }

  increaseSushi() {
    this.state.increaseSushi();
  }

}

如果您在 stackblitz 上运行它并检查控制台，您将看到以下行为:

如果我们使用 combined latest，我们分别组合 observables 并且只关心最新的状态，导致 console.log 的 2 次调用。

我们可以改为使用 zip，它会等待两个 observable 发出，然后再生成输出。这看起来非常适合我们的“增加两者”按钮，但有一个问题:如果 starbucksSushi 单独增加(可能来自应用程序的不同部分)，压缩 版本也将等待加油站寿司更新。

建议第三种解决方案，您可以使用 combineLatest 组合寿司柜台，然后使用 debounceTime 运算符在发出输出之前等待一定的毫秒数。

const debounced = zip(
  this.gasSushi,
  this.sbSushi,
).pipe(
  tap((sushi) => {
    console.log('debounced', sushi);
    this.debouncedTimes++;
  }),
  map((sushi) => SushiState.compareSushi(sushi[0], sushi[1])),
  debounceTime(100),
);
debounced.subscribe(result => this.debounced = result);

这将对所有来源的变化使用react，但不会超过 100ms。

最后，您必须执行 first() 的原因:

forkJoin 在 observables 完成后加入它们(这只能发生一次，所以它不适合连续流)并且更适合“类似 promise ”的工作，例如HTTP 调用、流程完成等。顺便说一下，如果您只从流中获取一个元素，则生成的流会在单次发射后完成。

附言

我建议使用 async 管道来处理可观察对象(就像我处理属性一样

gasSushi: Observable<number>;
sbSushi: Observable<number>;

然后在模板内部

<div>
  <h3>starbucks sushi</h3>
  <p>{{sbSushi | async}}</p>
</div>

代替

result => this.zipped = result

我在此示例中同时使用了两者，因此您可以比较它们。根据我的经验，使用可观察对象会变得更加容易，一旦您提前停止转换“去观察”它们并且只允许 async 管道完成它的工作。

最重要的是，如果您在组件中的某处使用了subscribe，那么当组件被销毁时您应该unsubscribe——这一点也不难，但是如果我们永远不要显式订阅，并允许 async 管道进行订阅，它还为我们处理销毁 :)