RxJava。加入来自两个Observable/Flowable流的所有项目

Question

我正在加载一些项目以在单个网页上显示。与此同时，我正在加载物品的价格。简化它看起来像这样：

Observable<Integer> ids = itemIdsToShow(); // COLD Observable 
Observable<Item> items = ids.flatMap(id -> loadingItem(id)); 
Observable<Price> prices = ids.flatMap(id -> loadingPrice(id)); 

更新：“物品”和“价格”都没有排序。

现在我想将它们结合在一起。

Observable<Long> wait = Observable.interval(1000, TimeUnit.SECONDS); 
Observable<Pair<Item, Price>> pairs = items.join(prices, (ii)->wait, (pp)->wait, Pair::of); 

Observable<Item2> items2 = pairs.filter(p->p.a.id == p.b.id).map(p->new Item2(p.a, p.b)); 

它有效，但Observable“wait”看起来很奇怪。或者，我可以使用任何从未完成的Observable。其实我需要一个可观察的，完成“项目”和“价格”的完成。

下一个建议不起作用。

Observable<Object> wait = items.mergeWith(prices).takeLast(1); 

法“加入”订阅这个新创建的再观察的，从而启动了新一代的ID序列的（一切从头开始）。

我很确定，有一个干净而好看的方式来完成全连接而不创建自定义实现。

Answer 1

Zip operator完全是这样。为了您的例子中，你想最终是这样的：

Observable.zip(Item, Price, Pair<Item, Price>)(items, prices, 
BiFunction { item, price -> 
    Pair(item, price) 
}) 

你可以看到对此进行了详细解释here一个例子。

编辑：根据更新的问题，他们没有订购，因为flatMap不保留秩序。取而代之的是结帐concatMap哪（direct link to the marble diagram）。