Как получить наблюдаемые возвращаемые значения один за другим

Question

Существует ли функция rxjs / другой способ сделать это (где, мы надеемся, существует функция):

    let testOb$ = of(4).pipe(
      <???>(input => of(input + 2)),
      <???>(input => of(input - 3))
    )
    testOb$.subscribe(x => console.log(x))

    // logs 4, then 6, then 3
    // switchMap and mergeMap just returns 3

Обратите внимание, что значения должны приходить один за другим, а не заполнять все сразу, а затем записать 3 значения отдельно.

Вариант использования:

Я хочу загрузить несколько проблемных data, состоящих из массива объектов.

Каждый из объектов имеет одно значение, которое слишком велико и, если загружено все вместе, вызывает значительное отставание. Но до тех пор, пока кто-то загружается первым, задержка пользовательского интерфейса отсутствует.

Итак, я хотел бы:

data$ = this._store.dispatch(fetchInitial())
    .pipe(
        <???>(data => fillInitial(data))
        <???>(data => fillRemainder(data))
    )
data$.subscribe(data => renderInput(data))

Это правильный подход?

Если нет, то что считается наилучшей практикой для этого вида использования с rxjs?

Answer 1

Создайте 3 отдельных потока и объедините их, когда вам нужно.

   let testOb$ = of(4).pipe(share());

   const six$ = testOb$.pipe(mergeMap(input => of(input + 2)));
   const three$ = testOb$.pipe(mergeMap((input => of(input - 3)));

   merge(testOb$, six$, three$).subscribe(x => console.log(x))

Answer 2

Ваш желаемый результат может быть достигнут следующим образом:

let testOb$ = of(4).pipe(
  mergeMap(input => from([input, input + 2])),
  mergeMap(input => from([input, input - 3])),
  filter((val, index) => index != 1)
);
testOb$.subscribe(x => console.log(x))

Это выглядит странно, потому что ваш желаемый результат на самом деле не следует простому правилу. Но это работает.

https://stackblitz.com/edit/rxjs-cmxgan?devtoolsheight=60

Теперь, для вашего второго примера, мы бы сделали что-то подобное, чтобы избежать задержки:

data$ = this._store.dispatch(fetchInitial())
  .pipe(
    mergeMap(data => fillInitial(data).pipe(startWith(data)))
    mergeMap(data => fillRemainder(data).pipe(startWith(data))),
    filter((val, index) => index != 1)
)