RxJSで特定の条件になるまでポーリングする

スポンサーリンク

interval

RxJSで一定時間ごとに処理を流す時には interval を使う。

例えば、１秒ごとに何か処理をする場合は、次のように書く。

interval(1000).subscribe(console.log);

↓ 実行結果

0
1
2
...

takeWhile で終了条件を指定する

interval は無限に処理を流すので、特定の条件で interval を抜けたい時は takeWhile を使用する。

次の例は、takeWhile で５秒経ったら interval を抜けるように条件を指定する例である。

interval(1000)
  .pipe(
    takeWhile((n) => n < 5)
  )
.subscribe(console.log)

スポンサーリンク

ポーリングする（リソースの排他制御）

上で紹介した interval と takeWhile を使用して、同一リソースへの同時アクセスを防ぐためにポーリングする処理が実装できる。

次の例は、同時に ajaxが処理が行われないように、排他制御を行うサンプルコードである。

// 処理中かどうか（リソースへの同時アクセスを制御するフラグ）
let processing = false;

// processingがfalse（未処理）になるまでポーリングする interval
const pooring = interval(100)
  .pipe(
    takeWhile((n) => processing)
  );

// リソースにアクセスする処理（今回は ajax処理）
const ajaxProcess = of(true)
  .pipe(
    mergeMap(flg => {
      processing = flg;
      return ajax(`https://api.github.com/users?per_page=5`);
    }),
    map(userResponse => {
      console.log(userResponse.response);
    }),
    finalize(() => processing = false)
  )

// ポーリングとajax処理を行うObservableを結合
concat(pooring, ajaxProcess).subscribe()