問題タブ [rx-java]

clojure 1.5.1 を使用して lighttable でrxjava ( http://netflix.github.io/RxJava/javadoc/ ) を使用する場合は、次のことを考慮してください (モナドからデータを抽出するために、副作用のある変更 (恐ろしい!) 外部コレクターを使用する)

これは以下を生成します:

言いたい

しかし、驚き

実際、ちょうど

同様の例外を生成します。Observable/toObservable次のような式の場合、Clojure が検出できるのはなぜですか。

のような表現ではありません。

?

このメモの長さについて前もってお詫び申し上げます。短くするのにかなりの時間を費やしましたが、これは可能な限り小さくなりました。

私には謎があり、あなたの助けに感謝します。observerこのミステリーは、私が Clojure で書いた rxjava がobservable、オンライン サンプルから引用したいくつかの簡単な s に対して動作することに起因しています。

1 つのオブザーバブルは、そのオブザーバーのハンドラーにメッセージを同期的に送信しonNext、私の原則的なオブザーバーは期待どおりに動作します。

もう一方のオブザーバブルは、Clojure を介して、別のスレッドで同じことを非同期に行いますfuture。まったく同じオブザーバーが、その に投稿されたすべてのイベントをキャプチャーするわけではありませんonNext。末尾のランダムな数のメッセージが失われるようです。

promise以下では、 dの待機の満了と、コレクターonCompletedに送信されるすべてのイベントの待機の満了との間に意図的な競合があります。が勝ったagent場合、. が勝った場合、のキューからのすべてのメッセージが表示されることを期待しています。私が期待していない 1 つの結果は、AND からの短いキューです。しかし、マーフィーは眠らない、それはまさに私が見ているものです. ガベージ コレクションに問題があるのか​​、それとも Clojure の STM への内部キューイングなのか、それとも私の愚かさなのか、それともまったく別の何かなのか、私にはわかりません。promisefalseonCompletedagentagenttrueonCompletedagenttrueonCompletedagent

を介して直接実行できるように、ソースを自己完結型の順序でここに示しますlein repl。最初に、 Netflix の rxjava のバージョンへのproject.clj依存関係を宣言するleiningen プロジェクト ファイルです。0.9.0

これで、名前空間と Clojure 要件、および Java インポートが次のようになります。

最後に、コンソールへの出力用のマクロ:

最後に、私のオブザーバーに。を使用しagentて、任意のオブザーバブルから送信されたメッセージを収集しますonNext。を使用しatomてポテンシャルを収集しonErrorます。オブザーバーの外部の消費者がそれを待つことができるように、 promiseforを使用します。onCompleted

さて、ここに同期オブザーバブルがあります。それは 25 のメッセージをonNextオブザーバーの喉に送り込み、次に を呼び出しますonCompleted。

オブザーバーをこのオブザーバブルにサブスクライブします。

期待どおりに動作し、コンソールに次の結果が表示されます

futureのスレッドでのみ、まったく同じことを行う非同期オブザーバブルを次に示します。

しかし、驚いたことに、コンソールに次のように表示されtrueます。ただし、非同期メッセージの一部のみです。表示されるメッセージの実際の数は実行ごとに異なります。これは、何らかの同時実行現象が発生していることを示しています。手がかりに感謝します。onCompletedpromise

Clojure でhttps://github.com/Netflix/RxJavaで頻繁にビルドされている最新機能のいくつかを使用したいのですが、ビルドしたローカル .jar を Leiningen に参照させるのに苦労しています。Clojars にリリースされた rxjava の最後のバージョンは 0.9.0 で、projects.clj で次のようにすると正常に到達できます。

ここで、現在の rxjava ソースのビルドに成功しました。これにより、次のファイルが生成されます。

Clojure プロジェクト ディレクトリで、次のことを行います。

https://gist.github.com/stuartsierra/3062743 (下部を参照) http://www.pgrs.net/2011/10/30/using-local-jars-with-leiningen経由の指示に従います。 /およびleiningen - ローカル jar の依存関係を追加する方法は? .

これは成功を報告し、次の結果を生成します。

projects.clj ファイルを修正して、次の内容を含めます。

しかし、lein deps はローカル リポジトリを見つけることができません。

project.clj の次の推測も機能しません。

続行する方法についてのアイデアはありますか?

Is this code thread-safe?

I'm curious because ArrayList is not a thread-safe data structure.

異なるパラメーターを使用して 5 秒ごとに実行する必要がある、パラメーター化されたレスト コールがあります。

Observable<TResult>param1 の異なる値を使用して、5 秒ごとに restCall をポーリングするを作成する必要があります。API 呼び出しが失敗した場合は、エラーを取得して 5 秒以内に次の呼び出しを行う必要があります。呼び出し間の間隔は、restCall が終了したとき (成功/エラー) にのみ測定する必要があります。

私は現在 RxJava を使用していますが、.NET の例も良いでしょう。

だから私は、機能的なリアクティブ パラダイム全体に強い関心を持っており、これらのプラクティスのいくつかを導入しようとしています。しかし、まだ理解に苦しんでいるのは、これらの非同期サーバー側呼び出しをフロントエンドで使用する方法です。では、サービス レイヤーの結果がObservable<?>オブジェクトになり、コントローラーによって呼び出されるとします。偉大な。しかし今、クライアント側からのこの非同期の結果をどのように処理するのでしょうか? 通常、これは、たとえばオブジェクトからの結果をブロックし、結果Future<?>オブジェクトを構築して呼び出し元のクライアントに送り返す場所です。しかし、要点はブロックしないことです。では、Observable をどうしますか? 結果が利用可能になったら、Web ソケットを使用して結果を「プッシュ」する必要がありますか?

この質問が理にかなっていることを願っています。呼び出しと結果のコンシューマーがサーバー側に存在する限り、これはすべて理にかなっています。しかし、クライアントがコントローラーを呼び出した後に結果を期待する Web アプリケーションでは、ここで非同期でリアクティブなパラダイムを維持する方法がわかりません。これを理解する上で、どんな助けも大歓迎です。

Android で rx-java を使用して、いくつかの連続した http 要求を実行しようとしていますが、それぞれが前者の応答に依存しています。これは map() / doFinall() モデルに完全には適合しないため、「コールバック地獄」に陥ったり、簡潔なコードを記述したりせずにこれを行うための最良の方法が何であるかはわかりません。

より具体的には：

1. http GET "/x" を実行します
2. (2)が成功した場合は http GET "/y" を実行
3. GET /y の結果を計算する

これをどうするかについての提案はありますか？

Observable の作成後に要素を追加したい場合、シナリオを実装するにはどうすればよいですか? オブザーバー パターンでは、イベントを発生させるだけです。何かアイデアはありますか？

RxScala を学習しようとしています。サブスクリプションを使用Observable.interval(1 second).buffer(10 seconds)して値を出力していますが、サブスクリプションを正しく解除できません。サブスクリプションのサブスクリプションを解除した後も、サブスクリプションは実行され続けますが、結果は空になります。

私のコード:

結果をうまく出力します：

しばらくして、購読を解除します。

そして、私はまだプリントを取得しますが、空のバッファを使用します:

これは、サブスクリプションがまだ実行されていることを意味すると思いますが、空のバッファーを受信して​​います。サブスクリプションを停止するにはどうすればよいですか?