rxjs - 合并 Observables 数组的结果

Question

我需要从我的州获取 ID 作为单独的 Observable s, 将它们合二为一 Observable和 subscribe()到它。
我需要我的订阅者最终会得到一个已解析的数组 Observable结果。

另外我需要保持这个订阅打开，所以如果我内心的一个 Observable s 更改我的订阅者将收到通知。

这是我的代码:

getTransactionsByIDs(transactionsIDs){
return Observable.of(transactionIDs
  .map(transactionID => this.getTransactionByID(transactionID)));
}

this.transactionsService.getTransactionsByIDs(transactionsIDs)
.subscribe(transactions=>{
 ....
})

订阅者函数中上述代码的结果是一个未解析的存储数组。

如何解决每个商店并将它们组合在一起？

我也试过使用 Observable.from()在 transactionsID s 将每个 ID 转换为 Observable然后解决它。它工作正常，但是我的订阅者会分别收到每个 ID 的通知。如果有办法批量处理所有 Observable.from()结果(同时保持订阅开放)，请告诉我。

这是我的 Observable.from()看起来像:

getTransactionsByIDs(transactionsIDs){
return transactionIDs
  .mergeMap(transactionID => this.getTransactionByID(transactionID));
}

this.transactionsService.getTransactionsByIDs(Observable.from(transactionsIDs))
.subscribe(transactions=>{
 ....
})

Answer 1

我想你想要的是combineLatest .在所有内部 observable 发出至少一个值之前，它不会发出任何值。之后，每次从内部 observable 之一发出新的发射时，它都会发射最新的。

这是一些阅读 Material :https://www.learnrxjs.io/operators/combination/combinelatest.html

下面是一个例子:

function getTransactionByID(transactionId) {
  let count = 0;
  return Rx.Observable.of(transactionId)
    .delay(Math.random() * 4000)
    .map(x => `${x}: ${count++} `)
    .repeat();
}

function getTransactionsByIDs(transactionsIDs){
  return Rx.Observable.combineLatest(transactionsIDs.map(transactionID => getTransactionByID(transactionID)));
}

const transactionsIDs = [1,2,3];
getTransactionsByIDs(transactionsIDs)
  .take(10)
  .subscribe(x => { console.log(x); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.min.js"></script>

take(10)只是防止这个例子永远持续下去。