rxjs/Observable: Running a function once after getting first stream (continuous observable)

Question

First off, sorry for the long title.

I am trying to subscribe to an array of continuous streams with forEach from angularfire2, but I would also like to run a function after I have confirmed that the first set of data has come in:

this.people.forEach((person) => {
    person.items = this.database.list('/items' + person.key);
    person.items.subscribe((data) => {person.itemsList = data});
});

myIntendedFunction();

Is there a way to place myIntendedFunction() such that:

1. It runs after the first stream of data is received for each person, and
2. It runs only once?

Answer 1

This is a little more complicated to accomplish if you don't want the first request to happen twice:

const connectables: ConnectableObservable<any>[] = [];

this.people.forEach(person => {
    person.items = this.database.list('/items' + person.key);
    const connectable = person.items.publish();
    connectables.push(connectable);
    connectable.subscribe((data) => {person.itemsList = data});
});

Observable.zip(...connectables).take(1).subscribe(myIntendedFunction);

connectables.forEach(c => c.connect());

What happens here is this: The effect of publish() is basically that you can subscribe to the same stream of data multiple times. Also the subscription function won't be invoked until you call connect(). If we used person.items.share() which is sugar for person.items.publish().connect(), it would make the requests right away and our application might be buggy due to race-conditions.

zip() waits for every passed observable to emit an item and emits those items all at once as an array. We only want this to happen on the first set of items, so we just take(1).