【RxJS入門】Expand関数を使った再帰的なObservable

こんにちは、ムラびとです。今回は、RxJS で再帰的、While 文を Observable でどのように表現するかをまとめてみました。
今回は以下のように解説していきます。
なお、今回は Rxjs6 を使い、TypeScript を使って書いていきます。

• RxJS とは
• expand 関数とは
• expand 関数の使い方

RxJS とは

RxJS を使うと、非同期処理をみやすく簡単に使うことができるよう作られたライブラリです。
考え方としては、ベルトコンベアーに乗せられた Http リクエストなどの非同期のデータをオペレーターを使って処理していきます。
公式サイトはこちら
詳しく使い方をしりたいひとはこちら
インストール方法は
npm i rxjs 注意：rxという名前のライブラリもありますが、こちらは RxJS5 を使っており、書き方が若干異なります。 例としてqiita.comをスクレイピングしてみましょう。

import { from } from 'rxjs'
import { map } from 'rxjs/operators'
import * as request from 'request-promise'
from(request('https://qiita.com/'))
  .pipe(map(d => 'data:' + d))
  .subscribe(v => console.log(v))

このように from 関数で Promise から Observable に変換し、map 関数でそのデータを加工、subscribe 関数でデータの受け取りをしています。
subscribe 関数で受け取りを指定するまでは Observable はデータを流さないので注意です。

expand 関数とは

Observable をスクレイピングで次のページを取得していってほしいなど再帰的に使いたい場合というのは多々あると思います。そのような時に expand 関数を使います。
expand 関数はある値を引数に Observable を生成する関数を受け取り、empty()を返すまでその関数を再利用していきます。
リファレンスはこちら

expand 関数の使い方

例としてコラッツの問題の過程を出力していきましょう。
比較のため普通の再帰関数を使った場合をみてみましょう。

function collatz(n: number) {
  console.log(n)
  if (n === 1) return 1
  if (n % 2 === 0) return collatz(n / 2)
  if (n % 2 === 1) return collatz(3 * n + 1)
}
collatz(250)

テストはこちら 次に Observable で実装してみます。

import { from, of, empty } from 'rxjs'
import { expand } from 'rxjs/operators'
of(200)
  .pipe(
    expand(n => {
      if (n === 1) return empty()
      if (n % 2 == 0) {
        return of(n / 2)
      } else {
        return of(3 * n + 1)
      }
    })
  )
  .subscribe(v => console.log(v))

テストはこちら

終わりに

いかがだったでしょうか？
他にもいくつか RxJS についての記事を書いていますのでみていただけると嬉しいです。