在 Observables 中扁平化 Observable

Question

我需要合并两个 Obersvables。

products
 ---1
    ---name: 'product 1'
    ---supplierId: '1'
 ---2
    ---name: 'product 2'
    ---supplierId: '2'

suppliers
 ---1
  ---name: 'supplier 1',
  ---contact: 'contact 1',     
 ---2
  ---name: 'supplier 2',
  ---contact: 'contact2'

最后我想收到以下内容：

[
   {
     name: 'product 1'
     supplierId: '1',
     supplier: { name: 'supplier 1', contact: 'contact1'}
   },
   {
     name: 'product ´2'
     supplierId: '2',
     supplier: { name: 'supplier 2', contact: 'contact2'}
   },
]

目前我有以下代码：

this.af.list('products')
  .map(products => {
    let items = [];
    (<Array<any>>products).forEach(p => {
      if (p.supplierId) {
        let supplier$ = this.af.object(`suppliers/${p.supplierId}`).take(1);

        items.push(Observable.forkJoin(Observable.of(p), supplier$, (p1, supplier) => {
          return {
            name: p1.name,
            supplier: supplier
          };
        }))
      }
    })
    return items;
  })
  .do(res => console.log('After Map', res))
  .flatMap(res => Observable.combineLatest(Observable.of(res)))
  .subscribe(res => console.log('Final Response', res))

不管我尝试什么，我要么一无所获，要么收到一个 ForkJoinObservables 数组。

Answer 1

无需使用flatMap 或combineLatest。

我制作了一个小的Plunkr 来演示如何做到这一点。

首先，我不想放太多与问题无关的代码，我在没有 Angular 的情况下构建了我的示例，并使用了模拟：

模拟来自后端的数据：

// raw objects that simulate the backend
const products = [
  {
    name: 'product 1',
    supplierId: '1'
  },
  {
    name: 'product 2',
    supplierId: '2'
  }

];

const suppliers = [
  {
    name: 'supplier 1',
    contact: 'contact 1'
  },
  {
    name: 'supplier 2',
    contact: 'contact2'
  }
];

Mock 以模拟您的 Angular 服务（返回一个 Observable）：

// simulate the angular service with HTTP
// use Observable.of and not Observable.from because we get the product list in one reponse
const getProducts = () => {
  return Observable.of(products).delay(1000);
};

const getSupplierByName = (name) => {
  const supplier = suppliers.find(s => s.name === `supplier ${name}`);

  return Observable.of(supplier).delay(1000);
};

最后，按预期检索数据的代码：

// now let's try to get your data as you want, which is : 
// get the whole product list
// for each product, get his supplier
// merge the data into one big result
getProducts()
  .switchMap(products => 
    Observable
      .forkJoin(products
        .map(product => getSupplierByName(product.supplierId)
          .map(supplier => (Object.assign({}, product, { supplier })))))
  )
  .do(console.log)
  .subscribe();

这是我们在控制台中得到的结果：

Answer 2

使用扫描运算符。将产品、供应商和结果保存在累加器中，并尽可能加入。当您能够创建结果对象时，您可能还需要清理产品和供应商列表（我没有这样做）

扫描：

在源 Observable 上应用累加器函数，并返回每个中间结果，并带有可选的种子值。

http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-scan

  let o1$ = Rx.Observable.of({
      name: 'product 1',
      supplierId: '1'
    },
    {
      name: 'product 2',
      supplierId: '2'
    });

  let o2$ = Rx.Observable.of({
      name: 'supplier 1',
      contact: 'contact 1'
    },
    {
      name: 'supplier 2',
      contact: 'contact 2'
    });

  Rx.Observable.merge(o1$, o2$)
    .scan((acc, el) => {
        // product
        if (el.supplierId) {
          // check in supplier queue
          let supplier =
            acc.suppliers.find(supplier => supplier.name === ('supplier ' + el.supplierId));
          if (supplier) {
            acc.results.push({
              name: el.name,
              supplierId: el.supplierId,
              supplier: {name: supplier.name, contact: supplier.contact}
            });
          }
          else {
            acc.products.push(el);
          }
        }
        // supplier
        else {
          // check in product queue
          let product =
            acc.products.find(product => ('supplier ' + product.supplierId) === el.name);
          if (product) {
            acc.results.push({
              name: product.name,
              supplierId: product.supplierId,
              supplier: {name: el.name, contact: el.contact}
            });
          }
          else {
            acc.suppliers.push(el);
          }
        }
        return acc;
      },
      {suppliers: [], products: [], results: []}
    )
    .subscribe(({results}) => console.log(results));