Web API のページネーションへの FS2 Stream の応用

先日、Qiita に上がった記事『[ページネーションを利用したAPIと再帰関数の相性が抜群だった](https://qiita.com/DaisukeKURATA/items/d3a74bd3dbef6a6473e0)』の中で、無限リストを使えないかという面白そうな[お題](https://qiita.com/DaisukeKURATA/items/d3a74bd3dbef6a6473e0#2-%E3%81%A9%E3%81%86%E3%81%9B%E5%BC%95%E6%95%B0%E3%81%A0%E3%81%91%E3%81%A7%E8%A1%A8%E7%8F%BE%E3%81%97%E3%81%AA%E3%81%84%E3%81%AA%E3%82%89-%E7%84%A1%E9%99%90%E3%83%AA%E3%82%B9%E3%83%88%E4%BD%BF%E3%81%A3%E3%81%A6%E3%81%84%E3%81%84%E6%84%9F%E3%81%98%E3%81%AB%E3%81%A7%E3%81%8D%E3%81%AA%E3%81%84%E3%81%8B%E3%81%97%E3%82%89)があった。ちょうど何か FS2 を使った記事を書きたかったので、今回はそれをやってみる。

## お題と方針
Google Calendar API を使って、以下のような趣向でカレンダーのイベントリストを取得してみる。

- Google で公開されている「日本の祝日」カレンダー[^calendar_id]を対象とする。
- 祝日（=イベント）の数が十分に多い期間を指定、または期間指定なしで検索する。
- 5件ごとのページネーションで12件まで取得する。
- 祝日ストリームに終わりがあるかどうか分からない「てい」でやってみる。つまりページネーションを繰り返して全件取得してから12件だけ取り出すことは避けて、なるべく必要な分だけの API 呼び出しに抑える。最短では、5件 + 5件 + 2件 = 12件で 3回の呼び出しになる。
- Scala 標準の Stream と、FS2 の Stream の2つの方式で実装してみる。

※ 使用するライブラリのバージョンは[このあたり](https://github.com/yasuabe/fs2-trials/blob/master/build.sbt)
[^calendar_id]: Calendar ID = ja.japanese#holiday@group.v.calendar.google.com

## 実装
見通しを良くするため、Java で書かれた Google Calendar API へのアクセスは、以下のような専用のモジュールに隔離しておく[^null_excuse]。使う側では `calendar.retrieveEvents(...)` を、`EventsFunc` 型（下記コード参照）の関数として扱う形になる。

```scala:CalendarApiOps.scala
object CalendarApiOps {
  import collection.JavaConverters._

  ... 略

  val calendar: Calendar = new Calendar.Builder(httpTransport, jacksonFactory, credential)
    .setApplicationName(appName)
    .build()

  type PageToken  = String
  type EventsFunc = Option[PageToken] => Events

  implicit class CalendarOps(val cal: Calendar) extends AnyVal {
    def retrieveEvents(calendarId: String, modifier: EventsModifier = identity): EventsFunc =
      maybeToken => {
        println("********* calling API ***********") // カレンダーAPI呼び出し確認用
        modifier(cal.events().list(calendarId))
          .setTokenOrNull(maybeToken)
          .setMaxResults(5) // 5件ずつ Event を得る
          .execute()
      }
  }
}
```
[ソース: CalendarApiOps.scala](https://github.com/yasuabe/fs2-trials/blob/master/calendar_api_sample/src/main/scala/misc/CalendarApiOps.scala)

[^null_excuse]: クラスパスに credentials.json がないと NullPointerException が発生したりするが、そのあたりは適当に妥協した。

### Scala 標準 Stream 版
Scala 標準 `Stream` は Haskell の `Data.List` におおよそ対応するが、Haskell の方では以下のような関数 [unfoldr](http://hackage.haskell.org/package/base-4.12.0.0/docs/Data-List.html#g:9) がある[^project_euler]。簡単に言うと、リストを畳み込んで値を得る `fold` と逆に、ある値から出発してリスト（無限かもしれない）を生成する働きの関数になるが、詳しいことはリンク先を読むとわかる。

[^project_euler]: 余談ながら Project Euler のような数学系のプログラミングパズルだと、お題となる無限数列から何かを探したり、合計したりする問題が多いので、Haskell で挑戦していると `unfoldr` は結構多用する。

```haskell
unfoldr :: (b -> Maybe (a, b)) -> b -> [a]
```
ここでは Haskell 版のシグネーチャを参考にして定義しておく[^scalaz_unfold]。

```scala
def unfoldr[B, A](f: B => Option[(A, B)], b: B): Stream[A] = f(b) match {
  case Some((a, newB)) => a #:: unfoldr(f, newB)
  case None            => Stream.empty
}
```
[^scalaz_unfold]: Scalaz では同様の関数が提供されている

これを用いて祝日ストリーム `Stream[Event]` を得る関数は以下のようになる。見通しを得るため `Try` などでラップせずに直接 `EventsFunc` 型の関数を評価してみた。

```scala
def eventStream(f: EventsFunc): Stream[Event] = {
  val g: Option[Events] => Option[Events] = {                    // 前回の f の結果にマッチ
    case Some(e) if !e.hasNext => None                           // 前回トークンがなければ終了
    case _                     => f(prev >>= (_.nextToken)).some // unfoldr を継続
  }
  unfoldr[Option[Events], Events](g(_).map(e => (e, e.some)), None)
    .flatMap(_.items.toStream)
}
```

以下のようなコードを実行して標準出力を見てみると、、、

```scala
eventStream(calendar.retrieveEvents(calendarId, _.setStartDate(start).setEndDate(end)))
  .take(12)
  .map(e => s"start:${e.getStart}, summary: ${e.getSummary}")
  .foreach(println)
```
12件のみが3回の呼び出しで取得されていることが確認できる。

```
********* calling API ***********
start:{"date":"2018-01-01"}, summary: 元日
start:{"date":"2018-01-08"}, summary: 成人の日
start:{"date":"2018-02-11"}, summary: 建国記念の日
start:{"date":"2018-02-12"}, summary: 建国記念の日 振替休日
start:{"date":"2018-03-21"}, summary: 春分の日
********* calling API ***********
start:{"date":"2018-04-29"}, summary: 昭和の日
start:{"date":"2018-04-30"}, summary: 昭和の日 振替休日
start:{"date":"2018-05-03"}, summary: 憲法記念日
start:{"date":"2018-05-04"}, summary: みどりの日
start:{"date":"2018-05-05"}, summary: こどもの日
********* calling API ***********
start:{"date":"2018-07-16"}, summary: 海の日
start:{"date":"2018-08-11"}, summary: 山の日

Process finished with exit code 0
```

[ソース: UnfoldMain.scala](https://github.com/yasuabe/fs2-trials/blob/master/calendar_api_sample/src/main/scala/plain_unfold/UnfoldMain.scala)

### FS2 の Stream 版
次に Doobie や http4s などでも使われている、関数型ストリーミングライブラリ FS2 の `Stream` を試してみる。ストリームの型は、エフェクト`F[_]` を加味して `fs2.Stream[F, Event]` のようになるが、`F[_]` は Cats Effect の `Sync`、`ContextShift` のインスタンスを持つ必要がある。

前節で書いた `unfoldr` 相当の関数は `Stream` のメソッドとして最初から提供されていて、`F[_]`を除けば上で書いた Scala標準 Stream 版の `eventStream` とほぼ同じコードで祝日ストリームを得ることができる。

```scala
implicit val SyncF: Sync[F]
implicit val ContextShiftF: ContextShift[F]

def eventStream(f: EventsFunc): Stream[F, Event] = {
  val g: Option[Events] => Option[Events] = {
    case Some(e) if !e.hasNext => None
    case prev                  => f(prev >>= (_.nextToken)).some
  }
  Stream.unfold[F, Option[Events], Events](None)(g(_).map(e => (e, e.some)))
        .flatMap(e => Stream(e.items: _*))
}
```
このストリームから 12個の祝日を取り出して標準出力に書き出すために、通常の `println` を使えなくもないが、せっかくなので FS2 の `fs2.io.stdin` と Cats Effect の `Resource` を使ってみる。以下のようなコードになる。

```scala
def blockingEC: Resource[F, ExecutionContextExecutorService] =
  Resource.make(
    SyncF.delay(ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(2)))
  )(ec => SyncF.delay(ec.shutdown()))

def program(calendarId: String)(modifier: EventsModifier): Stream[F, Unit] =
  Stream.resource(blockingEC) flatMap { ec =>
    eventStream(calendar.retrieveEvents(calendarId, modifier))
      .take(12)
      .map(e => s"start:${e.getStart}, summary: ${e.getSummary}\n")
      .through(text.utf8Encode)
      .through(io.stdout(ec))
  }
```
実行するには `F[_]` を具体的に与える必要があるが、例えば `IO` だとすれば以下のように書ける。

```scala
object FS2main extends IOApp with Program[IO] {
  val SyncF:         Sync[IO]         = Sync[IO]
  val ContextShiftF: ContextShift[IO] = ContextShift[IO]
  ...

  def run(args: List[String]): IO[ExitCode] =
    program(calendarId)(_.setStartDate(start).setEndDate(end))
      .compile
      .drain
      .as(ExitCode.Success)
}
```
動かしてみると、ほとんど同じような実装ながら FS2 の場合一回だけ多く先読みして、都合 4回呼ばれているのがわかる（コメントは後で書き足した）。

```
********* calling API ***********     // 1) '18年元旦から'18年春分の日まで
********* calling API ***********     // 2) '18年昭和の日から'18年こどもの日まで
start:{"date":"2018-01-01"}, summary: 元日
start:{"date":"2018-01-08"}, summary: 成人の日
start:{"date":"2018-02-11"}, summary: 建国記念の日
start:{"date":"2018-02-12"}, summary: 建国記念の日 振替休日
start:{"date":"2018-03-21"}, summary: 春分の日
********* calling API ***********     // 3) '18年海の日から'18年秋分の日まで
start:{"date":"2018-04-29"}, summary: 昭和の日
start:{"date":"2018-04-30"}, summary: 昭和の日 振替休日
start:{"date":"2018-05-03"}, summary: 憲法記念日
start:{"date":"2018-05-04"}, summary: みどりの日
start:{"date":"2018-05-05"}, summary: こどもの日
********* calling API ***********     // 4) '18年体育の日から'18年天皇誕生日振替休日まで
start:{"date":"2018-07-16"}, summary: 海の日
start:{"date":"2018-08-11"}, summary: 山の日

Process finished with exit code 0
```
[ソース: FS2Main.scala](https://github.com/yasuabe/fs2-trials/blob/master/calendar_api_sample/src/main/scala/with_fs2/FS2Main.scala)

## 謝辞
@DaisukeKURATA お題お借りしました。ありがとうございました。

[^calendar_id_how_to]: [カレンダーIDの取得の仕方](https://docs.simplecalendar.io/find-google-calendar-id/)