問題タブ [reactive-programming]

私はリアクティブ拡張の初心者です。Gideon Engelberth は、私の質問で Reactive Extension について素晴らしい回答をくれました。

LINQ または Rx を使用して、1 つのメソッド チェーンで HTML の img url を BASE64 文字列に変換する方法

ここで、IObservable の戻り値を XDocument インスタンスに割り当てる方法という 2 番目の質問があります。

ギデオンは私に以下のサンプルをくれました。

私はこのようにしたいと思います。つぼみ 難しそうですね…

反応バナナで特定のタイプのイベントのスロットリングを実装したいと思います。通過した最後のイベントからデルタ秒未満で到着した場合、イベントが通過しないように動作する必要があります。通過させられない場合は保存され、最後に発生したイベントから delta 秒後に発生します。

以下は、タイムスタンプ付きの番号のリストに対してこれを実装するプログラムです。これを react-banana に変換することは可能でしょうか?

また、リアクティブバナナでは、他のイベントが入ってからx秒後にイベントを発生させるにはどうすればよいですか?

次のような拡張メソッドを追加できるようにしたいと思います。

どこ：

出力を生成します。

上記の方法を実装する方法に関する提案はありますか? 使用できる組み込みメソッドはありますか? (前提: 同期/即時サブスクリプション)

まず背景として、 Duplicitiy (github.com で)という名前のオープン ソース .NET ライブラリを作成しました。これは を使用して、 FileSystemWatcher2 つのディレクトリ間ですべてのファイル変更を複製します。

実装するFileSystemObservableクラスを作成しました（ FSWatcherIObservable<FileSystemChange>を使用して実際のをラップします）。ファイルまたはディレクトリが作成、変更、または削除されると、Reactive Extensions を使用して変更が公開されます。FileSystemWatcherSubject<FileSystemChange>

次に、次のサブスクリプションを使用して、このオブザーバブルをサブスクライブします。

変更は、最大 1 分間、変更がない状態が少なくとも 2 秒間続くまでバッファリングされます。これは、ディレクトリを削除するときに、FileSystemWatcherが含まれているすべてのファイルとディレクトリについて通知するためです。ディレクトリ内に含まれる変更を飲み込み、サブスクライバーの親を削除するだけで、動作を最適化できます。これはPrioritizeFileSystemChangesフィルターによって処理されます。また、バッファ ウィンドウ内で作成され、その後削除されたファイルを無視することもできるため、ターゲットでの IO 操作が再び減少します。

これは、失敗/再試行をサポートしていない現時点では単純な方法ではありますが、機能します。

ただし、私の質問は、このオブザーバブルのサブスクライバーが各変更を処理するのにかなりの時間がかかる可能性があるということです。たとえば、大きなファイルを低速のファイル システムにコピーします。現在コピーされている同じファイルに対して新しいファイル システムの変更が発生した場合、進行中の操作を中止するにはどうすればよいですか。または、ファイルがバッファリングされたリストに含まれているが未処理の場合、どのように削除または除外できますか?

元のオブザーバブルへの別のサブスクリプションが必要になると思いますが、状態を共有したり、保留中のタスクを変更したりするのに最適な方法がわかりませんか? 変更は、受信した順序で処理する必要があります。これはキューを示します。ただし、新しいファイル システムの変更は、キャンセルまたは削除する必要があるキュー操作に適用される場合があります。キューは、順不同の削除用に設計されていません。

たとえば、現在ファイルFoo\Bar.txtをコピーしていて、Fooディレクトリが削除されているとします。次に、ディレクトリおよびすべてのサブディレクトリに対する進行中または保留中の変更をキャンセルする必要があります。これは Task Parallel Library の使用例でしょうか、それとも私が取れるリアクティブなアプローチはありますか?

github のプル リクエストもお待ちしております。

次のような RX コードがあります。

数日に一度、プログラムが onEvent メソッドに入らなくなり、回復することはありません。(それでも OnNext 呼び出しは発生し続けます)。

サブスクリプションに対して破棄が呼び出されませんでした。

rx-experimental 1.11111 を使用しています。

何が原因でしょうか?

私は現在 Haskell と Yampa を学んでおり、反応機能について質問があります。

型シグネチャからわかるように、sense 関数の出力の一部は、関数の現在の呼び出しと前回の呼び出しの間の時間差です。私が見た例では、この時間差は意味の中で「手動で」計算され、IORef を使用して前の呼び出しの値を保持します。

外部状態を使用して時間差を追跡しなければならないのは奇妙に思えますが、なぜこの計算が反応関数で行われないのですか? IORef はそれを処理する良い方法ですか?

Rxで最初の一歩を踏み出すと、ここで立ち往生しています。

驚いたことに、上記のコードは

両方のサブスクリプションの後にConnect（）を手動で呼び出すことは、ここで必要なことではないことに注意してください。ただし、出力は期待どおりです。

これどうやってするの？

私はコードを持っています：

問題は、この最初の呼び出しがコールド オブザーバブルであるため、別の呼び出しでは次のようになることです。

myObserver は onNext をまったくトリガーしません。どうやら最初の呼び出しが公開されるため1,2,3, ENDです。後でサブスクライブを続けたいので、オブザーバブルに「END」を呼び出させたくありません。これを簡単に行う機能はありますか？

私は現在、Windows Phone アプリケーションに取り組んでおり、Reactive Extensions を使用して非同期を作成し、UI エクスペリエンスを向上させたいと考えています。

私は MVVM パターンを使用します。私のビューには、ViewModel でアイテムの ObservableCollection にバインドされた ListBox があります。Item には、Name や IsSelected などの従来のプロパティがあります。

App.xaml.cs で MainViewModel をインスタンス化します。

そして、データを MainPage_Loaded にロードします。

ViewModel からデータをロードします (後でそのコードをモデルに移動します)。

そしてそれは正常に動作し、アイテムはビューで更新されます。

今、私は非同期性を作りたいと思っています。LoadData の代わりにビューから呼び出す新しいメソッドの従来の BeginInvoke を使用すると、正常に動作します。

私は CurrentDispatcher と呼ばれるプロパティを使用します。これは、App.xaml.cs に App.Current.RootVisual.Dispatcher で埋められます。

しかし、Reactive Extentions を使用したいと思います。たとえば、ToAsync や ToObservable などの Rx のさまざまな要素を試してみましたが、チェックリストに項目を追加すると、「無効なクロススレッド アクセス」で「UnauthorizedAccessException が処理されませんでした」というメッセージが表示されました。

他のスレッドを ObserveOn しようとしましたが、エラーは UI とバックグラウンド スレッドが混在していることが原因である可能性がありますが、機能しません。たぶん、その特定のケースのように Rx を使用しないでしょうか?

どんな助けでも大歓迎です。

あなたの答えの後に編集してください：

これはうまく機能するコードです！

ご覧のとおり、以前は正しく使用していませんでした。ObservableCollection を使用して、Subscribe で使用できるようになりました。パーフェクトだ！どうもありがとう！

以下は、System.Diagnostics.Process を IConnectableObservable に変換する試みです。このソリューションには問題があります。標準出力とエラーを継続的にリッスンし、イベント Process.Exited を OnCompleted のトリガーとして使用したいと考えています。残念ながら、出力バッファーが空になる前にProcess.Exited が発生することがわかりました。これは、スレッド スリープを使用した醜い回避策がなくても、OnNext ステートメントを介して出力が提供されない状況を再現できることを意味します。

Q1: この問題の回避策はありますか?

Q2: System.Reactive に関して: 自分のソリューションで何が改善できたでしょうか?

よろしく、

マーカス