как использовать многоадресную рассылку сообщений с поведением объекта?

Question

В общем нам нужна функциональность предмета поведения. Но только при первой подписке мы должны отправить подписку на сервер в REST. И чтобы отправить отказ от подписки при последнем отказе от подписки, и все поздние подписавшиеся наблюдатели получат gwt самое последнее json, полученное с первого. Могу ли я сделать это с помощью операторов rx js и как? или я использую пользовательский 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