問題タブ [rxjs]

Promises と Functional Reactive Programming を知っています。

私がやりたいことは、非同期関数を呼び出して .then(callback) を実行することですが、すぐにコールバックを定義したくありません。派手な機能がすべてないだけで、RxJS Observableを思い出します..

誰かが私が探しているものを知っているかもしれません。

以下のコードのスニペットは機能します (動作しているという意味では ;-)) が、せいぜい不十分なように見えます...

これをより構成可能にする方法、または少なくとも醜さを軽減する方法を誰かが提案できますか?

コードは、このページの例に基づいています: RxJS を使用して既存の API をラップする

最近、リアクティブ プログラミングの概念に基づいて動作するRxJSおよび RxJava (Netflix の) ライブラリを調べ始めました。

Node.js は、イベント ループに基づいて動作します。イベント ループは、非同期プログラミングのためのすべての武器を提供し、「クラスター」などの後続のノード ライブラリは、マルチコア マシンを最大限に活用するのに役立ちます。また、Node.js は EventEmitter 機能も提供します。この機能では、イベントをサブスクライブして非同期で処理できます。

一方、私が正しく理解していれば、RxJS (および一般的なリアクティブ プログラミング) は、イベント ストリームの原則に基づいて動作し、イベント ストリームをサブスクライブし、イベント ストリーム データを非同期的に変換します。

そこで、問題は、Node.js で Rx パッケージを使用するとはどういう意味かということです。Node のイベント ループ、イベント エミッター、Rx のストリームとサブスクリプションに対するサブスクリプションの違い。

ポイントに到達するためにボイラープレートを削除しました

// a.js

// b.js(上記でエクスポートしたモジュールが必要です)

私が期待していたもの：

{item: 'a', a: 'a'}コンソールに出力

どうしたの：

Hit!24回前に印刷されました{item: 'a', a: 'a'}。これは、zipが からすべての値を取得しaStream、それらをバッファリングしてから、本来の処理を実行したことを意味します。

zip同じ機能を遅延して提供するにはどうすればよいですか? 私の目標は、無限のストリーム/オブザーバブルを使用し、有限 (非同期) のもので圧縮することです。

編集

ランナブルを介して参照/編集: RX Zip test 編集 2コードは回答に基づいて更新されました -> 現在は出力されていません。

私のシステムでは、ソース、ソースを新しい値にマップする 2 つの「ステップ」、およびそれらの 2 つのステップを組み合わせて最終的な値を作成する合計があります。このシステムの最初の実行は期待どおりに機能し、単一の合計 3 が生成されました。

出力:

しかし、次に source に新しい値を入力すると、次のような 2 つの結果が得られます。

1 つ目は中間結果です。新しいソース値がシステムの一部を通過し、すべての値の伝播が完了すると最終結果が得られます。

私の質問は、ソースにプッシュされた新しい値がシステムをアトミックに通過し、1 つの合計結果のみを生成するように構成するための推奨される方法は何ですか?

ありがとう！

次のようなコードスニペットがあります

videoEntries.json は、videoEntry オブジェクトの JSON シリアル化配列です。サブスクリプション A とサブスクリプション B の両方が、videosNeedFix オブザーバブルによって発行されたデータを受け取ることを期待しています。

ただし、コンソール ログによると、サブスクリプション A のみがデータを受信し、サブスクリプション B は受信しません。2 つのサブスクリプションを作成する順序を入れ替えると、subscriptionB だけがデータを参照できます。オブザーバブルが最初のサブスクリプションにのみデータを発行するのはなぜですか?

ベーコンのようなものについてRxJSドキュメントを見てflatMapWithConcurrencyLimitいますが、何も表示されません。Rxで利用できるかどうかはわかりませんか？