問題タブ [rx-py]

最小限の実例 - RxPY (v.1.2.4)'Hello world'を使用して単語ごとにエコー サーバーに送信するクライアント。 Observable

クライアント:

サーバー: ねじれた例からの単純なエコー サーバー -- echoserv.py

サーバーが実行されている状態でクライアントを起動すると、次のようになります。

私は何を間違っていますか？

インストールしたRxPYのバージョンを思い出せず、明らかに機能していないようです。

編集

また、helpバージョンが含まれていません:

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

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

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

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

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

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

問題は、RX でダブル ループ動作をエミュレートすることです。

2 つのループが 2 つのオブザーバブルに置き換えられ、一方が外側のオブザーバーとして機能し、do_something(value)それ自体をオブザーバーに置き換えることができれば、きれいになります。例外も適切に処理できます。外側のループはブロックする必要がありますが、バックオフ関数で再試行関数を使用して外側のループを使用して例外を処理しようとしているため、内側のループはそうでない場合があります。

これまでのところ、次を使用してシーケンスを構築できます。

しかし、アウターのブロッキングモードで同様の構造を作成するにはどうすればよいですか?

を誤ってイテラブルとして使用するバグがありObservableました。ほとんどのオブジェクトでは、これは通常簡単に検出されます。

ただし、Rx オブザーバブルの場合、Python は黙ってクラッシュします。

MWE:

トレースバックはなく、問題があるという兆候もありません。これにより、すでにデバッグが困難な同時リアクティブ コードのデバッグがさらに困難になります。最終的にこれを追跡するのに 2 時間かかりました。

Observableよく調べてみると、オブザーバブルにメソッドがないことを考えるとどこからかわからないものの、を繰り返し処理すると新しいオブザーバブルが作成されるように見えます__iter__。

これはバグですか、それとも機能ですか? Observables は反復可能であることを意図していますか?

ロギング目的でオブザーバーsourceにサブスクライブするオブザーバブルがあります。logger

sourceまた、計算を実行できるように購読しています。私の計算が完了すると、私は終わったので、次sourceのものを処分したいと思いますlogger:

ただし、logger適切なタイミングで完全に破棄されるわけではありません。通常、1 つまたは 2 つの余分なティックが発生します。

MWE

しかし、私は得る：

7 はどのようにログに記録されましたか? 私たちの計算sourceは 5 を発行した後に終了する必要があり、その時点でloggerが破棄されます。

私は何を間違っていますか？

ObservableRxPy でシーケンスから要素を復元するにはどうすればよいですか

AnonymousObservable1 を出力する必要がありますが、要素の代わりに別の を返します。

Observable一般に、シーケンスから要素を復元するのに最適な演算子は何ですか?

を使用group_byしてObservableいますが、新しく作成されたグループごとに、次を使用してグループを作成する原因となった要素を (新しいキーで) スナップしたいと考えていますwith_latest_from。

以下の両方が印刷されることを期待していますが、毎回どちらか一方しか表示されません。

奇妙な機会に、スナップされた要素が 2 として表示されることもあります。

何がうまくいかないのですか？