[spring webflux] coroutine reactor

[backtony]

coroutine reactor

coroutine reactor 라이브러리를 사용하면 spring webflux와 함께 코루틴을 사용할 수 있다.

리액티브 코루틴 변환 방식

//Mono → suspend 
fun handler(): Mono<Void> -> suspend fun handler()

//Flux → Flow
fun handler(): Flux<T> -> fun handler(): Flow<T>

Mono는 suspend, Flux는 Flow로 변환된다.

// 리액티브 방식 
interface ContentReactiveRepository : ReactiveCrudRepository<Content, Long> {

    fun findByUserId(userId: Long) : Mono<Content>

    fun findAllByUserId(userId: Long): Flux<Content>
}

// 코루틴 방식
interface ContentCouroutineRepository : CoroutineCrudRepository<Content, Long> {

        suspend fun findByUserId(userId:Long) : Content?

    fun findAllByUserId(userId: Long): Flow<Content>
}

r2dbc는 coroutineCrudRepository가 지원된다. ReactiveCrudRepository를 써야하는 상황이라면 Mono, Flux에 대한 확장함수로 awaitXXX 함수를 지원한다.

mono, flux 다루기

coroutine reactor : https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-reactor/kotlinx.coroutines.reactor/

• awaitSingle
  • 주어진 Mono에서 스레드를 차단하지 않고 결과값을 기다리며, 결과값이 있다면 반환하거나 에러가 발생했다면 예외를 throw한다.
  • 이 일시중단 함수는 취소가 가능하다. 현재 코루틴의 job이 취소되거나 완료될 때, 이 일시중단 함수가 대기중인 동안, 함수는 즉시 subscribetion을 취소하고 CancellationException과 함께 재개된다.
• awaitSingleOrNull
  • awaitSingle과 같으나 Mono가 값이 없으면 null을 반환한다.
• awaitFirst
  • 메서드는 여러 개의 Mono 또는 Flux 중 첫 번째로 완료되는 작업을 대기하고 해당 결과를 반환합니다. 이 메서드를 호출하면 현재 스레드는 여러 개의 리액티브 시퀀스 중 가장 빨리 완료되는 작업이 완료될 때까지 차단됩니다. 가장 빨리 완료되는 작업이 Mono인 경우 결과를 반환하고, Flux인 경우 첫 번째 결과를 반환합니다.
• awaitFirstOrNull
  • Mono가 비어있거나 여러 개의 요소를 포함하고 있는 경우 첫 번째 요소만 반환하고 나머지는 무시한다.
  • 비어있는 경우 null을 반환한다.
  • 예외 발생 시 throw 한다.
• awaitFirstOrDefault
  • awaitFirstOrNull에서 null 대신 default value를 반환한다.
• awaitFirstOrElse
• awaitLast
  • 마지막값을 반환
• awaitBody
  • webflux의 request를 mono로 바꿀 때 사용되는 bodyToMono와 같다.
• awaitBodyOrNull
  • 없다면 null 반환
• bodyToFlow
  • webflux의 bodyToFlux와 같다.
• awaitExchange, awaitExchangeOrNull
  • webclient에서 retrieve를 사용해서 응답 데이터를 정의하는데 exchangeToMono나 exchangeToFlux를 통해 한번에 정의하고 요구조건에 맞게 처리할 수 있다. 코루틴에서는 이걸 awaitXX가 대체한다.
• exchangeToFlow
  • exchangeToFlux를 대체한다.