Spring WebfluxとKotlinCoroutine

こちら、Kotlinアドベントカレンダー 18日目の記事です。
こんばんはこんにちは、今年は様々な事がありました。東京で暮らし始めたSatohJohnです。
今回は社内開発で利用したKotlinでのSpring WebFlux（主にrouter）の書き方についてです。
Spring WebFlux自体については、ググれば色々調べれば出てくると思います！！素晴らしい！！

環境

	Gradle(pluginId or artifactId)	バージョン
Spring Boot	org.springframework.boot	2.2.2.RELEASE
kotlin	org.jetbrains.kotlin.jvm org.jetbrains.kotlin.plugin.spring	1.3.61
kotlinx	kotlinx-coroutines-core kotlinx-coroutines-reactor	1.3.3

書き方

JavaでのSpring WebfluxのRouter

routerに対して Mono<ServerResponse> を返却する形になります。

@Bean
public RouterFunction<ServerResponse> routes() {
    return RouterFunctions
            .route(GET("/hello"), req ->
                    ServerResponse.ok().bodyValue(String.format("hello world. %s", req.queryParam("name").orElse("anonymous"))));
}

悪くないですね！！ラムダ式周りで結構書きやすいです

KotlinでのSpring WebfluxのRouter

@Bean
fun routes() = router {
    GET("/hello") {
        ServerResponse.ok().bodyValue("hello world, ${it.queryParam("name").orElse("anonymous")}")
    }
}

Kotlinの場合はdslで書けるので、少しシンプルに見えますね！

KotlinでのSpring WebfluxのRouter(Coroutineを使うパターン)

CoRouterFunctionDsl.GET が suspend の関数で ServerResponse を返却することを求めていますので、 ServerResponse.bodyValueAndAwait を利用します。

@Bean
fun routes() = coRouter {
    GET("/hello") {
        ServerResponse.ok().bodyValueAndAwait("hello world, ${it.queryParam("name").orElse("anonymous")}")
    }
}

なんとなくKotlinのDSLが読みやすいのはわかりますね。

複数の要素を返す場合

上記は要素一つを返す例でした。次は要素を複数返す場合についてです。

Javaでの場合

Fluxっぽく無限stream要素をから、1秒ごとに一つづつ返すものとします
JavaではMonoではなくFluxを返すことになります（bodyValueがMonoを返していた)

@Bean
public RouterFunction<ServerResponse> routes() {
    return RouterFunctions
            .route(GET("/hello"), req -> 
                    ServerResponse.ok()
                            .contentType(MediaType.APPLICATION_STREAM_JSON)
                            .body(
                                    Flux.fromStream(Stream.iterate(0, i -> i + 1))
                                            .delayElements(Duration.ofSeconds(1)),
                                    Integer.class
                            )
            );
}

Kotlinでの場合

KotlinではFlowを返します。
基本はListからFlowにします(無限ストリームを利用するのでsequenceから作成しています

@Bean
fun routes() = coRouter {
    GET("/hello") {
        ServerResponse.ok()
            .contentType(MediaType.APPLICATION_STREAM_JSON)
            .bodyAndAwait(flow {
                generateSequence(0) { it + 1 }.asFlow().collect {
                    delay(1000)
                    emit(it)
                }
            })
    }
}

リクエストを投げるとJavaのときと同じようにが返却されているのがわかります。

curl -v localhost:8080/hello -H 'Accept: application/stream+json'

*   Trying ::1...
* TCP_NODELAY set
* Connected to localhost (::1) port 8080 (#0)
> GET /hellok HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.54.0
> Accept: application/stream+json
> 
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: application/stream+json
< Cache-Control: no-cache, no-store, max-age=0, must-revalidate
< Pragma: no-cache
< Expires: 0
< X-Content-Type-Options: nosniff
< X-Frame-Options: DENY
< X-XSS-Protection: 1 ; mode=block
< Referrer-Policy: no-referrer
< 
0
1
2
3
4
5
6
7

考察

はてさて、Kotlinを使うということDSLだと短くきれいに書けますね。っていうのは良いと思います
ではcoroutineを使う必要はあるんでしょうか？使いたいんだから使うんだよという気持ち

個人的には、MonoやFluxの(reactor-core)に依存せず、Kotlinの言語にあるcoroutine(suspendやFlow)に依存するところの違いかなぁと思っています。
bodyValueAndAwaitやbodyAndAwaitは内部で reactive-stream には依存していますが、reactor-core というライブラリの実装に依存していない感じです。

詳しくは以下の通り書いてくださっているので、こちらも合わせて参照してもらえると良いのかなぁと思います
https://github.com/pljp/kotlinx.coroutines/blob/japanese_translation/reactive/coroutines-guide-reactive.md

まとめ

Kotlin書きやすいですね、読みやすいですし。
参考資料などはあまりなく、開発時は色々辛いこと（例外がどこで起きたのか全然追えないなど）もありましたが
現在はうまく使えていますし、また今回WebFluxと合わせることで、Coroutineのことが若干わかった気がしました。