Spring reactor Switch If Empty를 사용할 때 주의할 점

상황

 

작업을 하면서 아래와 같이, Flux를 가져오고, 만약 없다면 switchIfEmpty를 사용하여 default 값을 돌려주는 코드를 작성했습니다.

(아래 코드는 예시입니다.)

 

fun main() {
    val result = getFlux()
        .switchIfEmpty(mapToPlusOne())
        .collectList()

    println(result.block()!!)
}

private fun getDefaultFlux(): Flux<Int> {
    return Flux.fromIterable(listOf(1, 2, 3, 4))
}

 

 

 

switchIfEmpty의 경우 실제 Reactor 코드 주석 설명에서 empty 시퀀스를 돌려줄 경우에 다른 Publisher로 돌려주는 코드기 때문에, 시퀀스가 empty가 아닐 경우 동작하지 않아야 할것이라고 예상했습니다.

 

 

그런데, getFlux() 함수에서 Flux가 empty가 아닐 때도 계속 getDefaultFlux() 함수가 호출되는 상황이 발생하였습니다.

그래서, 실제로 getFlux() 함수가 empty Flux를 돌려주는지 확인하기 위해서 아래와 같이 Flux를 찍어주고, getDefaultFlux() 함수 내부에 print 문을 작성해서 Flux가 있는지를 확인하였습니다.

 

fun main() {
    val result = Flux.fromIterable(listOf(1, 2, 3, 4))
        .switchIfEmpty(mapToPlusOne())
        .collectList()

    println(result.block()!!)
}

private fun getDefaultFlux(): Flux<Int> {
    println(1)

    return Flux.fromIterable(listOf(1, 2, 3, 4))
}

 

 

확인 결과 아래처럼 실제로 empty가 아닐 때에도 1이 출력되는것을 확인했습니다.

 

 

원인 및 해결책

 

switchIfEmpty에 넣어준 Publisher가 바로 호출되었기 때문입니다.

 

Mono.just, Flux.just, Flux.fromIterable, 기타 다른 코드의 경우 즉시 호출되기 때문에, subscribe를 하기 이전에도 즉시 수행이 됩니다.

 

이를 방지하기 위해서는 Mono.defer, Flux.defer 등을 통해서 이를 subscribe를 할 때 실행되도록, lazy하게 만들어야 합니다.

 

실제로 아래 코드처럼 Mono.defer 변경하게 되면 아래와 같은 문제는 사라집니다.

 

(Mono.defer에 대한 자세한 설명은 https://stackoverflow.com/questions/55955567/what-does-mono-defer-do 를 참고하시면 됩니다.)

 

변경된 코드

fun main() {
    val result = Flux.fromIterable(listOf(1, 2, 3, 4))
        .switchIfEmpty(Flux.defer { getDefaultFlux() })
        .collectList()

    println(result.block()!!)
}

private fun getDefaultFlux(): Flux<Int> {
    println(1)

    return Flux.fromIterable(listOf(1, 2, 3, 4))
}

 

결과

 

 

그런데 주의할 점은, 위의 코드처럼 모든 케이스를 defer로 감싸줘야 합니다. 일부 코드만 감싸고, 일부 코드는 감싸지 않으면 감싸지 않은 코드는 동작할 수 있습니다.

 

fun main() {
    val result = Flux.fromIterable(listOf(1, 2, 3, 4))
        .switchIfEmpty(getDefaultFlux())
        .collectList()

    println(result.block()!!)
}

private fun getDefaultFlux(): Flux<Int> {
    println(1)

    return Flux.defer { Flux.fromIterable(listOf(1, 2, 3, 4)) }
}

 

위 처럼 일부 코드를 감싸지 않으면 아래와 같이 바로 호출이 됩니다.

 

 

관련 kotlin extension

 

이런 케이스가 많기 때문에 Mono, Flux에 관련 코틀린 extension을 제공해주고 있습니다.

 

파라미터로 교체할 publisher를 람다로 보내주게 되면, 그걸 defer로 감싸주고 있습니다.

 

 

보통은 아래와 같이 코틀린에서 작업을 하게 되면 람다로 호출하게 되고, 람다로 호출하면 자연스럽게 코틀린 extension을 사용하게 되기 때문에 별 문제가 없습니다.

 

 

하지만, 람다로 switchIfEmpty를 호출하지 않거나 저 코틀린 extension을 이용하지 않고, 아래처럼 Flux나 Mono의 switchIfEmpty를 직접 호출할 경우 조심해야 합니다.

 

 

결론

 

1. reactor에서 switchIfEmpty를 사용할 때는, 코드가 subscribe할 때 실행되는지 여부를 확인해야 합니다.

2. 코틀린에서는 관련 extension을 제공하고 있어서 편하게 사용할 수 있고, 대부분의 경우 문제가 발생하지 않습니다.

3. 즉시 실행되는 코드일 경우 Mono.defer, Flux.defer 같이 subscribe할 때 코드를 실행시켜주는 함수들을 사용하면 됩니다.

 

출처

 

https://stackoverflow.com/questions/55955567/what-does-mono-defer-do - what does Mono.defer do

https://github.com/reactor/reactor-core/issues/1164