Угловая наблюдаемая цепь Rxjs с магазином ngrx

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);
}));

С помощью нажатия кнопки пользователь может изменить оба значения sushi, что вызывает последний вызов подписки дважды, что имеет смысл (RxJS).Я мог бы удалить switchMap и написать что-то вроде

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

Я не очень большой поклонник этого, я уверен, что должно быть rxjs/operator вроде.Может ли кто-нибудь сделать предложение?

Кроме того, пробовал forkjoin, но с хранилищем ngrx кажется, что нужно вызвать first или last, как показано ниже.Вот ссылка на приведенный выше оператор forkjoinWithstore

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

Если я воспользуюсь приведенным выше шаблоном, подписки будут срабатывать в первый раз, но не после этого вообще.

Answer 1

Прежде всего, вот рабочий пример для 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 и проверите консоль, вы увидите следующее поведение:

Если мы используем объединенные последние, мы объединяем наблюдаемые отдельно и заботимся только о последнем состоянии, что приводит к 2 вызовам console.log.

Мы могли бы вместо этого использоватьzip, который ожидает испускания обеих наблюдаемых, прежде чем выдавать результат.Это выглядит как идеально подходящее для нашей кнопки «Увеличить оба», но есть проблема: если starbucksSushi увеличивается отдельно (возможно, из другой части приложения), версия zipped будет ждать суши на заправкечтобы получить обновление тоже.

Чтобы предложить третье решение, вы можете использовать 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);

Это будет реагировать на изменения во всех источниках, но не чаще, чем 100ms.

Наконец, причина, по которой вам пришлось сделать first():

forkJoin присоединяется к наблюдаемым после их завершения (что может произойти только один раз, поэтому он не подходит для непрерывного потока) и больше подходит для работы, подобной обещанию, например, HTTP-вызовы, завершение процесса и т. Д., если вы берете только один элемент из потока, результирующий поток завершается после одного выброса.

PS

Я рекомендую использовать async pipe дляr работает с наблюдаемыми (как я сделал со свойствами

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

и внутри шаблона, затем

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

вместо

result => this.zipped = result

, который я использовалоба в этом примере, чтобы вы могли сравнить их.Исходя из моего опыта, работа с наблюдаемыми становится намного проще, если вы прекратите преобразование «отмены наблюдения» их заранее и просто дадите трубе async выполнять свою работу.

Кроме того, если выиспользуйте subscribe где-нибудь в вашем компоненте, вы должны unsubscribe, когда компонент уничтожен - что совсем не сложно, но если мы никогда не подписываемся явно, и разрешаем каналу async подписаться, он также обрабатываетуничтожение для нас:)