問題タブ [rx-java]

1 番目の観測結果を使用して 2 番目の観測結果を作成し、その結果をサブスクライブする必要があります。場合によっては、3 ～ 4 層の観測結果が必要になります。これを完了するための最良のアプローチは何ですか?

私はObservableから構築されたものを持っていますStream[Item]（Itemこれは単なる例です）このオブザーバブルは一連の変換を通過します：

各変換 (1 番目と 2 番目) の完了時にいくつかのアクションを実行したいと考えています。doOncompletedメソッド onobservable1を呼び出しobservable2ても、目的の効果が得られない場合 (実際には副作用は見られません)。

どうすればこれを達成できますか? これは可能ですか？

Android アプリのデータを更新するために、RESTful エンドポイントを定期的にポーリングする必要があります。また、接続に基づいて一時停止して再開する必要があります (電話がオフラインの場合は、試す必要さえありません)。私の現在のソリューションは機能していますがScheduledExecutorService、定期的なタスクを実行するために標準の Java を使用していますが、Rx パラダイムにとどまりたいと考えています。

これが私の現在のコードです。簡潔にするために一部を省略しています。

networkStatusObservableは基本的に にラップされたブロードキャスト レシーバObservable<Boolean>であり、電話がネットワークに接続されていることを示します。

私が言ったように、このソリューションは機能していますが、定期的なポーリングと新しいUserProfiles の発行には Rx アプローチを使用したいと考えています。Observable.timerとについては知ってObservable.intervalいますが、それらをこのタスクに適用する方法がわかりません (そして、それらを使用する必要があるかどうかはわかりません)。

これら2つの機能はかなり似ているようです。それらは同じ署名 (accepting rx.functions.Func1<? super T, ? extends Observable<? extends R>> func) を持ち、マーブル ダイアグラムはまったく同じに見えます。ここに写真を貼り付けることはできませんが、ここにconcatMap用の写真とflatMap用の写真があります。resultsの記述には微妙な違いがあるようですObservable。 によって生成されたconcatMapものには、結果の Observable を連結した結果のアイテムが含まれ、 によって生成されたflatMapものには、結果の Observable を最初にマージし、そのマージの結果を発行した結果のアイテムが含まれます。

しかし、この微妙さは私にはまったくわかりません。誰でもこの違いをよりよく説明できますか。理想的には、この違いを示すいくつかの例を挙げてください。

RxJava でモデル化された相互依存の非同期操作のグラフがあります。エラーによっては、グラフ全体を再実行する必要があります。retry(..) すべてのサブスクライバーにエラーが表示されるため、オペレーターはこれを直接サポートしていません。retry(..)オペレーターは再サブスクライブするだけなので、最終的なオブザーバブルから一度だけ計算されたエラーを常に取得します。つまり、再サブスクライブ時に作業が再度実行されることはありません。

すべてのサブスクリプションに対してオブザーバブル生成メソッドを呼び出す特別なオブザーバブルを作成しようとしました。その場合、再試行演算子はほとんど希望どおりに機能し、追加のキャッシュ操作の後、希望どおりに正確に機能します。

ただし、これは非常に一般的であるように思われるため、RxJava のどこかで既に提供されている作業を繰り返していると思われます。また、RxJava の十分な知識がなくても低レベルで何かをしようとしていることを考えると、ソリューションの堅牢性についても懸念しています。もう 1 つの問題は構成可能性です。3 つの形式すべてをサポートするretry(..)には、3 つのバージョンのラッパー メソッドが必要になります。

以下のデモンストレーションは、私がやろうとしていることとこれまでの成功を説明しています。

RxJavaでこの種の再試行を行うためのより単純またはより慣用的な(またはその両方の)方法はありますか?