問題タブ [system.reactive]

現在、RX フレームワークを使用して、ワークフローのようなメッセージ処理パイプラインを実装しています。基本的に、メッセージ プロデューサー (ネットワーク メッセージを逆シリアル化し、サブジェクトで OnNext() を呼び出す) があり、いくつかのコンシューマーがあります。

注: If と transform は、単に IObservable を返すようにコード化した拡張メソッドです。

コンシューマーは次のようなことを行います。

commerceRequestその後、別の同様のパイプラインによって消費され、これが最上部まで続きSubscribe()、最後のパイプラインを誰かが呼び出して終了します。私が抱えている問題は、サブスクライブがメッセージに対して直接どこかで呼び出されない限り、ベースからのメッセージが伝播しないことです。

メッセージをスタックの一番上にプッシュするにはどうすればよいですか? これが非正統的なアプローチであることは承知していますが、メッセージに何が起こっているのかをコードが非常に簡単に理解できるようになると思います。これがまったくひどい考えだと思うなら、誰かが同じことをする別の方法を提案できますか?

初めての単純な "hello world" RX アプリケーションに苦労しています。VS2010 RC と最新の RX ダウンロードを使用しています。

以下は単純なコンソール アプリです。

Where 句で ArgumentNullException が発生します。これがスタックです。

キュー (具体的には ConcurrentQueue として) を IObservable のソースとして使用できるかどうか疑問に思いましたか? 何かのようなもの;

何もデキューされていないので、本当に意味がないと思います。オブザーバーにプッシュされる「メッセージ」をサブスクライブできるノンブロッキングプロセスを実装しようとしているため、キューを使用しています。RXでこれを行うことができるはずだと確信していますが、頭が回らないようです!

これをどのように実装できるかについての提案に興味があります。ありがとう！

私が現在取り組んでいるシステムには、インターフェースと基本クラスとして定義されている多くのコンポーネントがあります。システムの各部分には、システムの他の部分と相互作用する特定のポイントがいくつかあります。

たとえば、データ準備コンポーネントは、最終的にデータ処理部分に移動する必要がある一部のデータを準備します。通信コンポーネントは、外部に中継するために、さまざまなコンポーネントのステータスを照会する必要があります。

現在、私は「神オブジェクト」、またはシステムのさまざまな部分についての深い知識を持つオブジェクトを使用して、システムのこれらの部分を接着します。ここでイベントに登録し、結果をあちらのメソッドにシャトルし、ここでコールバックメソッドを作成して、あちらでそのメソッドの結果を返します。特定のアクションがSTAスレッドなどで実行します。

便利ですが、この1つのタイプは、システム内の他のすべての人がどのように設計されているかについて多くのことを知っているのではないかと心配しています。イベント、メソッド、またはコールバックを公開できるインスタンス、またはこれらを消費できるインスタンスを指定できる、より汎用的なハブの方がはるかに望ましいです。

私はリアクティブフレームワークのIObservable/IObserver機能についてもっと見てきましたが、それは.NET 4.0に組み込まれています（私は信じています）。

このパターンを利用して、「神オブジェクト」を置き換えることはできますか？どうすればこれを行うことができますか？この特定の目的のためにこのパターンを使用するためのリソースはありますか？

並行して処理したい監視可能なコレクションがあり、フィルタリング中に処理された値を監視し、最後にフィルタリングされた値を受け取るハンドラーをサブスクライブします。

私のサンプルは構文的に正しく、正常にコンパイルされます。コードを実行するとWhere、フィルタリングを実行するステートメントが評価されます。ただし、サブスクリプションにはデータが届きません。を削除AsParallelして処理が通常どおりに行われるようにするIEnumerableと、データが通過し、すべてが期待どおりに機能します。

これが私のサンプルで、文字列に対していくつかの処理を行っています。

次の奇妙なことは、TakeWhile演算子を使用する場合です。これは、概念的にはWhereに似ていますが、ParallelQueryを観察すると期待どおりに機能します。

サブスクリプションにログコードを追加すると、データは変換まで受信されますが、ToObservable変換後は受信されないことが示されます。

4行目のラムダのブレークポイントはヒットしますが、6行目のラムダのブレークポイントはヒットしません。

なぜTakeWhileデータが加入者に届くのにWhere届かないのですか？

重要な場合は、.Net4.0Frameworkクライアントプロファイルを対象としたプロジェクトを使用してVisualStudio2010RCでコードを開発します。

更新： @Sergeysの回答に基づいて、フィルターの配置を作り直しましたWhere。次のコードは期待どおりに機能します。

processedStrings最初に最初のオブザーバブルを列挙可能に変換して並列化し、次にそれをオブザーバブルに変換して最終結果をサブスクライブする必要があるのは、まだ少し厄介です。

.Net3.5SP1のRxで遊んでいて、101Rxサンプルを試しています。最初のサンプル（[開始]-[コードを非同期で実行]）を試していますが、実際には非同期で実行されていないようです。例えば、

出力

これについての説明はありますか？私は何か間違ったことをしていますか？これは予想される動作ですか？

アップデート

私はそれが言っただろうと思ったでしょう

しかし、おそらく非同期呼び出しが発生している間、メインスレッドはブロックされます。

では、Reactive Framework (Rx) を使用して Silverlight のローカリゼーションを実装し、オンザフライで UI を変更するには、どのような方法が最適でしょうか? 例を挙げていただければ幸いです。

オンザフライ UI 変更の例を次に示します。しかし、私はそれがどのように機能するのか理解できません。

私が本当にやりたいことは、2 つの非同期操作を次々と完了することだけです。例えば

Web サイト X をダウンロードします。それが完了したら、Web サイト Y をダウンロードします。

ILMergeのドキュメントによると、競合するタイプのアセンブリをマージできます。FSharp.Core.dllとSystem.Reactive.dllをマージしようとしていますが、このためのコマンドラインパラメーターを理解できません。

ILMergeは、の競合するバージョンについて不平を言っていますSystem.IObservable<T>。この競合を解決するためにどのようにそれを伝えることができますか。

よろしくお願いします、forki

これの簡略版はありますか。あるシーケンスを呼び出し、終了したら別のシーケンスを呼び出したい。