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

Flow는 요청이 한 방향으로 흐르고 요청으로 생성된 값이 다른 방향으로 흐르는 파이프라 할 수 있다.

Flow가 끝나거나 예외 발생 시 이런 정보가 전달돼 중간 단계가 종료된다.

모든 정보가 Flow로 전달되므로 값, 예외, 시작 / 완료 등 다른 특정 이벤트를 감지할 수 있다. 아래 메서드들을 쓰면 된다.

 

onEach

 

Flow의 값을 하나씩 받기 위해 쓴다.

 

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.onEach

suspend fun main() {
    flowOf(1, 2, 3, 4)
        .onEach { print(it) }
        .collect()
}

// 1234

 

onEach 람다식은 중단 함수고 원소는 순서대로 처리된다. 따라서 onEach에 delay를 넣으면 각 값이 흐를 때마다 지연된다.

 

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.onEach

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

// (1초 후)
// 1
// (1초 후)
// 2

 

onStart

 

최종 연산이 호출될 때와 같이 Flow가 시작되는 경우에 호출되는 리스너를 설정한다.

onStart는 원소 생성을 기다렸다가 호출되는 게 아니란 것이 중요하다. 첫 원소를 요청했을 때 호출된다.

 

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart

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

// Before
// (1초 후)
// 1
// (1초 후)
// 2

 

onCompletion

 

Flow를 완료할 수 있는 여러 방법이 있다. 잡히지 않은 예외가 발생하거나 코루틴이 취소됐을 때도 포함이지만 가장 흔한 방법은 flow 빌더가 끝났을 때(마지막 원소가 전송됐을 때 등)다.

onCompletion을 사용해서 Flow 완료 시 호출되는 리스너를 추가할 수 있다.

 

import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach

suspend fun main() = coroutineScope {
    flowOf(1, 2)
        .onEach { delay(1000) }
        .onCompletion { println("Completed") }
        .collect { println(it) }
}

// (1초 후)
// 1
// (1초 후)
// 2
// Completed

 

안드로이드에선 네트워크 응답을 기다리는 척도인 프로그레스 바를 표시하기 위해 onStart를 쓰고, 가릴 땐 onCompletion을 쓴다.

 

onEmpty

 

Flow는 예기치 않은 이벤트가 발생하면 값을 내보내기 전에 완료될 수 있다. onEmpty는 원소를 내보내기 전에 Flow가 완료되면 실행되서 기본값을 내보내기 위해 사용할 수 있다.

 

import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onEmpty

suspend fun main() = coroutineScope {
    flow<List<Int>> { delay(1000) }
        .onEmpty { emit(emptyList()) }
        .collect { println(it) }
}

// (1초 후)
// []

 

catch

 

Flow를 만들거나 처리하는 중 예외가 발생할 수 있다. 예외는 아래로 흐르며 처리하는 단계를 하나씩 닫는다.

하지만 catch로 예외를 관리할 수도 있다. 리스너는 예외를 인자로 받고 정리하기 위한 연산을 수행할 수 있다.

 

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onEach

class MyError: Throwable("MyError")

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

suspend fun main() {
    flow.onEach { println("Got $it") }
        .catch { println("Caught $it") }
        .collect { println("Collected $it") }
}

// Got 1
// Collected 1
// Got 2
// Collected 2
// Caught com.example.kotlinpractice.MyError: MyError

 

catch는 예외를 잡아 전파되는 걸 멈춘다. 이전 단계는 이미 완료됐지만 catch는 여전히 새 값을 내보낼 수 있어서 남은 Flow를 지속할 수 있다.

안드로이드에선 Flow에서 일어나는 예외를 보여주거나 빈 리스트처럼 화면에 표시되는 기본값을 내보내기 위해 쓸 수 있다.

 

flowOn

 

onEach, onStart, onCompletion 같은 Flow 연산과 flow, channelFlow 같은 플로우 빌더의 인자로 쓰이는 람다식은 모두 중단 함수다. 중단 함수는 컨텍스트가 필요하고 구조화된 동시성을 위해 부모와 관계를 유지한다.

Flow는 collect가 호출되는 곳에서 컨텍스트를 얻어온다. 아래는 flowOn()으로 컨텍스트를 바꾸는 예시다.

 

import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.withContext
import kotlin.coroutines.coroutineContext

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"))
            .onEach { present("onEach", it) }
            .flowOn(CoroutineName("Name2"))
            .collect { present("collect", it) }
    }
}

// [Name3] Message on flow builder
// [Name2] Message on onEach
// [Name1] Message on collect

 

launchIn

 

collect는 Flow가 완료될 때까지 코루틴을 중단하는 중단 함수다. launch 빌더로 collect를 래핑하면 Flow를 다른 코루틴에서 처리할 수 있는데, launchIn을 쓰면 유일 인자로 scope를 받아서 collect를 새 코루틴에서 시작할 수 있다.

 

import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart

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

// Users:
// User1
// User2