問題タブ [rx.net]

Switch ステートメントを使用した Rx サブスクリプションに問題があります。

フローは次のとおりです。

1. 一部のプロパティが変更され、 performSearchSubject.OnNext が呼び出されます
2. ヒットするたびにオブザーバーを返す PerformPositionQuery() が呼び出されます。
3. このオブザーバーを介して応答するサービスは、データ受信が完了すると OnNext を 2 回呼び出し、OnCompleted を 1 回呼び出します。
4. メソッド DataArrivedForPositions が予想どおり 2 回呼び出される
5. データ サービス内でobserver.OnCompleted()が呼び出されますが、メソッドPositionQueryCompletedは呼び出されません。

dataService のコードは次のとおりです。

EventStore で Reactive Extensions を使用することを検討し始めました。概念実証として、Rx でイベント ストリームを消費し、1 秒間のウィンドウでタイプごとにグループ化されたイベントの数を出力できるかどうかを確認したいと思います。

たとえば、「orders」という名前のストリームを使用しているとします。コンソールに次のようなものが表示されることを確認したいと思います。

(秒が過ぎます..)

等々。

これまでのところ、 1 秒あたりのすべてのイベント数の出力を取得できました。しかし、イベントの種類ごとにグループ化できないようです。

私が使用しているコードは、James Nugentの要点に基づいています。

非同期で開始および終了するストリーム (オブザーバブル) を結合しようとしています。

必要なもの: オーディオ ストリームを一緒に追加します。それらはオーディオの「チャンク」のストリームですが、ここでは整数で表現します。最初のクリップが再生されます。

そして、少し後に 2 番目のものが開始されます。

それらを合計で結合した結果は次のようになります。

ただし、圧縮されたストリームのいずれかが終了すると、標準の zip は完了します。ストリームの 1 つが終了しても、この optional_zip を続行したいと考えています。Rx でこれを行う方法はありますか、または既存の Zip を変更して自分で実装する必要がありますか?

注: 私は RxPy を使用していますが、ここのコミュニティは小さく、Rx オペレーターは言語間で非常に普遍的であるように思われるため、rx-java および rx-js としてもタグ付けしました。

サウンド処理アプリケーションでは、多くのイベントは、このシナリオでは壁時間と呼ばれるシステム時間ではなく、オーディオ ストリーム時間 (最初の概算では、消費されたサンプルの数に係数を掛けたもの) から時間を計られます。たとえば、ストリーム時間に基づいて将来のイベントの時間を計るのが望ましい場合がよくあります。たとえば、一定時間音声アクティビティがなかった場合に録音を受け入れるようにタイマーをスケジュールできます。この目的では、実時間よりもストリーム時間の方が時間を表すのに適しています。

HistoricalSchedulerそのような活動のタイミングに を使用することを考えています。私が理解している限り、それはすべてのスケジュールされたアクションを呼び出した同じスレッドで実行しますAdvanceBy/To。また、少なくとも RX を使用するようなサードパーティの部分がないという意味で、アプリケーション全体が私たちの管理下にあります。

質問 1:この考えは正しいですか、大丈夫ですか、それとも信じられないほどばかげていますか? 私はRXを頭で理解していますが、経験はありません。必要に応じて直感で理解してください。

ストリームは、.NET イベント コールバックを通じて配信されます。ストリーム バッファを配信する (およびイベントを呼び出す) スレッドは、できるだけブロックしないようにする必要があります。オーディオは有限長のエンドレス バッファに転送されます。この実際のバッファリングは、専用スレッドで監視しているこのイベントのオブザーバで発生し、ここでスケジューラの時間が進みます。基本的に、これは次のようになります

質問 2:上記の例のように、監視関数内からスケジューラを進めるのは奇妙に感じます。これは容認できる慣行ですか、それともすぐに怪しいにおいがしますか (または、両極端の中間にあるのでしょうか)。

もう 1 つの複雑な問題は、ストリームが最終的にデータの生成を停止し、ストリーム時間も停止することです。

質問 3:一部のスケジュールされたアクションが決して実行されず、永久にキューに残るように、スケジューラーで時間を停止することの影響はありますか? スケジューラへの参照を削除するだけで、使い捨てのリソース リークが発生しますか?

また、ストリーミング時間は実時間とは異なります。

質問 4:一貫性のない概念を提供するさまざまなスケジューラの影響は、あるとすれば何Nowですか?

他のシナリオではリアルタイムスケジューラと平和的に共存するため、ここでの私の最善の推測は「まったくない」ですがHistoricScheduler、RXのこの部分を十分に理解しているかどうかはわかりません。

最近、 C#rxのバージョンをいじり始めましたが、次の問題をどのように解決できるのか疑問に思っています。

次の方法でサーバーからアイテムのリストを取得するためにrefitを使用しています。

後で各アイテムを処理したいのですが、その方法がわかりませんでした。RxJava には、各項目を処理できるようにするという演算子があることは知っていますが、 C#flatMapIterable()に似たものは見つかりませんでした。

ありがとう

メッセージを表す受信バイト ストリームを解析しようとしています。ストリームを分割し、各部分のメッセージ構造を作成する必要があります。

メッセージは常に 0x81 (BOM) で始まり、0x82 (EOM) で終わります。

データ部分はエスケープ バイト 0x1B (ESC) を使用してエスケープされます。データ部分のバイトに制御バイト {ESC、BOM、EOM} のいずれかが含まれる場合は常に、先頭に ESC が付きます。

ヘッダー部分はエスケープされず、制御バイトが含まれる場合があります。

IObservable<byte>を消費して に変換することにより、Rx.Net を使用してこれを機能的なリアクティブ スタイルでコーディングしたいと思いますIObservable<Message>。

これを行うための最も慣用的な方法は何ですか?

いくつかの例：

このためのステート マシンの図を次に示します。

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

Observable.Interval(TimeSpan.FromMilliseconds(2500)).SubscribeOn(XXX).ObserveOn(YYY).Subscribe( t => SendCounter(t), e => HandleException(e));

XXX、YYY はSchedulersです。

SendCounter(t) 内で、t 値でテキストを設定します。

問題は、コードを実行すると、次のエラーが発生することです。

私はこのコンポーネントを使用しています: https://components.xamarin.com/view/rxforxamarin

次のような更新メソッドを持つゲーム(MonoGame / XNAに基づく)があります:

これをリアクティブパターンに変換したいと思います。私の現在の解決策は次のとおりです。

私は Rx を初めて使用するので、最善の方法をまだ学んでいます。Subjectこれは避けるObservable.Createべきであり、代わりに使用する必要があると読みました。

Subjectここで適切ですか？

Observable.Createこの場合、どのように使用できますか？