サーバーサイドKotlinにおけるCoroutinesとReactor(Reactive Streams)の相互運用ざっくりまとめ

`Coroutines`と`Reactor`(`Reactive Streams`)の相互運用について、`Java`ベースのノンブロッキングなフレームワークを使ってサーバーサイド`Kotlin`をやる上でまず最初に必要かなと思っている部分をまとめます。
誤っている部分が有りましたらご指摘下さい。

# 頻出の技術
`Spring WebFlux`を前提に、以下3つの技術をざっくり解説します。

- `Reactive Streams`
- `Project Reactor`
- `Coroutines`

## `Reactive Streams`
`Reactive Stream`は、非同期ストリーム処理を実現するためのオープンな仕様です。
`Java 9`からは`java.util.concurrent.Flow`として`JDK`に取り込まれました（これは1対1対応する`API`としてであって、クラス名等は後述する内容と異なっています）。

https://www.reactive-streams.org/

### 頻出のクラス
サーバーサイド`Kotlin`をやっていてよく登場するクラスは`Publisher`です。
ただし、サーバーサイド`Kotlin`をやる場合は`Reactor`の実装や`coroutine`に変換することが多く、経験上これらを直接触ることは少なかったです。

https://javadoc.io/doc/org.reactivestreams/reactive-streams/latest/org/reactivestreams/Publisher.html

また、`Project Reactor`が導入されている場合、後述する`Mono`や`Flux`の方がより具体的な表現になるため、アプリケーションのコードではなるべく`Publisher`を書かない方が良いと思っています。

## `Project Reactor`
`Project Reactor`は、`Reactive Streams`の実装の1つです。
有名どころとして、`Spring WebFlux`は`Project Reactor`をベースに開発されています。

https://projectreactor.io/

### 頻出のクラス
`Project Reactor`を使っていてよく登場するクラスは`Mono`と`Flux`です。
両者ともに前述した`Publisher`の実装で、値を1つ返す場合は`Mono`、複数返す場合は`Flux`を利用します。

誤解を招きそうな表現ですが、`Mono`は単なる値、`Flux`は`List`や`Stream`のようなものです（どちらかと言えば`Stream`の方が近いかも）。

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html

### 頻出の記法
`Kotlin`を使う場合`coroutine`に変換してしまうのが簡単ですが、基盤部分の実装では`Mono`・`Flux`に触れなければならない場面も有ります。
詳しい部分はクラスの`Javadoc`から探すべきですが、以下の操作を覚えておけば大体の部分は読み書きできると思います。

- `Mono#map`: `Mono`内の値を別の値に変換する
- `Mono#flatMap`: 単に`Mono#map`を使ってしまうと`Mono<Mono<T>>`になってしまうような場合に使う
- `Mono#flatMapMany`: 値が複数になる（`Flux`に変換したい）ような場合に使う
- `Mono.just`: 値から`Mono`を作る

ここでは`Mono`を例に紹介しましたが、`Flux`に関しても大体同様の操作が有ります。

### `Reactor Context`
サーバーサイド開発をする際には、層を跨いでリクエスト単位の値を参照したい場合が有ります。

従来のフレームワークでは、このような用途に`ThreadLocal`が使われていました。
例えば、`Spring Web MVC`における`Transactional`アノテーションの制御なども、内部的には`ThreadLocal`が利用されています。

一方、ノンブロッキングフレームワークでは複数リクエスト間で限られたスレッドを使い回すため、`ThreadLocal`を利用することができません。
そこで、`Project Reactor`を利用したフレームワークでは、`ThreadLocal`の代わりに`Context`という機能が利用されます。
実際、`Spring WebFlux`における`Transactional`アノテーションの制御は、`Context`を利用して実装されています。

https://spring.io/blog/2023/03/28/context-propagation-with-project-reactor-1-the-basics

なお、後述する`Coroutines`にも同じような機能かつ同名の`Context`が登場するため、この記事では`Reactor`の`Context`を`Reactor Context`と呼んで区別します。
また、`Reactor Context`と`Coroutine Context`には完全な互換性が有り、後述するような相互変換を行った場合も値が引き継がれます。

### 注意点
`Reactive Streams`に対応しているライブラリであっても、`Project Reactor`の機能は使えない（例えばそのライブラリを介すると`Reactor Context`が消えてしまう）場合が有ります。
自分の知っている例だと、`jOOQ`にはこの問題が有ります。

https://qiita.com/wrongwrong/items/c0617954c9a9585a1dee

## `Coroutines`
`Coroutines`は非同期・ノンブロッキングなプログラムを簡単に書ける`Kotlin`の言語機能です。

