問題タブ [rx-java2]

次のコードを使用すると:(Observable.create())

メソッド repeatUntil は実行されません。コンソールには何も記録されません。
しかし、次のコードを使用すると:(Observable.just())

メソッド repeatUntil が実行されます。コンソールには次のように記録されます。

Observable.create() が repeatUntil() メソッドを実行しない理由がわかりませんか?

現在、 FirebaseMessagingServiceから MainActivity にEventBusデータを転送するために使用していますが、複雑さが増すにつれてこれは難しくなり、複数の通知を受け取った場合はどうなりますか? onMessageReceived一方で、

EventBus により、データ転送には 1 つの余分なクラスと 2 つのボイラー プレート関数のコストがかかります。

問題は、 Rxjava を使用して FirebaseMessagingService から Activity にデータを転送する方法です。このサービス全体をいくつかのオブザーバブルに変換する方法はありますか?

私は非同期呼び出しを行っています.10秒後に1分間、つまり約6回の呼び出しが行われますが、問題はdelay特定のものに適用したいcondition

私が望むのは、最初の呼び出しを除いてネットワーク呼び出しを 10 秒間遅らせ、10 秒後にネットワーク呼び出しをキャンセルすることです。したがって、1 分間に正確に 6 つの呼び出しが必要です。

編集

シナリオの混乱のため、シナリオを再定義します。

私が持っているのはローカルのドライバーの大きなリストであり、10 秒ごとに各ドライバーに要求を送信し、別のサブスクライバーをリッスンして、ドライバーが 10 秒以内にキャンセルしなかったかどうかを確認したいのですが、このプロセスは約 1 分間続きます。あるドライバーがキャンセルした場合、すぐに次のドライバーにリクエストを送信する必要があります

これまでに書かれたコード:

私は Rx 1 から Rx 2 への移行の過程にあり、投稿を読んでいるときに突然、Singleが改造呼び出しに使用するオブザーバブルのタイプであることがわかりました。

そこで、試してみることにしました。改良呼び出しを Rx 2 に移行するときに、戻り値も に変更しましたSingle<whatever>。

問題は、一部のテストで次のようなネットワーク サービスをモックすることです。

ご覧のとおり、呼び出しを移行する前に、userServiceモックに空のオブザーバブルを返すように指示して、ストリームを完成させていました。

Single呼び出しの「バージョン」に移行する間Observable.empty()、呼び出しは を返さず、 を返すため、使用できなくなりましObservableたSingle。

私は次のようなことをしてしまいました：

私の質問は次のとおりです。

1. これを行うより良い方法はありますか？
2. 知っておくべき重要なことを見逃していますか?このようなものは、実際には期待どおりに動作しません。

ボタンが押されるたびにブール値を発行するプロセッサがあるとしましょう。これをトグルと考えてください。

私がやりたいことは、ゲートの値を使用して監視可能なシーケンスを一時停止および再開し、一時停止中に放出された値をバッファリングすることです。

私はこれをよく読みましたが、他の言語のリアクティブ拡張機能では可能だと思われますが、RxJava はそれをサポートしていないようです。

これは私が達成したいことの例です。毎秒増分値を出力するだけです。ボタンを押すと、もう一度押すまで出力を停止し、2 つのボタンを押す間に発行されたすべてのアイテムを出力する必要があります。

このようなことを達成する方法を知っている人はいますか?

編集これを達成する方法についてブログ投稿を書きましたhttps://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

短編小説: 1 つの目的を持つ 2 つの Observable がある状況があります。

• 彼らはいくつかのデータを受け取ります
• 変更されたデータを返します
• データを処理できない場合はエラーをスローします

それぞれが異なる種類のデータの処理を担当しています。さらに、両方のデータが処理されたときに何かをしたいと考えています。

私の現在の最良の実装は次のとおりです。これらは私の Observables です。

また、次のサブスクリプションがあります。

私の問題: 青と赤のデータは 2 回処理されます。これは、Observable.zip() で作成された結合されたオブザーバブルにサブスクライブすると、両方のサブスクリプションが再度実行されるためです。

両方の操作を 2 回実行せずにこの動作を行うにはどうすればよいですか?

RxJava のバージョン 2.0.0 を使用していますが、にアクセスできないようAndroidSchedulersです。RxJava を介してメインスレッドにアクセスできません

たとえば、次の実行可能な Java コードがあります。

これは、プロデューサーと複数の並行コンシューマーに関するものです。これらのコンシューマは、時間のかかるジョブを実行しており、並行して実行しています。

このユースケースが rx-java と一致するかどうか、また rx-java でどのように書き換えるかが問題です。

出力は次のとおりです。

(B)が新しい を持つたびにそのアイテムを放出する( Observable<Item> A)があります。PublishSubject<Item> Item

Aは次のように使用されます。A.subscribeOn(computationScheduler).observeOn(mainThread)

目標: Aからの完全なストリームを計算スケジューラで実行し、メインスレッドで結果を消費します。

実際: Bが観察される場所に応じて、ストリーム全体が異なるスケジューラで実行されます。.subscribeOn()以下の例では、呼び出しを介してもメインスレッドで。

A からの完全なストリームが特定のスケジューラでその作業を実行し、別のスケジューラで結果をディスパッチするにはどうすればよいですか? A.compose() はうまくいきません。

実際のコード: