RxJs Combine Observables

Question

I'm trying to solve problem when i have two Observables which I want to combine into single value in Angular2 and Angularfire2

lets imagine I have bank account and transactions of two types, outgoing and incoming (its just use case i created for purpose of practice). And I want to sum them together to get actual account balance.

Template:

....
 <div *ngFor="let account of accounts | async">
    {{ (transactionService.getAccountBalance(account) | async )?.balance }}
 </div
 ....

and finally the service which should provide the data

    getAccountBalance(account){
    return Observable.zip(
        this.getIncomingTransactionsForAccount(account), // FirebaseListObservable
        this.getOutcomingTransactionsForAccount(account) // FirebaseListObservable
    , (incoming, outcoming) => {
            ....
            let result = {
                account: account.$key,
                balance: someCalculatedNumber
            };
            console.log(result); // result is correct there
            return result;
        }
    )

}

This one causes infinite loop on Firebase (and doesn't display data in template). I also tried Observer.merge() and some others but I think I'm just trying to use absolutely wrong approach.

Appreciate any help, thanks!

Edit:

just tried another solution, use just one stream to count all incoming transactions:

public getAllIncoming(account: IAccount){

    return this.getIncomingTransactionsForAccount(account)
        .scan((x, y) => {
            console.log(x, y);
            return 1;
        }, 0);
}

generates infinite queries again.

Edit2:

i tried also forkJoin

    public getAccount2Balance(account: IAccount) {

    return Observable.forkJoin([
            this.getIncomingTransactionsForAccount(account),
            this.getOutcomingTransactionsForAccount(account)
        ], (incoming, outcoming) => {
            ...
            let result = {
                account: account.$key,
                balance: someCalculatedNumber
            };
            console.log(result);
            return result;
        })
}

with same template as above. result: value is not displayed in template and console.log with result is NOT printed

yes, i did. I wasn't successful with that too. As said, i was just getting errors or infinite API calls. :( — Michal Hatak
What error it throws? Operator Observable.forkJoin() should do what you need. — martin
Updated question with forkJoin - in this implementation its not throwing any error, but it does not return/process value either. — Michal Hatak

Caleb Caleb · Accepted Answer · 2016-09-30T06:57:55

Another option is to try combineLastest. Below is assuming your incoming and outgoing account calls are returning a FirebaseListObservable and you are trying to combine them into one list (accounts).

return Observable.combineLatest(

  // added .do() to make sure incoming is working
  this.getIncomingTransactionsForAccount(account)
    .do(x => console.log('incoming transactions', x),

  // added .do() to make sure outcoming is working
  this.getOutcomingTransactionsForAccount(account)
    .do(y => console.log('outcoming transactions', y),

  (incoming, outcoming) => {
    // ... do your calculations
    let result = {
      // results
    };
    return result;
  }

);

It might be good to add temporary .do() side effects to each input streams to see if they are being called and returning results in the first place, especially since the combine block doesn't seem to get called in your previous examples.

RxJs Combine Observables

1 Answers