【Kotlin Coroutines】撃ち放し（fire and forget）処理にReactive Contextの値を引き継ぐ【Spring WebFlux】

筆者はこの辺りの話に言うほど詳しくないため、より良いやり方が有りましたらご教授頂けると助かります。

TL;DR

• Context.asCoroutineContext()で作成したCoroutineContextを渡すのが簡単に見える

前提

Reactive Contextについて

Mono/Fluxには、上流のコンテキストに書き込んだ値を下流のスコープから参照する機能が有ります。
この機能はドキュメント上単にContextと呼称されていますが、分かりやすさのためこの記事ではReactive Contextと呼称します。

Spring WebFluxでは、例えばWebFilter等を用いてリクエスト等から読み出した値をReactive Contextへ書き込んでおき、下流の処理でそれを読み出して利用するようなことができます。
認証やトランザクション制御に関しても、この機能を利用して作成されています。

値をセットするWebFilterの例

import org.springframework.stereotype.Component
import org.springframework.web.server.ServerWebExchange
import org.springframework.web.server.WebFilter
import org.springframework.web.server.WebFilterChain
import reactor.core.publisher.Mono

@Component
class SetFooIdFilter : WebFilter {
    override fun filter(exchange: ServerWebExchange, chain: WebFilterChain): Mono<Void> {
        return chain.filter(exchange).contextWrite { it.put(FOO_ID_KEY, /* ヘッダ等からのFooId読み出し処理 */) }
    }
}

値を読み出す例

import reactor.core.publisher.Mono

// この関数そのものはトップレベルに定義可能
fun getFooId(): Mono<String> = Mono.deferContextual {
    it.get<String>(FOO_ID_KEY)
}

Coroutineでの撃ち放し処理について

Coroutineでは、例えば以下のように書くことで撃ち放し（処理の発火後結果を待たずに終了する）処理を書くことができます。

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch

CoroutineScope(Dispatchers.Default).launch {
    /* 撃ち放しにしたい処理 */
}

やりたいこと

先ほど紹介したCoroutineでの撃ち放し処理の書き方では、呼び出し元のReactive Contextが引き継がれないため、撃ち放しにしたい処理の中で共通利用したい値の読み出しができません。
この問題を解決します。

やり方

CoroutineにはReactorContextという機能が用意されており、それを利用することで実現できます。

サンプルコードは以下の通りです。
asCoroutineContextの戻り値がReactorContextです。

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.reactor.asCoroutineContext
import kotlinx.coroutines.reactor.awaitSingle
import reactor.core.publisher.Mono
import reactor.util.context.Context
import kotlin.coroutines.CoroutineContext

// 現状のReactive Contextから必要な値を引き継いだCoroutineContextを作成する
suspend fun getTakenOverContext(): CoroutineContext {
    val fooId: Mono<String> = getFooId()
    return Context.of(FOO_ID_KEY, fooId.awaitSingle()).asCoroutineContext()
}

CoroutineScope(Dispatchers.Default).launch(context = getTakenOverContext()) {
    /* 撃ち放しにしたい処理 */
}

補足

上で紹介した例で、「FOO_ID_KEYに対してMono<String>をバインドする形にすれば便利なのでは？」と考えて幾つか試しましたが、自分はこれを実現できませんでした。

この形にできると以下のような利点が有ります。

• getTakenOverContextを非suspend関数にできる（= 非suspend関数からも撃ち放し処理を起動できる）
• getFooId()の戻り値を直接登録できる

起きた問題
Monoは遅延評価されるため、getFooId()で取得したMonoから値を読み出そうとすると、deferContextualへアクセスします。
これによって無限ループになり、StackOverflowが発生しました。

何かやり方が有るような気はしますが、自分のユースケースでは、撃ち放し処理の起動は必ずsuspend関数から行えたため、調査を打ち切っています。