선언적인 처리

• 선언적으로 접근하면, Flow는 Flow의 수집이 완료되었을 때 실행되는 onCompletion 중간 연산자가 있음.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun simple(): Flow<Int> = flow {
    emit(1)
    throw RuntimeException()
}

fun main() = runBlocking {
    simple()
        .onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
        .catch { cause -> println("Caught exception $cause") }
        .collect { value -> println(value) }
}

실행결과
1
Flow completed exceptionally
Caught exception java.lang.RuntimeException

• onCompletion 연산자는 catch 와 다르게 예외를 처리하지 않음.
• 예외는 여전히 다운스트림으로 흐름.
  • onCompletion 연산자로 전달되며, 이후에 catch 연산자를 사용해 처리될 수 있음.

성공적인 완료

• catch 연산자와 또 다른 점은 onCompletion는 모든 예외를 볼 수 있고, 업스트림 Flow가 취소나 실패 없이 성공적으로 완료되었을 때 null을 수신함.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun simple(): Flow<Int> = (1..3).asFlow()

fun main() = runBlocking {
    simple()
        .onCompletion { cause -> println("Flow completed with $cause") }
        .collect { value -> println(value) }
}

실행결과
1
2
3
Flow completed with null

• 위 예제에선 취소나 실패 없이 성공적으로 완료되었으므로, null을 수신함.

fun main() = runBlocking {
    simple()
        .onCompletion { cause -> println("Flow completed with $cause") }
        .collect { value ->
            check(value <= 1) { "Collected $value" }
            println(value)
        }
}

실행결과
1
Flow completed with java.lang.IllegalStateException: Collected 2
Exception in thread "main" java.lang.IllegalStateException: Collected 2
...

• 다운 스트림 예외로 인해 Flow가 중단되었으므로, onCompletion 에 들어온 예외는 null이 아님.

명령적으로 다루기 vs 선언적으로 다루기

• 명령적 방식과 선언적 방식은 두 접근 방식 모두 유효함.
• 이는 선호도와 코드 스타일에 따라 선택되어야 함.

Flow 실행하기

• 비동기 이벤트를 표현하기 위해 Flow를 사용하기 쉬움.
  • 들어오는 이벤트에 대한 반응을 코드로 등록하고 이후의 작업을 계속해서 수행하도록 하는 addEventListener 함수와 비슷한 역할을 하는 것이 필요함.
  • onEach 연산자가 그런 역할을 해줌.
    • 이는 중간 연산자임.
    • 터미널(종단) 연산자가 있어야함. 그렇지 않으면 onEach만 호출하는 것으로는 효과가 없음.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

fun main() = runBlocking {
    events()
        .onEach { event -> println("Event: $event") }
        .collect()
    println("Done")
}

실행결과
Event: 1
Event: 2
Event: 3
Done

• luanchIn 터미널(종단)연산자가 여기서 편리하게 사용될 수 있음.
• collect를 luanchIn 으로 대체함으로써 Flow의 수집을 별도의 코루틴에서 실행할 수 있음.
  • 이후 코드들은 즉시 계속해서 실행될 수 있음.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

fun main() = runBlocking {
    events()
        .onEach { event -> println("Event: $event") }
        .launchIn(this)
    println(this)
}

실행결과
"coroutine#1":BlockingCoroutine{Active}@394e1a0f
Event: 1
Event: 2
Event: 3