angular2 observable 扫描一次然后被许多异步订阅

Question

我需要在不同的时刻两次绑定到共享的可观察对象。第二个绑定在第一次计算时得到 null，直到下一个项目出现。

这里是组件 class：

export class App {
  subject = new BehaviorSubject(1);
  scan = this.subject.scan((current, change) => current + change, 0).share();
  other = this.scan.map(item => item);
}

这是模板：

<div>
  <button (click)="clickedScan=true">show scan</button>
  <button (click)="clickedOther=true">show other</button>
  <button (click)="subject.next(1)">next</button>
  <div *ngIf="clickedOther">
    other | async: <b>{{ '' + (other | async) }}</b>
  </div>
  <div *ngIf="clickedScan">
    scan | async: <b>{{ '' + (scan | async) }}</b>
  </div>
</div>

这里是 plunker（更新：plunker 已更新为已接受的答案）

share() 是必需的，否则会为每个订阅者重复调用 scan 方法，但一段时间后完成的下一个 async 绑定无法访问最后一个元素。在不使用 share() 的情况下，所有绑定从一开始就有效，但随后 scan 被调用两次 subject.next() 调用（在这个 plunker 示例中的单独项目实例上）。出于多种原因，我想避免这种重复的 scan 调用 - 至少不要为每个订阅者重复完全相同的工作并获得相同的结果。

我想知道避免多次 share（即使用其他一些 Observable 方法）调用并在新 async 绑定时仍然提供最后一个元素的正确方法是什么。

Answer 1

是的，您需要一个热可观察对象，以便共享一个订阅而不是单独订阅。您可能对安德烈的这段视频感兴趣：https://egghead.io/lessons/rxjs-demystifying-cold-and-hot-observables-in-rxjs. You might also be interested in Paul Taylor's talk from Reactive 2015: https://youtu.be/QhjALubBQPg?t=385

基本上，您可以像这样重写代码：

import {Subject, ReplaySubject} from "rxjs/Rx"; // only really need these two

/* your other code */

export class App {

  // plain old subject for clicks
  // i believe you can get an event stream somewhere?
  // sorry, i don't know angular2
  subject = new Subject();

  // replays the last event for observers on subscription
  main = new ReplaySubject(1);

  // you could apply transforms here if you want
  scan = this.main
  other = this.main

  constructor() {
    // take the events that come in
    this.subject

      // start our observable with an initial event
      .startWith(0)

      // when subject emits, this will run and update
      .scan((current, change) => current + change)

      // now we can subscribe our replay subject
      .subscribe(this.main);
  }
}

希望对您有所帮助。

angular2 observable 扫描一次然后被许多异步订阅

angular2 observable scan once then subscribe by many async

rxjs

angular