Как объединить наблюдаемые неблокирующим образом?

Question

У меня есть коллекция Observables, которые получают разные типы данных каждый.Я соединяю эти Observable, чтобы получить все данные, которые я хочу для своей страницы.Факт в том, что вся эта информация независима, поэтому загрузка одного не должна блокировать или мешать загрузке других.чего я не могу достичь.

Вот пример того, что я сделал на данный момент:

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        combineLatest([
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([])),
        ])
      )
    );
  }

Здесь наблюдаемые успешно соединены в цепочку, поэтому я получаю вседанные, которые я хочу, но combineLatest() выполняет следующий запрос после завершения предыдущего, а родительский объект отправляет собранные данные после завершения всего, что создает задержку.Я пытался использовать merge() вместо combineLatest(), чтобы все полученные данные мгновенно отправляли их родительскому объекту Observable, но, очевидно, я просто не могу этого сделать

Итак, как мне это сделать?Цепочка этих Observables так, чтобы каждый подчиненный в дочернем Observable излучал непосредственно родительскому Observable?

Answer 1

Вам нужно forkJoin, он возвращает Array с результатами каждого Observable (в том же порядке, что и forkJoin параметры.

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([]))
        )
      )
    );
  }

Answer 2

Если они независимы, вы можете обращаться с ними независимо и выдавать их по теме.

  getAll(): Observable<any> {
    const subject = new Subject<any>();
    const handler = (res: any) => subject.next(res);

    this.getHome().pipe(
      tap((home: Home) => this.application = home),
      tap(handler),
      tap((home: Home) => {
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([]),tap(handler))
          this.getStyling(home.styling).pipe(defaultIfEmpty({}), tap(handler))                  
          this.getCategories(home.categories).pipe(defaultIfEmpty([]), tap(handler))
        ).subscribe(res => subject.complete())
      })
    ).subscribe();
    return subject.asObservable();
  }

При использовании этого метода проверка типа должна выполняться потребителем.

Предметы документированы здесь .

Answer 3

Я пытался использовать метод merge () вместо функцииlateLatest (), чтобы все полученные данные мгновенно отправляли его родительскому объекту Observable, но, очевидно, я просто не могу этого сделать

Похоже, вы ищете частичные данные, как только они будут готовы.И combLatest () , и forkJoin () будут выдавать данные только тогда, когда они могут заполнять массив равной длины с количеством наблюдаемых.Таким образом, вы либо ждете хотя бы одно значение в каждой наблюдаемой, либо ждете их завершения, но они излучают массив, и им нужно знать, что в него вставить.

Я думаю, что вы хотите испустить каждый кусокданных, как только они будут готовы.

Начните с использования merge () для выдачи значений из всех 3 наблюдаемых, но для каждой наблюдаемой используйте map () для созданияпара ключ / значение объекта для данного типа (изображения, стили, категории).

Наконец, используйте оператор scan () , чтобы объединить значения в один объект, который будет излучать каждыйсвойство как оно обновлено.

Потребители потока могут проверить, является ли свойство null, чтобы увидеть, готово оно или нет.Поток передаст 3 значения до его завершения.

  getAll(): Observable<{home: Home, images: any[], styling: any[], categories: any[]}> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      switchMap((home: Home) =>
        merge([
          this.getAssets(home.images).pipe(
             defaultIfEmpty([]),
             map(images => ({images}))
          ),
          this.getStyling(home.styling).pipe(
             defaultIfEmpty({}),
             map(styling => ({styling}))
          ),
          this.getCategories(home.categories).pipe(
             defaultIfEmpty([]),
             map(categories => ({categories})
          ),
        ]),
        scan((acc, next) => ({...acc, ...next}), {home, images: null, styling: null, categories: null})
      )
    );
  }

Вы можете добавить оператор startWith({}) к оператору merge(), если хотите создать первое значение, где все свойства равны null.Как раннее значение для потребителей, чтобы они знали, что чтение для других значений началось.