如何将多播 obs 与行为主题一起使用？

Question

一般来说，我们需要行为主体功能。但只有在第一次订阅时，我们才应该将订阅发送到 REST 中的服务器。并在最后一次取消订阅时发送取消订阅，并且所有订阅的后期观察者将 gwt 从第一个接收到的最新 json。我可以使用 rxjs 操作符来做到这一点吗？怎么做？还是我应该使用自定义 obserbale ？

目前的自定义代码是这样的：

public observable: Observable<TPattern> = new Observable((observer: Observer<TPattern>) => {
 this._observers.push(observer);
 if (this._observers.length === 1) {
  this._subscription = this.httpRequestStream$
    .pipe(
      map((jsonObj: any) => {
        this._pattern = jsonObj.Data;
        return this._pattern;
      })
    )
    .subscribe(
      (data) => this._observers.forEach((obs) => obs.next(data)),
      (error) => this._observers.forEach((obs) => obs.error(error)),
      () => this._observers.forEach((obs) => obs.complete())
    );
}
if (this._pattern !== null) {
  observer.next(this._pattern); // send last updated array
}
return () => {
  const index: number = this._observers.findIndex((element) => element === observer);
  this._observers.splice(index, 1);
  if (this._observers.length === 0) {
    this._subscription.unsubscribe();
    this._pattern = null; // clear pattern when unsubscribed
  }
};

});

Answer 1

听起来您需要shareReplay(1)，它将与所有订阅者分享最新的回复。

const stream$ = httpRequestStream$.pipe(
  shareReplay(1),
),

stream$.subscribe(); // sends the request and gets its result
stream$.subscribe(); // doesn't send it but gets cached result
stream$.subscribe(); // doesn't send it but gets cached result
stream$.subscribe(); // doesn't send it but gets cached result