Concat Observables

Question

I have some function to work with DB:

setupData(param) {
    return Observable.create((observer) => {
        this.db.executeSql('SELECT .... ?', param).then(() => {
            console.log('DB QUERY DONE');
            observer.complete();
        }, (error) => {
            observer.error(error);
        });
    });
}

And inside a loop in another function I need to run these observables sequentially. I do it like this:

processData() {

    ...

    let observers = [];
    storage.forEach((data) => {
        observers.push(this.setupData(data));
    });

    Observable.concat(observers).subscribe(() => {
        console.log('NEXT');
    }, () => {
        console.log('ERROR');
    }, () => {
        console.log('COMPLETE');
    });
}

So my output:

NEXT
NEXT
NEXT
...
NEXT
NEXT
COMPLETE

but I never seen "DB QUERY DONE". There is no subscribe on each observable. If I replace concat with forkJoin - I see that I expect, but I need to run SEQUENTIALLY, not parallel... Is there any solution?

Answer 1

What you actually need is concatMap. You can perform this operator on an observable and what it will do is.

1. Take the next value.

2. Perform a function on this value which should return an observable.

3. Start listening to this observable and send the values from this observable to next.

4. Once this observable has stopped emitting, take the next value form the source observable and repeat.

Code wise this looks like this:

const processData = () => {
    let storageIds$ = Rx.Observable.from(["1", "2", "3"]);

    storageIds$.concatMap((val) => {
      return setupData(val);
    }).subscribe(() => {
        console.log('NEXT');
    }, () => {
        console.log('ERROR');
    }, () => {
        console.log('COMPLETE');
    });
}

You can see that I have created an observable storageIds$from numbers 1, 2, 3. Then I perform a concatMap over this observable. I just call the setupData method which returns an observable, and return that immediately. This will provide you with the expected behaviour.

Full jsbin example can be found here: http://jsbin.com/katame/edit?js,console