https://kotlinlang.org/docs/coroutines-overview.html

前述した2つの技術とは別物ですが、`Kotlin`と周辺ライブラリによって完全な相互変換が可能です。

### 頻出のクラス
`Coroutines`を利用する上で固有の頻出クラスは`Flow`です[^not-jdk]。
これは前述した`Flux`と同様、複数の値を返す時に用いる`List`や`Sequence`のようなものです。

https://kotlinlang.org/docs/flow.html

[^not-jdk]: これは前述の`java.util.concurrent.Flow`とは異なるクラスです

`Mono`に対応するクラスは無く、後述する`suspend`関数がその役割を担います。

### 頻出の記法
基本的に以下3点を覚えるだけで、アプリケーション上の大半の処理は通常と同様に書けます。

- 単一の値を返す場合、関数に`suspend`と付ける
- 複数の値を返す場合、戻り値を`Flow`にし、`suspend`は付けない
  - `Flow`を返す非`suspend`関数から`suspend`関数を呼びたい場合、[`flow`関数](https://kotlinlang.org/docs/flow.html#flow-builders)を使う

# 相互運用について
次に、`Coroutines`と`Reactor`(`Reactive Streams`)の相互運用についてざっくりまとめます。
前提として、`org.jetbrains.kotlinx:kotlinx-coroutines-reactor`が導入されているものとします。

https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-reactor

## `suspend`関数の呼び出しを`Mono`にする
`suspend`関数の呼び出し結果を`Mono`として返す場合、`mono`関数が利用できます。

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-reactor/kotlinx.coroutines.reactor/mono.html

これを利用した場合、`suspend`関数の呼び出しは以下のように書けます。

```kotlin
import kotlinx.coroutines.reactor.mono
import reactor.core.publisher.Mono

suspend fun foo(): Foo = TODO()
fun bar(): Mono<Foo> = mono { foo() } // ブロック内ではsuspend関数が呼び出せる
```

`Flux`に関しても同様に`flux`関数が用意されていますが、後述する`Flow`からの変換を利用する機会の方が多いため、こちらを利用する機会は少なかったです。

## `Flux`・`Flow`の相互変換
`Flux`と`Flow`に関しては、相互に変換することが出来ます。

```kotlin
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactor.asFlux
import reactor.core.publisher.Flux

val flux: Flux<Int> = flowOf(1, 2, 3).asFlux()
val flow: Flow<Int> = Flux.just(1, 2, 3).asFlow()
```

`Flow`に変換した後であれば、`Kotlin`の`Collection`同様に表現力豊かな拡張関数を利用することが出来ます。

## `suspend`関数で`Publisher`から値を取り出す
`suspend`関数では、ブロック無しに`Publisher`から値を取り出すことが出来ます。

### `Mono`の場合
`Mono`から値を取り出す場合、`awaitSingle`または`awaitSingleOrNull`関数を使います。

```kotlin
import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.awaitSingleOrNull
import reactor.core.publisher.Mono

suspend fun foo(mono: Mono<Any>) {
    val v1 = mono.awaitSingle()
    val v2 = mono.awaitSingleOrNull()
}
```

### `Flux`(`Publisher`)の場合
`Flux`の場合、`asFlow().toList()`といった操作で通常の`Collection`が得られます。
`Publisher`も`Flux`と同様の操作で値を取り出すことが出来ます。

他にも、`await`系の関数を使って1つずつ値を取り出すこともできます。

## その他
その他`Java`で実装された非同期・ノンブロッキング実装に関しても、これまで説明したような`Coroutine`との相互運用関数が提供されています。
`suspend`関数呼び出しは対象名に由来するトップレベル関数、値の取り出しは`await~`という名前の拡張関数になっている場合が多いです。

自分の経験では、例えば`AWS SDK`の`S3AsyncClient`は`Java`の`CompletableFuture`を返すため、これを`await`するために`kotlinx-coroutines-jdk8`を導入したことがありました[^aws]。

[^aws]: 今新規で開発する場合[aws-sdk-kotlin](https://github.com/awslabs/aws-sdk-kotlin)を使うべきかもしれません（自分はまだ手を出せておらず必要性を理解し切れていませんが……）

利用可能なモジュールに関しては以下をご覧下さい。

https://kotlinlang.org/api/kotlinx.coroutines/

また、`reactor-kotlin-extensions`のように、`kotlinx.coroutines`以外からもサポートライブラリが提供されている場合が有ります。

https://github.com/reactor/reactor-kotlin-extensions