RxJs 合并可观察对象

Question

我正在尝试解决一个问题，当我有两个可观察对象时，我想将它们合并为单个值在Angular2和Angularfire2中

假设我有银行账户和两种类型的交易：支出和收入（这只是我为实践目的创建的用例）。 我想将它们相加以获得实际账户余额。

模板：

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

最后是提供数据的服务

    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;
        }
    )

这个问题会在Firebase上引起无限循环（并且不会在模板中显示数据）。我也尝试了Observer.merge()和其他一些方法，但我认为我只是在尝试使用完全错误的方法。

非常感谢任何帮助，谢谢！

编辑：

刚刚尝试了另一个解决方案，只使用一个流来计算所有传入的交易：

public getAllIncoming(account: IAccount){

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

生成无限查询。 编辑2： 我也尝试了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;
        })
}

使用与上面相同的模板。结果： 数值在模板和控制台中未显示，并且未打印console.log的结果。

- Michal Hatak

---

你尝试过 Observable.forkJoin([obs1, obs2, ...]) 吗？ - Brad

是的，我试过了。但我没有成功。就像我说的那样，我一直收到错误消息或无限的API调用。:( - Michal Hatak

它会抛出什么错误？操作符 Observable.forkJoin() 应该能满足你的需求。 - martin

使用forkJoin更新问题 - 在这个实现中它没有抛出任何错误，但也没有返回/处理值。 - Michal Hatak

Answer 1

另一种选择是尝试使用combineLastest。假设您的传入和传出帐户调用都返回FirebaseListObservable，并且您正在尝试将它们合并为一个列表（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;
  }

);

可能很有必要给每个输入流添加临时的 .do() 副作用，以查看它们是否被调用并返回结果，尤其是由于在您先前的示例中似乎没有调用组合块。

- Caleb