javascript - Cómo unir dos observables anidados de forma recursiva

Tengo arreglos observables, parents$ y children$ como se define a continuación. Luego tengo una clase AttachChildren. La clase tiene un método getChild$ (parentId) que se usa para obtener un hijo relevante para un padre dado. Finalmente, el método tesCollection() se utiliza para poblar la propiedad 'children' de los elementos parent$. He intentado usar forkJoin para intentar fusionar los dos flujos de manera que pueda tener los datos finales, pero hasta ahora no he tenido éxito.

Aquí están los datos relevantes y la clase. Estoy usando Typescript en node.js y rxjs versión "^7.3.0".

const parents$ = of([
    {
        parentName: 'a',
        parentId: 0,
        children: []
    },
    {
        parentName: 'b',
        parentId: 1,
        children: []
    },
    {
        parentName: 'c',
        parentId: 2,
        children: []
    },
]);

const children$ = of([
    {
        parentId: 0,
        childName: 'a-a',
        childData: [
            {
                itemId: 0,
                itemName: 'd'
            },
            {
                itemId: 1,
                itemName: 'e'
            }
        ]
    },
    {
        parentId: 2,
        childName: 'a-b',
        childData: [
            {
                itemId: 0,
                itemName: 'f'
            },
            {
                itemId: 1,
                itemName: 'g'
            }
        ]
    }
]);

Necesito que los niños se adjunten a los padres a través de la propiedad parentId relevante que es común tanto a parent$ como a children$.

Lo que he intentado es esto:

class AttachChildren {

   tesCollection() {
        parents$
            .pipe(
                mergeMap((m: Parent) => {
                    return m.map(parent => {
                        const child$ = this.getChild$(parent.parentId);
                        return forkJoin(
                            {
                                parent: of(parent),
                                child: child$
                            }
                        )
                    })
                })
            )
            .subscribe((p: any) => {
                console.log('subscribe/parents:', p);
            });

    }

    getChild$(parentId: number): Observable<Child> {
        return children$
            .pipe(
                mergeMap(c => c.map(child => child.parentId === parentId))
            )
    }
}

El resultado que estoy obteniendo es el siguiente:

subscribe/parents: Observable { _subscribe: [Function] }
subscribe/parents: Observable { _subscribe: [Function] }
subscribe/parents: Observable { _subscribe: [Function] }

¿Cómo puedo conseguir aplanar este resultado?

Respuesta #1

No estoy seguro de por qué necesitas usar RxJS de manera tan enrevesada, de una forma muy imperativa. Prefiero usarlo de una manera más funcional.

Aquí hay un fragmento que asocia padres e hijos. Incluso si esto no es exactamente la respuesta que buscas, espero que te dé una idea de lo mucho más simple (de leer y mantener) y menos imperativo que podría ser el código.

forkJoin({
  parents: parents$,
  children: children$,
}).pipe(
  map(({children, parents}) => {
    const getChildrenOfParent = id => {
      return children.filter(child => child.parentId === id)
    }

    return parents.map(parent => ({
        ...parent,
        children: getChildrenOfParent(parent.parentId)
    }));
  })
).subscribe(console.log)

La salida es:

Estoy usando forkJoin aquí, ya que este es un creador de Observable que emitirá cuando todos los flujos hayan completado, lo cual es perfectamente correcto para combinar múltiples of como en este ejemplo. También podrías usar combineLatest en el mundo real, si esperas que padres e hijos lleguen en momentos aleatorios. Esto comenzará a emitir una vez que cada flujo haya emitido al menos una vez.