問題タブ [rx-scala]

rx-java のサブプロジェクトである rx-scala を使用しています。私は Scala 構文を使用しますが、誰もが理解できることを願っています。

奇妙な動作に遭遇しましたが、それがバグなのか、rx オペレーターの誤用なのかわかりません。

問題文

私はox: Observable[X]とトリガーオブザーバブルを持っていtr: Observable[()]ます。function を使用しoyた変換であるオブザーバブルが必要ですが、潜在的に高価であるため、トリガーされた場合のみです。oxf: Function[X,Y]f

の最後の値に変換された値がない場合はox、 にoyなるはずですnull。

いくつかのコメント:

• oxUIイベントの結果であるため、ホットです。
• oxprintlnデバッグで確認したように、正しく動作します (値とタイミングの両方) 。
• oy適切なタイミングで発火します。oxnull 以外の値の場合、の古い値を使用しているだけです。

現在のコード

上記のコードの問題点は次のとおりです。最初は機能しますが、しばらくすると、 をトリガーするとtr、の古い値にoy適用されます。変更していないときは、繰り返しトリガーすると、結果が新しくなり、最終的に追いつきます。foxoxtr

mergeにリセットしないように を削除するnullと、すべてが正常に機能します (おそらく、効果が非決定論的に見えるため)。

質問

提示されたコードはバグがあります。

1. 私が何か間違ったことをしているかどうか知りたいです。
2. 必要なものを達成する別の方法を歓迎します。

ジャブの人々のために

• ジェネリック/型注釈:ox: Observable[X]手段Observable<X> ox
• ラムダ:x => null手段x -> null

TableEntries私は RxScala から始めており、テーブル内のいくつかの行に変更があったかどうかを確認するために、データベースを間隔 (たとえば 20 秒) ごとにチェックするポーリング メカニズムを考え出そうとしています。

List[MyTableEntries] を出力する Observable から始める必要があります。そこで、次のことから始めました。

関数に渡す Observer には、onNext、onError、および onCompleted が実装されています。ただし、いくつか質問があります。

1. データベースの応答に 30 秒以上かかるとどうなりますか
2. データベースが完全にダウンした場合はどうなりますか?

これは私が行った有効なアプローチですか? 提案？

replayRxScalaで理解しようとしています。次のようなオブザーバブルを作成します。

where はintPerSecond、0 から始まる 1 秒あたり 1 つの整数を発射します。サブスクライブする最初のオブザーバーは、実際に 1 秒あたり 1 つの整数を認識します。2 番目のオブザーバーが t=6 秒で参加すると、その時点から、両者は一致するストリーム 6...7...8...9... を 1 秒間隔で見ます。それは予想通りです。

ここで追加すると.replay(3)、2 番目のオブザーバーが参加すると、3456...7...8...9 が表示されると予想されます。つまり、彼はすぐにキャッシュから 3 つの整数を取得し、それらを受け取ります。 1 秒に 1 つずつ生成されます。しかし、代わりに、どちらのオブザーバーも今は何も見ていません。構文が間違っていますか？

type の要素のイベント ストリームがあるとしますIn。

Intype のオブジェクトを type のオブジェクトに変換する関数ですOutが、「将来的には」:

この時点で、observableInmy function に従ってmy の要素を変換したいと思いますfutureInToFutureOut。つまり、結果としてOut、元のストリームの要素と一致する type の要素のイベント ストリームが必要ですが、 function を介して変換されますfutureInToFutureOut。

私はこれがうまくいくと思います：

これは正しいですか？これを行うより良い方法はありますか？

Socket プロトコルを介してサービスに接続する Observable があります。ソケットへの接続は、クライアント ライブラリを介して行われます。私が使用するクライアント ライブラリには java.util.Observer があり、プッシュされるイベントを登録できます。

わからない未解決の問題が 2 つあります。

ステップ 3 の結果をサブスクライバーで取得するにはどうすればよいですか?

以下のようなサブスクライバーで MyEvent を取得するたびに、新しい接続が作成されていることがわかります。最終的に、受信イベントごとにステップ 1、ステップ 2、およびステップ 3 が実行されます。