問題タブ [backpressure]

Android ReactiveLocation ライブラリを使用して、定期的な位置情報の更新を受信して​​います。必要に応じてすぐに使用できる最新の位置情報を常に保持できるように、自分のアプリで位置情報を使用していない場合でも、位置情報の更新を受信し続けたいと考えています。

これが、アプリのコア コンポーネントで位置情報の更新を開始する方法です。最新の値と新しく検出された場所を、アプリのチェーンのさらに下にあるすべてのサブスクライバーに再公開したいと考えています。これreplay(1)が の目的です。

アプリの別の場所で、この再公開された obrvable をサブスクライブします。

これは仕事をしているように見えます: 私の最終的なサブスクライバーは最新の場所をすぐに返し、新しい更新を受け取り続けますが、最終的なサブスクライバーがチェーンのどこかに未使用の場所の巨大なバッファーを構築していないことを確認したいです。購読されていません。私はRx初心者です。この状況に背圧はどのように適用されますか? 私replay(1)が期待していることを行っており、最新の場所以外の不要な場所をすべて破棄していますか?

書き込み可能なストリームであるモジュールを作成しています。ユーザーのためにパイプ インターフェイスを実装したいと考えています。

エラーが発生した場合は、読み取り可能なストリームを一時停止してエラー イベントを発行する必要があります。次に、ユーザーが決定します。エラーが発生しても問題がなければ、データ処理を再開できるはずです。

ノードがreadable.pause()、読み取り可能なストリームを一時停止するために使用できるメソッドを提供していることがわかります。しかし、書き込み可能なストリーム モジュールから呼び出す方法がわかりません。

書き込み可能なストリームでどのようにバック プレッシャーを実装できますか?

PSpipe/unpipeパラメータとして読み取り可能なストリームを提供するイベントを使用することができます。しかし、パイプされたストリームの場合、一時停止する唯一の機会は、書き込み可能なストリームから読み取り可能なストリームをアンパイプすることであるとも言われています。

私はそれを正しく理解しましたか？ユーザーの呼び出しが再開されるまで、書き込み可能なストリームをアンパイプする必要がありますか? そして、ユーザー呼び出しが再開された後、読み取り可能なストリームをパイプで戻す必要がありますか?

SyncOnSubscribe を作成するコツをつかもうとしていgenerateStateますが、状態の初期化中にメソッドが失敗した場合の対処方法がよくわかりません。

これを手に負えないように処理できるいくつかの方法を考えることができます。

1. 実行時例外がスローされた場合、ライブラリは自動的に呼び出しo.onErrorますか? （編集を参照）。
2. MyStateエラーを格納する別の変数をラップしo.onErrorて、最初に呼び出されたときに自分で呼び出すことができnextます。

これに関して提案された実践があるかどうか、私はただ興味がありますか?

ありがとう！

編集:メソッドでランタイム例外をスローしようとしましたが、メソッドで検査するエラーでgenerateStateラップすることを示していると思います。より良い提案がある場合は、コメント/回答してください。MyStatenext

上記のmain関数を使用するgetEventsOnSubscribe1と、getEventsSyncOnSubscribe両方ともsubscriberを呼び出しますonErrorが、プログラムはハングしたままになります。whichを使用getEventsOnSubscribe2して例外をラップし、手動で呼び出すs.onErrorと、プログラムは終了できます。

いくつかのファイルをダウンロードする処理キューを RxJava に実装したいと考えています。ダウンロードしたいファイルの数は、最大で 100 程度です。

すべてが RxJava 1.1.1 を使用して Android で開発されています

私の現在の実装は次のようになります。

Where_getObserver()は、「onNext」メソッドでファイルにダウンロードする新しいオブザーバー オブジェクトを返します。

しかし、私の問題は、MissingBackpreasureException理解できない をすぐに取得することです。を実装しようとしましたbackpreasurebufferが、呼び出されていないようです。

私は何を間違っていますか？

core.async のミックスを試しています。ミックス内の入力チャンネルをミュートすることは、バックプレッシャーを実装する可能な方法の 1 つと思われます。以下のコードを使用しています。

REPL の最後の行を評価すると、

CompilerException java.lang.IllegalArgumentException: No implementation of method: :toggle* of protocol: #'clojure.core.async/Mix found for class: java.lang.Boolean.

上記のコード例の何が問題になっていますか?

ありがとう！