javascript - 根据特定条件加入 RxJs 中的两个可观察流

Question

我有两个对象流，账户和余额。

我需要根据id和account_id合并(加入)两个流

var accounts = Rx.Observable.from([
    { id: 1, name: 'account 1' },
    { id: 2, name: 'account 2' },
    { id: 3, name: 'account 3' },
]);

var balances = Rx.Observable.from([
    { account_id: 1, balance: 100 },
    { account_id: 2, balance: 200 },
    { account_id: 3, balance: 300 },
]);

预期的结果:

var results = [
    { id: 1, name: 'account 1', balance: 100},
    { id: 2, name: 'account 2', balance: 200},
    { id: 3, name: 'account 3', balance: 300},
];

这对 RxJs 可行吗？

明确地说，我知道如何使用纯 js/lodash 或类似的东西来做到这一点。在我的例子中，我从 Angular Http 模块获取这些流，所以我想问在这种情况下我是否可以从 RxJs 中受益

Answer 1

根据您的一条评论，您的示例是模拟来自 Angular Http 调用的流。

所以代替:

var accounts = Rx.Observable.from([
    { id: 1, name: 'account 1' },
    { id: 2, name: 'account 2' },
    { id: 3, name: 'account 3' },
]);

var balances = Rx.Observable.from([
    { account_id: 1, balance: 100 },
    { account_id: 2, balance: 200 },
    { account_id: 3, balance: 300 },
]);

我宁愿说它是:

var accounts = Rx.Observable.of([
    { id: 1, name: 'account 1' },
    { id: 2, name: 'account 2' },
    { id: 3, name: 'account 3' },
]);

var balances = Rx.Observable.of([
    { account_id: 1, balance: 100 },
    { account_id: 2, balance: 200 },
    { account_id: 3, balance: 300 },
]);

为什么: from 将逐一发出每个项目，of 将发出整个数组，我猜你的 http 响应是整个数组。

也就是说，您可能想要实现的是:

const { Observable } = Rx;

// simulate HTTP requests
const accounts$ = Rx.Observable.of([
  { id: 1, name: 'account 1' },
  { id: 2, name: 'account 2' },
  { id: 3, name: 'account 3' }
]);

const balances$ = Rx.Observable.of([
  { account_id: 1, balance: 100 },
  { account_id: 2, balance: 200 },
  { account_id: 3, balance: 300 }
]);

// utils
const joinArrays = (accounts, balances) =>
  accounts
    .map(account => Object.assign({}, account, { balance: findBalanceByAccountId(balances, account.id).balance }));

const findBalanceByAccountId = (balances, id) =>
  balances.find(balance => balance.account_id === id) || { balance: 0 };

const print = (obj) => JSON.stringify(obj, null, 2)

// use forkJoin to start both observables at the same time and not wait between every request
Observable
  .forkJoin(accounts$, balances$)
  .map(([accounts, balances]) => joinArrays(accounts, balances))
  .do(rslt => console.log(print(rslt)))
  .subscribe();

输出:

[
  {
    "id": 1,
    "name": "account 1",
    "balance": 100
  },
  {
    "id": 2,
    "name": "account 2",
    "balance": 200
  },
  {
    "id": 3,
    "name": "account 3",
    "balance": 300
  }
]

这是一个可用的 Plunkr:https://plnkr.co/edit/bc0YHrISu3FT45ftIFwz?p=preview

编辑 1:在一个数组上工作来组成你的结果可能不是性能的最佳主意，而不是返回一个数组，也许尝试返回一个对象，该对象具有作为键的帐户 ID。通过这种方式，您可以简单地删除 findBalanceByAccountId 函数并拥有更快的应用程序(此处仅修改代码):

const balances$ = Rx.Observable.of({
  1: { account_id: 1, balance: 100 },
  2: { account_id: 2, balance: 200 },
  3: { account_id: 3, balance: 300 }
});

// utils
const joinArrays = (accounts, balances) =>
  accounts
    .map(account => Object.assign(
      {}, 
      account, 
      { balance: balances[account.id].balance }
    ));