How to join two nested observables recurssively

Question

I have an observable arrays, parents$ and children$ as defined below. I then have a class AttachChildren. The class has a method getChild$(parentId) that is used to get a relevant child for a given parent. Lastly, the method tesCollection() is used to populate the 'children' property of parent$ items. I have tried to use forkJoin to try and merge the two streams so that I can have the final data but so far have not been successfull. Blow are the relevant data and the class. I am using typescript in node.js and rxjs version "^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'
            }
        ]
    }
]);

I need the children to attach to parents via the relevant parentId property that is common to both parent$ and children$; What I have tried is:

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 {
        return children$
            .pipe(
                mergeMap(c => c.map(child => child.parentId === parentId))
            )
    }
}

The result I am getting is as below:

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

How can I get to flatten this result?

user_776686 · Answer

I am not sure why you need to use so convoluted RxJS, in a very imperative way. I prefer to use it in a more functional way.

Here is a snippet that does associate parents and children. Even if this is not exactly the answer you are after, I hope it gives you a gist of how much simpler (to read and maintain) and less imperative the code could be.

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)

The output is:

I am using forkJoin here, as this is an Observable creator that will emit when all the streams have completed - which is perfectly fine for combining multiple of as in this example. You could also use combineLatest in real world, if you expect parents and children to arrive at random times. This will start emitting once every stream has emitted at least once.

How to join two nested observables recurssively

1 Answers