Rxjs вложенная подписка с несколькими внутренними подписками

Question

Исходный код на основе обещаний, который я пытаюсь переписать:

parentPromise
    .then((parentResult) => {
        childPromise1
            .then(child1Result => child1Handler(parentResult, child1Result));
        childPromise2
            .then(child1Result => child2Handler(parentResult, child2Result));
        childPromise3
            .then(child1Result => child3Handler(parentResult, child3Result));
    });   

Я пытаюсь найти способ избежать использования шаблона вложенных подписок в следующем сценарии. :

parent$
    .pipe(takeUntil(onDestroy$))
    .subscribe((parentResult) => {
        child1$
            .pipe(takeUntil(onDestroy$))
            .subscribe(child1Result => child1Handler(parentResult, child1Result));
        child2$
            .pipe(takeUntil(onDestroy$))
            .subscribe(child2Result => child2Handler(parentResult, child2Result));
        child3$
            .pipe(takeUntil(onDestroy$))
            .subscribe(child3Result => child3Handler(parentResult, child3Result));
    });

Каким будет правильный способ RxJS сделать это?

Answer 1

Это кажется довольно странным для меня. Вы создаете новую подписку для каждого ребенка каждый раз, когда приходит parentResult. Даже если те в конечном итоге действительно будут уничтожены (при условии, что реализация onDestroy$ верна), это кажется неправильным.

Вы, вероятно, хотите withLatestFrom(parent$) и три отдельных канала для каждого дочернего элемента.

Это может выглядетьчто-то вроде: child1$.pipe(takeUntil(globalDeath$), withLatestFrom(parent$)).subscribe(([childResult, parentResult]) => ...). Не уверен, что мой JS правильный, не могу проверить его в данный момент;но дело в том, что вы получаете последний результат от parent$ каждый раз, когда child1$ срабатывает. Обратите внимание, что при необходимости вы можете изменить направление (withLatestFrom(child1$)).

Answer 2

Если вы использовали Обещания, то соответствующие Наблюдаемые объекты излучают только один раз, а затем завершают.

В этом случае вы можете использовать forkJoin для параллельного выполнения дочерних Наблюдаемых.

Таким образом, код может выглядеть как

parent$.pipe(
   takeUntil(onDestroy$),
   // wait for parent$ to emit and then move on
   // the following forkJoin executes the child observables in parallel and emit when all children complete - the value emitted is an array with the 3 notifications coming from the child observables
   concatMap(parentResult => forkJoin(child1$, child2$, child3$)).pipe(
      // map returns both the parent and the children notificiations
      map(childrenResults => ({parentResult, childrenResults})
   )
).subscribe(
  ({parentResult, childrenResults}) => {
      child1Handler(parentResult, childrenResults[0]);
      child1Handler(parentResult, childrenResults[1]);
      child1Handler(parentResult, childrenResults[2]);
  }
)

Answer 3

Как насчет использования наблюдаемых более высокого порядка? Как то так:

const parentReplay$ = parent$.pipe(shareReplay(1));

of(
  [child1$, child1Handler],
  [child2$, child2Handler],
  [child3$, child3Handler]
).pipe(
  mergeMap([child$, handler] => parentReplay$.pipe(
    mergeMap(parentResult => child$.pipe(
      tap(childResult => handler(parentResult, childResult))
    )
  )
).subscribe();

Answer 4

Вы можете: 1) передать parent$ через share, и 2) использовать flatMap три раза, что-то вроде:

const sharedParent$ = parent$.pipe(share());

sharedParent$.pipe(
    flatMap(parentResult => forkJoin(of(parentResult), child1$)), 
    takeUntil(onDestroy$)),
.subscribe((results) => child1Handler(...results)); // repeat for all children

(Если есть более 2 детей, извлеките это вхорошая функция с дочерним потоком и обработчиком в качестве параметров).

Это соответствует исходному поведению ожидания с подписывающимися дочерними элементами до тех пор, пока parent$ не выдаст. Если вам это не нужно, вы можете пропустить flatMap и просто forkJoin sharedParent$ и детей.