如何使用 Angular/RXJS 对嵌套的可观察数组的内部属性求和？

Question

我无法在另一个 Observable 中获取 Observable 的内部数字属性的总和（或任何减少）。

我有一个可观察的“帐户”对象数组 (Observable<AppAccount[]>)。

export interface AppAccount {
    _id?: string;
    name: string;
}

还有一个“余额”对象的 Observable 数组，每个对象都有一个 accountId。许多余额可以与一个帐户相关联（按日期排序/过滤，但为简洁起见，该部分已删除）

export interface AccountBalance {
    _id?: string;
    accountId: string;
    amount: number;
}

我有一个帮助方法，它只返回给定帐户的最后一个余额对象的金额。

getLastAmount(account: AppAccount): Observable<number> {
    return this.balanceService.balances$.pipe(
      map(balances => {
        let last = balances.filter(balance => {
          return balance.accountId === account._id;
        }).sort().pop();
        //console.log(last)
        return last ? last.amount : 0;
      }),
      tap(amount => console.log(`getLastAmount() => ${amount}`)),
    );
  }

现在我正在尝试编写一个方法，该方法将遍历帐户，为每个帐户调用 getLastAmount()，然后将它们全部相加并返回一个 Observable。这是我到目前为止所管理的：

getTotalBalance(accounts$: Observable<AppAccount[]>): Observable<number> {
    return accounts$.pipe(
      map(accounts => from(accounts)),
      mergeAll(),
      mergeMap(account => this.getLastAmount(account)),
      reduce((sum, current) => {
        console.log(`${sum} + ${current}`);
        return sum + current;
      }, 0)
    );
  }

但这似乎永远不会返回，并陷入无限循环？？

只有一个帐户和一个余额关联，余额的“金额”为“10”，我从控制台日志中得到这个：“0 + 10”一遍又一遍，网络日志也确认它是连续调用 getBalances()。

我在正确的轨道上吗？有没有更好的办法？为什么这个 RXJS 管道会卡在一个循环中？

编辑：我根据 picci 的建议做了一些更改：

getTotalBalance(accounts$: Observable<AppAccount[]>): Observable<number> {
    return accounts$.pipe(
      map(accounts => accounts.map(account => this.getLastAmount(account))),
      concatMap(balances$ => { console.log('balances$', balances$); return forkJoin(balances$); }),
      tap(balances => console.log('balances', balances)),
      map(balances => balances.reduce(
        (amountSum, amount) => {
          console.log(`${amountSum} + ${amount}`)
          amountSum = amountSum + amount;
          return amountSum
        }, 0))
    );
  }

但这仍然没有返回，或者管道没有完成？ 我在这里做了一个stackblitz：https://stackblitz.com/edit/angular-rxjs-nested-obsv 如果您检查控制台输出，它似乎没有比 forkJoin 调用更进一步......

Answer 1

如果我理解正确，你可以这样进行

// somehow you start with the observable which returns the array of accounts
const accounts$: Observable<AppAccount[]> = getAccounts$()
// you also set the date you are interested in
const myDate: Moment = getDate()

// now you build the Observable<number> which will emit the sum of the last balance amounts
const amountSum$: Observable<number> = accounts$.pipe(
  // you transform an array of accounts in an array of Observable<number> representing the last Balance amount
  map((accounts: Account[]) => {
    // use the getLastAmount function you have coded
    return accounts.map(account => getLastAmount(account, myDate))
  }),
  // now we trigger the execution of the Observable in parallel using concatMap, which basically mean wait for the source Observable to complete
  // and forkJoin which actually executes the Observables in parallel
  concatMap(accounts$ => forkJoin(accounts$)),
  // now that we have an array of balances, we reduce them to the sum using the Array reduce method
  map(balances => balances.reduce(
    (amountSum, amount) => {
      amountSum = amountSum + amount;
      return amountSum
    }, 0)
  )
)

// eventually you subscribe to the amountSum$ Observable to get the result
amountSum$.subscribe({
  next: amountSum => console.log(`The sum of the last balances is: ${amountSum}`),
  error: console.err,
  complete: () => console.log("I am done")
})

可能还有其他组合可以产生相同的结果，但这似乎有效，可以在this stackblitz 中查看。

如果您对带有 http 调用的 RxJS 的一些常见模式感兴趣，您可能需要阅读this blog。

Answer 2

嗯 - 首先我认为你不应该使用这样的 observables。

如果你只需要totalBalance，你可以使用这样的东西（：

  private appAcount$ = from<AppAccount[]>([
    { _id: '1', name: 'user-1' },
    { _id: '2', name: 'user-2' },
    { _id: '3', name: 'user-3' },
  ]);

  // this would be your http call
  public getBalances(accountId: string): Observable<AccountBalance[]> {
    const ab = [
      { _id: '1', accountId: '1', amount: 100 },
      { _id: '2', accountId: '2', amount: 200 },
      { _id: '3', accountId: '2', amount: 300 },
      { _id: '4', accountId: '3', amount: 400 },
      { _id: '5', accountId: '3', amount: 500 },
      { _id: '6', accountId: '3', amount: 600 },
    ];

    return of(ab.filter(x => x.accountId === accountId));
  }

  lastAmounts$: Observable<AccountBalance[]> = this.appAcount$
    .pipe(
      switchMap(appAccount => 
        this.getBalances(appAccount._id)
          .pipe(
            // replace this with your date filter
            map(balances => [balances[balances.length - 1]]) 
          )
      ),
      scan((acc, c) => [ ...acc, ...c ])
    );

  totalBalance$ = this.lastAmounts$
    .pipe(
      map(x => x.reduce((p, c) => p += c.amount, 0))
    )

如果你只需要总余额，你可以订阅totalBalance$ observable。

不过，让我说，如果您可以批量获取您拥有的所有 appAccounts 的所有 AccountBalances，我不建议为每个 appAccount 执行 HTTP 调用 - 这样您就可以使用 combineLatest对于appAccounts$ 和balances$。