Как использовать combLatest с фильтром в определенных наблюдаемых?

Question

Это упрощение сложного случая, когда некоторые наблюдаемые в массиве могут быть отфильтрованы, если значения недопустимы. Проблема в том, что отфильтрованная наблюдаемая не позволяет другому завершить объединение. Какой оператор или подход может обработать этот случай, разрешив действительный журнал данных в подписке?

// RxJS v6+
import { fromEvent, combineLatest, of } from 'rxjs';
import { mapTo, startWith, scan, tap, map, filter } from 'rxjs/operators';

const userData$ = [
   of({ name: 'Joseph', age: 23}), 
   of({ name: 'Mario', age: 33}), 
   of({ name: 'Robert', age: 24}), 
   of({ name: 'Alonso', age: 25})
];

const names = ['Joseph', 'Mario', 'Robert', 'Alonso'];

combineLatest(
  names.map((name, i) => {
     return userData$[i].pipe(
         map(({name, age})=> { return{ name, age: age * 2} }),
         filter(({age}) => age < 67),
         map(({name, age})=> { return{ name: name.toLocaleUpperCase(), age} }),
     )
 })
)
   .pipe(
     tap(console.log),
   )
   .subscribe();

Образец в stackblitz

Если мы изменим значение на 67, то все наблюдаемые покажут данные.

Answer 1

Типичная проблема с combineLatest состоит в том, что все исходные наблюдаемые должны излучать хотя бы один раз, поэтому если вы используете filter, чтобы отбросить единственное значение, то combineLatest никогда не будет излучать ничего.

An простое решение - убедиться, что оно всегда излучается с defaultIfEmpty:

combineLatest(
  names.map((name, i) => {
    return userData$[i].pipe(
      map(({name, age})=> { return { name, age: age * 2} }),
      filter(({age}) => age < 66),
      map(({name, age})=> { return { name: name.toLocaleUpperCase(), age} }),
      defaultIfEmpty(null),
    )
  })
)

Live demo: https://stackblitz.com/edit/typescript-rsffbs?file=index.ts

Если ваш реальный пример использования использует другой источник Наблюдаемый, чем of(), который не завершается немедленно, вы можете вместо этого использовать startWith.

Answer 2

Вы можете заменить combineLatest на from, так как combineLatest не будет излучать, если один из элементов в массиве потоков не будет излучать

const userData$ = [
  { name: 'Joseph', age: 23 },
  { name: 'Mario', age: 33 },
  { name: 'Robert', age: 24 },
  { name: 'Alonso', age: 25 }
];

const names = ['Joseph', 'Mario', 'Robert', 'Alonso'];

from(
  userData$
)
  .pipe(
    map(({ name, age }) => { return { name, age: age * 2 } }),
    filter(({ age }) => age < 66),
    map(({ name, age }) => { return { name: name.toLocaleUpperCase(), age } }),
    tap(console.log),
  )
  .subscribe();