Scala3: IO と Stream に関する覚書

浅い理解で書き散らかした自分用メモ。

基本
----

### cats.effect.IO と fs2.Stream の関係

fs2 の内部で cats-effect を利用しているため、シームレスな連携が実現できている（らしい）。

### 依存など

```scala
    libraryDependencies ++= Seq(
      "org.typelevel" %% "cats-core" % "2.12.0",
      "org.typelevel" %% "cats-effect" % "3.5.4",
      "co.fs2" %% "fs2-core" % "3.11.0"
    ),
    initialCommands := s"""
      import cats.syntax.all.*, cats.effect.IO
      import cats.effect.unsafe.implicits.global
      import fs2.*
      import scala.concurrent.duration.*, java.util.concurrent.*
    """
```

IO
--

### IO とは

「副作用があり得る計算」を表す「値」である。ただの値なので、定義されただけでは実コードは実行（計算）されない。明示的に実行すると計算され、成功すると結果が返る（失敗すると例えば例外がスローされる）。遅延評価される [Future](https://docs.scala-lang.org/overviews/core/futures.html) と捉えてもよい。

* 同期計算あるいは非同期計算を記述できる
* 内部処理をキャンセル可能にすることもできる

次の２種類の値（状態）を持つ。概念としては Option 型が Some と None から成るのに近い。

#### Pure: 純粋関数

```scala
IO.pure(5 * 2)    // IO(10)
```

与えたコード（のブロック）は即時実行される。ただし返される値は IO 型であり、まだ「結果」は出ていないことに注意。ここでは、`5 * 2` 自体は即時計算されたが、結果の `IO.pure(10)` はただの「副作用があり得る計算」を表す「値」に過ぎない。

#### Delay: 遅延実行する関数

```scala
IO.delay(println("hello"))    // IO(...)
IO.delay(5 / 0)    // IO(...)    まだ実行されていないのでゼロ除算例外は発生しない
```

与えたコード（のブロック）は、あとで「実行」されるまで評価されない。

### IO を定義（生成）する

`IO.pure()` や `IO.delay()` を直接呼んでもいいが、実運用上は、あえて Pure であることを明示したい場合でなければ、一貫して `IO()` や `IO {}` で生成してもよいだろう。

```scala
import cats.effect.IO

val io: IO[Int] = IO(10 * 2)    // IO(...)
val io: IO[String] = IO { "hello" }    // IO(...)
val io: IO[Unit] = IO { println("hello") }    // IO(...)
```

ちなみに `IO(10 * 2)` と `IO.pure(10 * 2)` の結果型は違う（実用上の差違はほぼないだろうが）。

### IO を実行する

実行コンテキストが必要なのでインポートしておく。

`unsafeRunSync()` を呼ぶと、IO が実行され計算された値が返る。

```scala
import cats.effect.unsafe.implicits.global

IO(5 * 2).unsafeRunSync()    // Int = 10
IO { Seq(1, 2, 3).map(_ * 2) }.unsafeRunSync()    // Seq[Int] = List(2, 4, 6)
```

ほかにもいくつか `unsafeRunXXX` 関数が[ある](https://typelevel.org/cats-effect/api/3.x/cats/effect/IO.html)。

### IO 実行時に発生する例外を処理する

しかし `unsafeRunSync()` を生で実行すると、コードを実行中に発生した例外は普通にリスローされてしまう。

```scala
IO(5 / 0).unsafeRunSync()    // java.lang.ArithmeticException: / by zero
```

そこで `attempt()` を通し結果を Either に包むと便利。

```scala
IO(5 / 1).attempt.unsafeRunSync()    // Right(5)
IO(5 / 0).attempt.unsafeRunSync()    // Left(java.lang.ArithmeticException: / by zero)
```

### エラーハンドリング

`handleErrorWith()` が使える。副作用にしか興味がない場合に便利かもしれない。

```scala
IO(5 / 0)    // ゼロ除算！
.handleErrorWith { error =>
  IO.pure(error.getMessage)
}
.unsafeRunSync()    // Int | String = / by zero
```

### 複数の IO を合成する

IO 同士を `flatMap()` 等を使って合成することができる。

```scala
val io: IO[Int] = IO(5).flatMap { n => IO(n * 2) }
```

通常は、複数の IO を for 文で組み上げていくことになるだろう。

```scala
val io: IO[Int] =
  for
    n <- IO(5)
    _ <- IO.println("calculating..")
    result <- IO(n * 2)
  yield result
```

複数の IO がある場合は `mapN()` が使える。

```scala
import cats.syntax.all.*

val io: IO[Int] = (IO(10), IO(20), IO(30)).mapN { (a, b, c) => a + b + c }
io.unsafeRunSync()    // Int = 60
```

次の例は、順に実行するが、結果は最後の IO のものだけを返す。

```scala
import cats.syntax.all.*

val io: IO[Int] = IO.println("hello") *> IO(42)
io.unsafeRunSync()    // Int = 42
```

なお、途中の計算結果は（副作用を除いて）捨てられる。

```scala
val io: IO[Int] = IO(10) *> IO(20) *> IO(30)
io.unsafeRunSync()    // Int = 30
```

リストを順次実行し、結果のリストを返す。

```scala
val io: IO[List[Int]] = List(IO(10), IO(20), IO(30)).sequence
io.unsafeRunSync()    // List[Int] = List(10, 20, 30)
```

:::note info
IO に対しては >> も *> とほぼ同様に動作する。ただし、IO に留まらない広範な型を扱う際には、より（型的に）広い意味を持つ *> が望ましいようだ。とはいえ、IO くらいしか利用しないケースでは可読性を優先して >> で統一してもいいかもしれない。
:::

#### 方針

* 処理を小さく分割し IO に包む
* 複数の IO を合成して１つの IO を作る
* メイン処理（ビジネスロジック）の最後に１回だけ `unsafeRunSync()` を実行する

### 入れ子になった IO をまとめる

```scala
val doubleIO: IO[IO[String]] = IO { IO("hello") }    // 入れ子
val io: IO[String] = doubleIO.flatten
```

`flatMap()` も使える。

```scala
val io: IO[String] = doubleIO.flatMap(identity)
```

### Future から変換する

既存の Future を IO に変換し遅延実行できる。

```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext
import scala.concurrent.ExecutionContext.Implicits.global

def sum(list: List[Int])(implicit ec: ExecutionContext): Future[Long] =
  println("calc sum!")
  Thread.sleep(500)
  Future(list.foldLeft(0L)(_ + _))

val io: IO[Long] = IO.fromFuture(IO { sum(List(1, 2, 3, 4, 5)) })
```

### 先に完了した IO の結果だけ欲しい (race)

`race()` が使える。先に終わらなかったタスク（敗者）はキャンセルされる（Stream などで更に計算が続く場合）。

```scala
val task1 = IO.sleep(100.millis) *> IO(10)
val task2 = IO.sleep(200.millis) *> IO(20)

val io: IO[Either[Int, Int]] = IO.race(task1, task2)

io.flatMap {
  case Left(value)  => IO.println(s"Left: $value")
  case Right(value) => IO.println(s"Right: $value")
}.unsafeRunSync()    // Left(10)
```

### 並列実行させる (parSequence)

`parSequence()` が使える。それぞれの IO を並列実行し、すべてが終了するまで待機する（結果を返す）。固定長のタプルには `parTupled()` が使える。

```scala
def buildTask(i: Int): IO[Int] =
  IO.println(s"task $i") *> IO.sleep(2.seconds) *> IO { i * 10 }

Seq(1, 2, 3)
  .map(buildTask)
  .parSequence
  .unsafeRunSync()    // List(10, 20, 30)
```

並列数を指定したい場合は `Stream.parEvalMap()` 等を使う（後述）。

### 後処理を確実に実行したい (bracket)

`bracket()` を使えば try-finally 的な処理を書けて便利。

```scala
val io: IO[Int] =
  IO(10).bracket { in =>
    // try ブロックと見なせる
    IO.println("started") *> IO { in + 5 }
  } { _in =>
    // finally ブロックと見なせる
    IO.println("ended") *> IO.sleep(200.millis)
  }
```

より細かく制御したいなら Rsource 型というのがある。

### タイムアウトさせる (timeoutTo)

`timeout()` もあるが、タイムアウト時の処理を IO で指定できる `timeoutTo()` が便利そうだ。

```scala
val slowIO: IO[Int] = IO.sleep(5000.millis) *> IO { 42 }
val io = slowIO.timeoutTo(2.seconds,
  IO.raiseError(new RuntimeException("Operation timed out!")))
```
`Thread.sleep()` のようにスレッドをブロックしてしまう処理がある場合は Fiber を使ったより複雑な手法が必要。

細かく制御したい場合は次の手法もある。

```scala
import cats.effect.Temporal

val io = Temporal[IO].timeout(slowTask, 2.seconds)
```

### SyncIO

同期処理に限定された IO 型。

```scala
import cats.effect.SyncIO

val io: SyncIO[Int] = SyncIO { 42 }
```

`SyncIO` は `IO` と合成できず、しかも `unsafeRunSync()` でしか実行できない（`unsafeRunAsync()` 等は使えない）。短い処理で、パフォーマンスを出したい時に使うようだ。

IO と混ぜるには、IO に変換する必要がある。

```scala
val io: IO[Int] = SyncIO(42).to[IO]
```

### 参照 (Ref)

`IO.ref[A]` は、型 `A` を保持する可変な「参照（Ref）」を `IO` の中で作成する。参照はスレッドセーフで、安全に更新できる。つまり、スレッド間で共有可能な「状態」を**純粋に**扱える。

```scala
import cats.effect.Ref

val ref: IO[Ref[IO, Int]] = IO.ref[Int](10)
```

ちなみに、`Ref` は型パラメータを持ち、エフェクト型は `IO` に限定されない。

```scala
import cats.effect.SyncIO

val io: IO[Ref[IO, Int]] = Ref[IO].of(10)
val io: SyncIO[Ref[SyncIO, Int]] = Ref[SyncIO].of(10)
```

`get()` で現在の値を取得する。`update()` で値を更新する。

```scala
val io =
  for
    ref <- IO.ref(10)
    _ <- ref.update(_ + 1)
    value <- ref.get
    _ <- IO.println(value)    // 11
  yield ()
```

`modify()` は、現在の値に基づいて新しい値を計算し、同時に他の副作用を発生させることができる。

```scala
val io =
  for
    ref <- IO.ref(10)
    result <- ref.modify { oldValue =>
      val newValue = oldValue * 2
      
      // 副作用の例
      println(s"Updating counter from $oldValue to $newValue")

      (newValue, oldValue)       // 新しい値と戻り値を返す
    }
    _ <- IO.println(result)      // 元の値 (10)
    newValue <- ref.get
    _ <- IO.println(newValue)    // 更新後の値 (20)
  yield ()
```

複数のスレッド（ファイバー）で並列更新する例。並列でも正しい結果が得られる。

```scala
val io =
  for
    ref <- IO.ref(0)
    _ <- List.fill(1000)(ref.update(_ + 1)).parSequence
    value <- ref.get
    _ <- IO.println(value)    // 1000
  yield ()
```

安全なカウンターやフラグ（例：Stream を打ち切る）として使われることが多い。

Stream
------

以下は [fs2.Stream](https://www.javadoc.io/doc/co.fs2/fs2-docs_2.13/latest/fs2/Stream.html) について。Scala 組み込みの旧 [Stream](https://www.scala-lang.org/api/3.x/scala/collection/immutable/Stream.html) (現 LazyList) とは別モノなので注意。

一般化した型として `Stream[F, O]` と書ける。`F` はエフェクト型、`O` はアウトプット型である。具体的には、`Stream[Pure, Int]` や `Stream[IO, Int]` といった形で表現することになる。

### 基本

#### 作成する

```scala
val s: Stream[Pure, String] = Stream.emit("hello")    // 要素は１つのみ
val s: Stream[Pure, String] = Stream.constant("hello")    // 無限に続く。結果は List[String]

val s: Stream[Pure, Int] = Stream(1, 2, 3)    // 要素は３つのみ
val s: Stream[Pure, String] = Stream("apple", "banana", "chocolate")    // 同上
val s: Stream[Pure, Int] = Stream(1, 2, 3).repeat    // 無限に続く。結果は List[Int]
val s: Stream[Pure, Int] = Stream(1, 2, 3).repeatN(5)    // 1, 2, 3 が５回繰り返される

val numbers: Stream[Pure, Int] = Stream(1, 2, 3, 4, 5)
val oddNumbers: Stream[Pure, Int] = numbers.filter(_ % 2 != 0)
```

もちろんファイルや標準入力などからも作成できるが省略。

#### 評価する

Stream は無限に値を生成し続ける（可能性がある）ので、実際に計算する際には `take()` 等で有限の値に落とし込んでから処理するケースが多いだろう。

```scala
val result: List[Int] = Stream(1, 2, 3).repeat.take(5).toList    // List(1, 2, 3, 1, 2)
```

### 合成する

```scala
// 結合する
(Stream(1, 2, 3) ++ Stream(4, 5, 6) ++ Stream(7, 8, 9)).take(20).toList
(oddNumbers ++ Stream(0).repeat).take(10).toList

Stream(1, 2, 3).map(_ * 2).toList    // List[Int] = List(2, 4, 6)
```

### IO[A] のストリームを作成して実行する

Stream を使い、連続する IO の処理を、単一の IO に「まとめる」ことができる。

#### 1. IO に包まれた値を Stream に変換する

```scala
// すこし時間のかかる API があるとする
def apiCall: IO[Int] = IO.println("api call") *> IO.sleep(500.millis) *> IO { 10 }

val stream: Stream[IO, Int] = Stream.eval(apiCall).repeat
```

つまり、`eval()` は `F[A]` を `Stream[F, A]` へ変換する。

#### 2. さらに Stream を単一の IO へ変換する

```scala
val io: IO[List[Int]] = stream.take(5).compile.toList
```

`compile()` は `Stream[F, A]` を `Stream.CompileOps[F, F, A]`へ変換する。
`toList()` はこれを `F[List[A]]` へ変換する。

この例では、結果的に `IO[Int]` を `IO[List[Int]]` に変換し、さらに内部で `Stream` が使われていることを（型の上で）隠蔽できている。

#### 実行する

約 2.5 秒かかる。

```scala
scala> io.unsafeRunSync()
api call
api call
api call
api call
api call
val res5: List[Int] = List(10, 10, 10, 10, 10)
```

最終的に実行するのは単一の IO なので、要素中のどこかの API 呼び出しが失敗すればその時点で例外が発生する。

```scala
// 一定の割合で失敗する API
def unstableApiCall: IO[Int] =
  if scala.util.Random.nextInt(10) == 0 then
    IO.raiseError(new Exception("something's wrong"))
  else
    apiCall

Stream.eval(IO.defer(unstableApiCall)).repeat.take(5).compile.toList.attempt.unsafeRunSync()
```

なお、ここでは `unstableApiCall()` が即時実行（され結果をキャッシュ）されると困るので `IO.defer()` で包み遅延実行させる必要がある。

### リトライさせる (retry)

`.repeat.take(n)` するよりスマート。

```scala
val stream: Stream[IO, Int] = Stream.retry(
  IO.defer { unstableApiCall },
  delay = 1.second,        // リトライ間隔
  nextDelay = _ * 2,       // リトライごとに遅延を倍増
  maxAttempts = 5,         // 最大リトライ回数
  retriable = _ => true    // 常にリトライする
)
```

### 副作用しか必要ない場合の実行 (drain)

`drain()` を呼べば、返値の型は `IO[Unit]` となる。

```scala
def task: IO[Unit] = IO.println("hello") 

Stream.eval(task).repeat.take(3).compile.drain.unsafeRunSync()
```

### Stream の各要素を「値を IO へ変換する関数」でラップする (evalMap)

```scala
val wordLen: String => IO[Int] = (s: String) => IO { s.length }

val stream: Stream[IO, Int] = Stream("hello", "world").evalMap(wordLen)
stream.compile.toList.unsafeRunSync()    // List[Int] = List(5, 5)
```

### Either から作成する (fromEither)

```scala
val rightStream: Stream[IO, String] = Stream.fromEither[IO]("hello".asRight)
val leftStream: Stream[IO, String] = Stream.fromEither[IO](Exception("woa!").asLeft)
```

### Pipe で Stream を変換する (through)

Pipe を定義する。ここでは IO だけではなく他の型も渡せるよう `F[*]` で定義している。

```scala
def doublePipe[F[*]]: Pipe[F, Int, Int] =
  (inStream: Stream[F, Int]) => inStream.map(_ * 2)
```
Stream の例として `Stream[IO, Int]` を用意する。`.covary[IO]` は `Stream[Pure, A]` を `Stream[IO, A]` へ変換してくれる。

```scala
val input: Stream[IO, Int] = Stream.emits(List(1, 2, 3)).covary[IO]
```

パイプを適用するには `through()` を使う。

```scala
val result: Stream[IO, Int] = input.through(doublePipe)
```

### Option から成る Stream から None を取り除く (unNone)

```scala
Stream(Some(10), None, Some(20)).unNone.toList    // List[Int] = List(10, 20)

// IO に変換する例
Stream(Some(10), None, Some(20))
  .covary[IO]
  .unNone
  .compile.toList.unsafeRunSync()
```

`None` が現れたら Stream を打ち切る `unNoneTerminate()` もある。

### 途中のエラーをリカバリする (handleErrorWith)

`handleErrorWith()` を挟むことができる。

```scala
Stream(1, 2, 3, 4)
  .append(Stream.emit(5 / 0))    // ゼロ除算！
  .handleErrorWith { _e =>
    Stream.eval(IO(999))
  }
  .append(Stream(6, 7, 8))
  .compile.toList.unsafeRunSync()    // List(1, 2, 3, 4, 999, 6, 7, 8)
```

### 終了時に処理を実行する (onFinalize)

```scala
Stream(1, 2, 3)
  .covary[IO]
  .onFinalize(IO.println("stream finalized!"))
```

### 処理中にキャンセルさせる (interruptWhen)

`Stream.interruptWhen()` に条件を設定すれば処理中にストリーム全体をキャンセルできる。

```scala
import cats.effect.Temporal
import fs2.concurrent.SignallingRef

val task = Stream.eval(IO.println("run") *> IO.sleep(1.second)).repeat

val io =
  for
    cancel <- SignallingRef[IO, Boolean](false)
    s <- task.interruptWhen(cancel).compile.drain.start
    _ <- Temporal[IO].sleep(5.seconds) *> cancel.set(true)
  yield s

io.unsafeRunSync()
```

### 定期実行させる (awakeEvery)

`awakeEvery()` は指定した間隔ごとに正確にイベントが発生する。また、経過時間を出力してくれる。

```scala
def showTime(elapsed: FiniteDuration) = IO.println(elapsed)

Stream.awakeEvery[IO](1.second)
  .evalMap(showTime)
  .take(5).compile.drain.unsafeRunSync()
```

### 遅延させる (fixedDelay)

`awakeEvery()` と似ているが、`fixedDelay()` は各要素が発行された後に遅延が始まるのでズレが発生（累積）する可能性がある。

```scala
val greet = IO.println("hello")

Stream.fixedDelay[IO](1.second)
  .evalMap(_ => greet)
  .take(5).compile.drain.unsafeRunSync()
```

### スライドしながら出力する (sliding)

指定したウィンドウサイズで「スライド」しながら出力する。

```scala
val stream: Stream[Pure, Int] = Stream.iterate(1)(_ + 1)    // 1, 2, 3, 4, 5, ... の無限ストリーム
stream.sliding(3).map(chunk => chunk.toList.sum).take(5).toList
// List(6, 9, 12, 15, 18)
// つまり: List((1 + 2 + 3), (2 + 3 + 4), (3 + 4 + 5), (4 + 5 + 6), (5 + 6 + 7))
```

この例では３つずつの `Chunk` にまとめられる。`Chunk` は Stream 内部で効率的に値を扱うための型で、ここでは `toList()` でリストに変換してから処理している。

移動平均やパターン検出、サンプルの分割などに便利。

Stream の相互作用
---------------

### 交互に配置する (interleave)

`interleave()` は２つのストリームの要素を交互に取り出して１つのストリームに結合する。同期的。

```scala
val s1: Stream[IO, Int] = Stream(1, 2, 3, 4, 5).covary[IO]
val s2: Stream[IO, Int] = Stream(100, 200, 300).covary[IO]

(s1 interleave s2).compile.toList.unsafeRunSync()    // List(1, 100, 2, 200, 3, 300)
```

短い方のストリームに合わせて終了する（余ったほうの値は捨てられる）。基本的に、２つのストリームが同じ長さであることが前提。余ったほうのストリームも出力したければ `interleaveAll()` を使う。

```scala
(s1 interleaveAll s2).compile.toList.unsafeRunSync()    // List(1, 100, 2, 200, 3, 300, 4, 5)
```

到着順（＝非同期）に出力したい場合は `merge()` を使うこと。

### マージする (merge)

２つのストリームを並行して処理し、出力可能な要素があればどちらからでもすぐに取り出す。

```scala
val s1: Stream[IO, String] = Stream.awakeEvery[IO](1.second).map(_ => "s1")
val s2: Stream[IO, String] = Stream.awakeEvery[IO](500.millis).map(_ => "s2")

(s1 merge s2).take(5).compile.toList.unsafeRunSync()    // List(s2, s1, s2, s2, s1)
```

片方のストリームの速度が遅くても、もう片方のストリームが要素を出力できるので非同期処理に向いている。終了条件をコントロールできる `mergeHaltBoth()` や `mergeHaltL()`、`mergeHaltR()` などもある。

かならず交互（＝同期的）に出力したい場合は `interleave()` を使うこと。

### 結合する (zip)

２つのストリームを並行して進め、それぞれのストリームの要素をタプルにして結合する。

```scala
val s1: Stream[IO, Int] = Stream(1, 2, 3, 4, 5).covary[IO]
val s2: Stream[IO, Int] = Stream(100, 200, 300).covary[IO]

(s1 zip s2).compile.toList.unsafeRunSync()    // List((1,100), (2,200), (3,300))
```

実際には、さらに値を加工するケースが多いだろう。

```scala
val stream: Stream[IO, Int] = (s1 zip s2).evalMap { case (a, b) => IO(a + b) }
stream.compile.toList.unsafeRunSync()    // List(101, 202, 303)
```

短い方のストリームに合わせて終了する。基本的に、２つのストリームが同じ長さであることが前提。余ったほうのストリームも出力したければ `zipAll()` を使う。デフォルト値を指定する必要がある。

```scala
s1.zipAll(s2)(-1, -1).compile.toList.unsafeRunSync()    // List((1,100), (2,200), (3,300), (4,-1), (5,-1))
```

### 左結合する (zipLeft)

`zip()` した結果から、左側のストリームの要素だけを取り出す。

```scala
val s1: Stream[IO, Int] = Stream(1, 2, 3, 4, 5).covary[IO]
val s2: Stream[IO, Unit] = Stream.eval(IO.println("waiting") *> IO.sleep(500.millis))
  .repeatN(10)

(s1 zipLeft s2).compile.toList.unsafeRunSync()    // List(1, 2, 3, 4, 5)
```

右側のストリームは消費されるが結果には反映されない。つまり、副作用を実行するが結果は必要ない場合に使える。この例のようにストリームの速度を調整するのに便利。ウェイトを入れるなら次のようなコードでもいいかもしれない。

```scala
val s2: Stream[IO, Unit] = Stream.fixedDelay[IO](500.millis)
  .evalMap(_ => IO.println("waiting")).take(10)
```

右側のストリームの要素だけを取り出す `zipRight()` もある。

### 並列実行する (parJoin)

複数の非同期タスクの出力を１つのストリームにまとめる。効率よく並列処理ができる。

```scala
val s1: Stream[IO, Int] = Stream.eval(IO.sleep(100.millis) *> IO { 10 }).repeatN(5)
val s2: Stream[IO, Int] = Stream.eval(IO.sleep(200.millis) *> IO { 20 }).repeatN(5)
val s3: Stream[IO, Int] = Stream.eval(IO.sleep(300.millis) *> IO { 30 }).repeatN(5)

Stream(s1, s2, s3).parJoin(2).compile.toList.unsafeRunSync()
// List(10, 20, 10, 10, 20, 10, 10, 20, 20, 30, 20, 30, 30, 30, 30)
```

ストリームのストリーム `Stream[F, Stream[F, O]]` に対して適用される。内部のストリームを最大 n 個まで並列実行し、結果を１つのストリームとして出力する。

完了順は、各ストリームが要素を生成・完了するタイミングに依存し、一般に不規則な順序となる。

### 並列実行する (parEvalMap)

`parEvalMap()` を使って、各 IO の完了を待たずに、並列数を指定して逐次処理する。

```scala
def buildTask(i: Int): IO[Int] =
  IO.println(s"task $i") *> IO.sleep(1.second) *> IO { i * 10 }

Stream
  .emits(Seq(1, 2, 3, 4, 5))
  .parEvalMap(maxConcurrent = 2)(buildTask)
  .compile
  .toList
  .unsafeRunSync()    // List(10, 20, 30, 40, 50)
```

`parEvalMapUnordered()` もある。なお、並列数を指定する必要がないなら `IO.parSequence()` でよい。

Scala Native で使う
-------------------

2024 年現在、Cats Effect は Scala Native 0.4 系で動く。ただし、シングルスレッド上で動作する関係上、 `unsafeRunSync()` は存在しない。`unsafeRunAsync()` あたりを代わりに使うことになるだろう。Either を受け取るコールバック関数を渡す。

```scala
IO(42).unsafeRunAsync((result) => println(result))    // Right(42)
```

マルチスレッド化した Scala Native 0.5 系に対応してくれれば、おそらく `unsafeRunSync()` も提供されると期待している。が、[だいぶ先](https://github.com/typelevel/cats-effect/issues/4076#issuecomment-2131135348)になる予定だ。

TODO
----

* キャンセル可能な IO (Cancelable)