問題タブ [rx-kotlin2]

次のようなデータフローがあります。

ビジネス ロジック (実際には s を返すFuture) をに置き換えましたCompletableFuture.supplyAsync。はい、これは Kotlin ですが、意図は理解できたと思います。

「デッド」値 ( 57005, 0xDEAD) をコメントすると、出力は次のようになります。

しかし、その「死んだ」値がストリームに現れると、失敗します:

私はRXの初心者なので、すぐに解決策をグーグルで検索しました: onExceptionResumeNext: Observable.fromFuture(it)--> Observable.fromFuture(it).onExceptionResumeNext { Observable.empty<Int>() }. しかし、私のアプリケーションは永遠にハングアップします (私が期待する出力を生成した後)。ストリームは決して終わらないようです。

Observableどういうわけかそれを「シャットダウン」する必要がありますか？それとも、より一般的には、RX を使用する場合の適切なアプローチですか? 別の方法で考え直す必要がありますか？

私は RxJava 2.* を使用しており、zip 演算子を使用して 2 つのオブザーバブル (1 つは改造から、もう 1 つは部屋から) の結果をマージしたいと考えています (お気軽に提案してください)。

リモート サーバーからのモデル オブジェクトは、ルーム データベースからのモデル オブジェクトとは異なります。

1. リモートのオブジェクトをローカルのオブジェクトにマップしたい
2. これら 2 つの結果をマージする
3. 結果を表示します。

私のリモートAPIは次のようになります:

}

私の Room DAO クエリは次のようになります。

私は Observable> を Observable> に次のように変換しました:

しかし、これら2つのオブザーバブルを次のように圧縮しようとすると:

Type inference failed, Expected type mismatch というメッセージが表示されます