javascript - RxJS で観測可能な値を特定の時間で区切る

Question

特定の時間までに Observable の値を生成する最も慣用的な方法は何でしょうか? たとえば、大きな配列から作成された Observable があり、2 秒ごとに値を生成したいとします。intervalとの組み合わせがselectMany最善の方法ですか?

Answer 1

特定の例では、アイデアは、配列の各値を遅延後に結果を生成するオブザーバブルにマップし、結果のオブザーバブルのストリームを連結することです。

var delayedStream = Rx.Observable
    .fromArray([1, 2, 3, 4, 5])
    .map(function (value) { return Rx.Observable.return(value).delay(2000); })
    .concatAll();

他の例は実際にtimerorを利用するかもしれませんinterval。それはただ依存します。

たとえば、配列が非常に大きい場合、上記はかなりの量のメモリ プレッシャを引き起こします (非常にN大きな のオブザーバブルを作成しているためN)。interval配列を遅延ウォークするために使用する代替手段を次に示します。

var delayedStream = Rx.Observable
    .interval(2000)
    .take(reallyBigArray.length) // end the observable after it pulses N times
    .map(function (i) { return reallyBigArray[i]; });

これは、配列全体を反復処理するまで、2 秒ごとに配列から次の値を生成します。

Answer 2

RxJS v6 の場合、2 秒の遅延で次のものを取得します。

例 1. concatMap:

import {of} from 'rxjs';
import {concatMap, delay} from 'rxjs/operators';

of(1, 2, 3, 4, 5)
  .pipe(
    concatMap(x => of(x)
      .pipe(
        delay(2000))
    )
  )
  .subscribe({
    next(value) {
      console.log(value);
    }
  });

例 2. map + concatAll:

import {of} from 'rxjs';
import {concatAll, delay, map} from 'rxjs/operators';

of(1, 2, 3, 4, 5)
  .pipe(
    map(x => of(x)
      .pipe(
        delay(2000))
    ),
    concatAll()
  )
  .subscribe({
    next(value) {
      console.log(value);
    }
  });

Answer 3

ブランドンの答えはアイデアの要点を理解していますが、最初のアイテムをすぐに生成し、次のアイテムの間に時間を入れるバージョンを次に示します。

var delay = Rx.Observable.empty().delay(2000);

var items = Rx.Observable.fromArray([1,2,3,4,5])
  .map(function (x) {
    return Rx.Observable.return(x).concat(delay); // put some time after the item
  })
  .concatAll();

新しい RxJS 用に更新:

var delay = Rx.Observable.empty().delay(2000);

var items = Rx.Observable.fromArray([1,2,3,4,5])
  .concatMap(function (x) {
    return Rx.Observable.of(x).concat(delay); // put some time after the item
  });

Answer 4

zip がクリーンなアプローチであることに同意します。以下は、配列の間隔ストリームを生成するための再利用可能な関数です。

function yieldByInterval(items, time) {
  return Rx.Observable.from(items).zip(
    Rx.Observable.interval(time),
    function(item, index) { return item; }
  );
}

// test
yieldByInterval(['A', 'B', 'C'], 2000)
  .subscribe(console.log.bind(console));

これはfarincz の answer.zipに基づいていますが、インスタンス メソッドとして使用することでわずかに短くなります。

また、はdeprecatedRx.Observable.from()であるため、使用しました。Rx.Observable.fromArray()

Answer 5

concatMapこれは言及されていないので、 combined withdelayはかなり読みやすい と思います。

Rx.Observable.fromArray([1, 2, 3, 4, 5])
    .concatMap(x => Rx.Observable.of(x).delay(1000));

https://codepen.io/jmendes/pen/EwaPzwを参照してください

Answer 6

最初のアイテムをすぐに発行し、残りのアイテムを遅らせる RxJs 6 コード:

import { of, EMPTY, concat } from "rxjs";
import { concatMap, delay } from "rxjs/operators";

const delayed$ = EMPTY.pipe(delay(1000));

console.log("start");
of(1, 2, 3, 4)
  .pipe(concatMap(v => concat(of(v), delayed$)))
  .subscribe({
    next: console.log
  });

完全な Stackblitz の例

考え：

• アイテムごとにconcat、アイテムをすぐに出力し（of(v)）EMPTY、遅延後にオブザーバブルを発行するオブザーバブルを（ を使用して）作成します
• concatMap放出されたすべてのオブザーバブルを使用するため、正しい順序で放出されます