[코틀린 코루틴] 플로우 생명주기 함수

Flow에서는 완료, 예외, 시작 같은 이벤트도 데이터 스트림의 일부로 간주되어 감지하고 처리할 수 있다.

플로우 생명주기 함수를 사용하여 이러한 모든 상태를 Flow 안에서 관리할 수 있다.

 

• onEach: Flow에서 방출된 각 데이터를 처리하는 함수
• onStart: Flow가 시작되기 전에 실행할 작업 정의
• onCompletion: Flow가 완료되거나 취소되었을 때 실행
• onEmpty: Flow가 데이터를 방출하지 않을 경우 실행
• catch: Flow에서 발생한 예외를 처리
• flowOn: Flow가 실행되는 코루틴 디스패처를 변경

 

onEach

 

onEach 람다식은 중단 함수이며, 원소는 순서대로 처리된다.

다음 예제처럼 onEach에 delay를 넣으면 각각의 값이 흐를 때마다 지연되게 된다.

suspend fun main() {
    flowOf(1, 2)
        .onEach { delay(1000) }
        .collect { println(it) }
}
// (1초 후)
// 1
// (1초 후)
// 2

 

onStart

 

onStart 함수는 플로우가 시작되는 경우에 호출되는 리스너를 설정하며, 첫 번째 원소가 생성되기 전에 실행된다.

suspend fun main() {
    flowOf(1, 2)
        .onEach { delay(1000) }
        .onStart { println("Before") }
        .collect { println(it) }
}
// Before
// (1초 후)
// 1
// (1초 후)
// 2

 

onCompletion

 

onCompletion 함수를 사용해 플로우가 완료되거나 예외 혹은 취소가 발생했을 때 호출되는 리스너를 추가할 수 있다.

fun updateNews() {
    scope.launch {
        newsFlow()
            .onStart { showProgressBar() }
            .onCompletion { hideProgressBar() }
            .collect { view.showNews(it) }
    }
}

 

onEmpty

 

플로우는 예기치 않은 이벤트가 발생하면 값을 내보내기 전에 완료될 수 있다.

onEmpty 함수는 원소를 내보내기 전에 플로우가 완료되면 실행되며, 기본 값을 내보내기 위한 목적으로 사용될 수 있다.

suspend fun main() = coroutineScope {
    flow<List<Int>> { delay(1000) }
        .onEmpty { emit(emptyList()) }
        .collect { println(it) }
}
// (1초 후)
// []

 

catch

 

플로우를 만들거나 처리하는 도중에 예외가 발생할 수 있다.

이러한 예외는 아래로 전파되고, 처리되지 않은 예외는 Flow를 중단시키고 소비자 쪽으로 전달된다.

catch 함수는 이러한 예외를 감지하고 처리할 수 있으며, 예외를 인자로 받고 예외 처리를 위한 연산을 수행할 수 있다.

class MyError : Throwable("My error")

val flow = flow {
    emit(1)
    emit(2)
    throw MyError()
}

suspend fun main(): Unit {
    flow.onEach { println("Got $it") }
        .catch { println("Caught $it") }
        .collect { println("Collected $it") }
}
//Got 1
//Collected 1
//Got 2
//Collected 2
//Caught MyError: My error

catch 함수는 예외를 처리하면서 새로운 값을 방출할 수 있어, Flow가 중단되지 않고 남은 흐름을 계속 이어갈 수 있다.

fun updateNews() {
    scope.launch {
        newsFlow()
            .catch {
                view.handleError(it)
                emit(emptyList()) // 빈 값 방출
            }
            .onStart { showProgressBar() }
            .onCompletion { hideProgressBar() }
            .collect { view.showNews(it) }
    }
}

 

 

잡히지 않은 예외

 

플로우에서 잡히지 않은 예외는 플로우를 즉시 취소시키며, collect는 예외를 다시 던진다.

플로우 바깥에서 try-catch 블록을 사용해서 예외를 잡을 수 있다.

val flow = flow {
    emit("Message1")
    throw MyError()
}

