RxJava。 Observable/Flowableストリームからのすべてのアイテムに参加する

Question

1つのWebページに表示するアイテムをロードしています。並行して、商品の価格を読み込んでいます。簡素化され、それは次のようになります。

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

UPDATE：どちらも「項目」と「価格」は順序付けされていません。

今すぐ一緒に参加したいです。

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）です。