RxJS queueing dependent tasks

Question

If I have an array of arrays like this

{
    parent: [
        {
            name: 'stu',
            children: [
                {name: 'bob'},
                {name: 'sarah'}    
            ]
        },
        { 
          ...
        }
    ]
}

and I want to cycle through each parent and cycle through their children in series, so that I don't start the next parent until all the children have been processed (some long asynchronous process), how do I do this with RxJS?

I have attempted this:

var doChildren = function (parent) {
    console.log('process parent', parent.name);
    rx.Observable.fromArray(parent.children)
    .subscribe(function (child) {
        console.log('process child', child.name);
        // do some Asynchronous stuff in here
    });
};

rx.Observable.fromArray(parents)
.subscribe(doChildren);

But I get all the parents logging out and then all the children.

score 5 · Accepted Answer · answered Oct 23 '15 at 03:42

concatMap works better here. Because if iterating children is async, the order of child will be messed up. concatMap can ensure to finish one parent at a time.

Rx.Observable.from(parents)
  .concatMap(function (p) {
    return Rx.Observable.from(p.children)
  })
  .subscribe();

score 3 · Answer 2 · answered Jan 08 '15 at 17:20

It looks like this was asked a while ago, but here's one way to deal with this scenario:

Rx.Observable.fromArray(parents)
.flatMap(function(parent) {
  return parent.children;
})
.flatMap(function(child) {
  return doSomeAsyncThing(child); //returns a promise or observable
})
.subscribe(function(result) {
  // results of your async child things here.
});

The idea is to leverage flatMap which will take any Arrays, promises or observables returned and "flatten" them into an observable of individual things.

I think you might also benefit from flat mapping the results of your async things you're doing with the child nodes, so I've added that in there. Then you can just subscribe to the results.

I still feel like this question is lacking some context, but hopefully this is what you're looking for.

RxJS queueing dependent tasks

2 Answers2

Linked