suspend fun main(): Unit {
    try {
        flow.collect { println("Collected $it") }
    } catch (e: MyError) {
        println("Caught")
    }
}
//Collected Message1
//Caught

catch 함수는 윗부분에서 던진 예외에만 반응하며, collect에서 예외가 발생하면 예외를 잡지 못하게 되어 블록 밖으로 예외가 전달된다.

suspend fun main(): Unit {
    flow.onStart { println("Before") }
        .catch { println("Caught $it") } 
        .collect { throw MyError() } // 예외가 전파된다.
}
//Before
//Exception in thread "main" MyError: My error

예외 처리를 위해 자주 사용하는 방법은 collect의 로직을 onEach로 옮기고 이를 catch 이전에 배치하는 것이다.

suspend fun main(): Unit {
    flow.onStart { println("Before") }
        .onEach { throw MyError() } // catch 함수에 의해 예외처리된다.
        .catch { println("Caught $it") }
        .collect()
}
//Before
//Caught MyError: My error

 

flowOn

 

Flow 연산과 Flow 빌더에 사용되는 람다식은 모두 중단 함수이며, 중단 함수는 코루틴 컨텍스트를 필요로 한다. 

Flow의 함수들은 collect가 호출된 위치의 컨텍스트를 상속받아 실행된다.

fun usersFlow(): Flow<String> = flow {
    repeat(2) {
        val ctx = currentCoroutineContext()
        val name = ctx[CoroutineName]?.name
        emit("User$it in $name")
    }
}

suspend fun main() {
    val users = usersFlow()
    withContext(CoroutineName("Name1")) {
        users.collect { println(it) }
    }
    withContext(CoroutineName("Name2")) {
        users.collect { println(it) }
    }
}
//User0 in Name1
//User1 in Name1
//User0 in Name2
//User1 in Name2

최종 연산이 호출되면 상위에 있는 모든 원소를 요청하며, 이를 처리할 코루틴 컨텍스트를 함께 전달한다.

flowOn 함수는 이 과정에서 컨텍스트를 변경할 수 있도록 하며, 플로우에서 윗 부분에 있는 함수에서만 작동한다.

suspend fun present(place: String, message: String) {
    val ctx = coroutineContext
    val name = ctx[CoroutineName]?.name
    println("[$name] $message on $place")
}

fun messagesFlow(): Flow<String> = flow {
    present("flow builder", "Message")
    emit("Message")
}

suspend fun main() {
    val users = messagesFlow()
    withContext(CoroutineName("Name1")) {
        users
            .flowOn(CoroutineName("Name3")) // 윗부분에 있는 함수에서만 작동 [Name3]
            .onEach { present("onEach", it) }
            .flowOn(CoroutineName("Name2")) // 윗부분에 있는 함수에서만 작동 [Name2]
            .collect { present("collect", it) } // 위에서 전달받은 코루틴 컨텍스트 [Name1]
    }
}
//[Name3] Message on flow builder
//[Name2] Message on onEach
//[Name1] Message on collect

 

 

launchIn

 

collect 함수는 플로우가 완료될 때까지 코루틴을 중단하는 중단 연산이다.

launch 빌더로 collect를 래핑하면 플로우를 다른 코루틴에서 처리할 수 있다.

플로우의 확장 함수인 launchIn 함수를 사용하면 유일한 인자로 스코프를 받아 collect를 새로운 코루틴에서 시작할 수 있다.

 

collect 함수는 Flow가 데이터를 모두 방출할 때까지 기다리며, 이로 인해 코루틴이 멈추게 된다.

만약 collect를 다른 코루틴에서 실행하고 싶다면, launch를 사용하여 Flow를 별도의 코루틴에서 처리할 수 있다.
launchIn 함수를 사용하면 Flow를 자동으로 새로운 코루틴에서 실행할 수 있으며, 이 함수는 스코프만 인자로 받아 간단하게 새로운 코루틴을 생성한다.

fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = 
    scope.launch { collect() }

suspend fun main(): Unit = coroutineScope {
    flowOf("User1", "User2")
        .onStart { println("Users:") }
        .onEach { println(it) }
        .launchIn(this)
}
//Users:
//User1
//User2