Deep dive kotlin synergy (XXI): flow life cycle function

Series eBook ： Portal

---

Flow It can be imagined as a pipe , The requested value flows in one direction , The corresponding value flows in the other direction . When flow Completion or exception , These messages will also be transmitted , And close the intermediate steps on the way . therefore , When these values begin to flow , We can monitor the value 、 Abnormal or other characteristic events （ Such as start or finish ）. So , We used onEach、 onStart、onCompletion、onEmpty and catch Other methods . Let me explain these life cycle methods one by one .

onEach

In response to the value of each flow , We use onEach function .

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

onEach Of lambda Expressions are suspended , The elements are in order （ The order ） Processed . therefore , If we were onEach add to delay function , We will delay the flow of each value .

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

onStart

onStart Function to set a listener , once flow start-up , The listener will be called back . It should be noted that , onStart Don't wait for the response of the first element , But when we request the first element , It will be called .

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

stay onStart（ And in onCompletion、onEmpty、catch） Elements can be emitted in , These elements will flow down there .

suspend fun main() {
    
    flowOf(1, 2)
        .onEach {
     delay(1000) }
        .onStart {
     emit(0) }
        .collect {
     println(it) }
}
// 0
// (1 sec)
// 1
// (1 sec)
// 2

onCompletion

There are several ways to complete a flow. The most common is in flow When the builder is finished （ For example, the last element is sent ）, Although it may occur when an exception is not caught or the collaboration is canceled . In all these cases , We can all use onCompletion Method is flow Add a listener .

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

suspend fun main() = coroutineScope {
    
    val job = launch {
    
        flowOf(1, 2)
            .onEach {
     delay(1000) }
            .onCompletion {
     println("Completed") }
            .collect {
     println(it) }
    }
    delay(1100)
    job.cancel()
}
// (1 sec)
// 1
// (0.1 sec)
// Completed

stay Android in , We use it a lot onStart To show the progress bar （ Indicator waiting for network response ）, Then we use onCompletion To hide it .

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

onEmpty

flow It may be done without emitting any values , There may be an unexpected situation . In this case , There is one onEmpty function , It's in flow When it is completed but no element is emitted, it will call back . We can use onEmpty To emit some default values .

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

catch

stay flow Any time the builder or processes values , Can be abnormal . Such exceptions will flow downward , Close each processing step on the way ; However , It can be captured and managed . So , We can use catch Method . This listener receives exceptions as parameters , And allows you to perform recovery operations .

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

In the example above ,onEach No response to exception . The same thing happens to other functions , Such as map、 filter etc. . Only onCompletion Will be called .

catch Method to prevent the propagation of exceptions by capturing . Although the previous steps have been completed , however catch New values can still be emitted , And keep flow The rest of is active .

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

suspend fun main(): Unit {
    
    flow.catch {
     emit("Error") }
        .collect {
     println("Collected $it") }
}
// Collected Message1
// Collected Error

catch It will only respond to exceptions thrown in upstream defined functions （ As you can imagine , When there is something abnormal about obscenity , Still need to catch exceptions ）.

stay Android in , We use it a lot catch To show flow What happened in ：

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

We can also use catch To send default data for display on the screen , For example, an empty list .

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

Uncaught exception

flow If an uncaught exception occurs in the, the... Will be canceled immediately flow, also collect This exception will be thrown again . This behavior is typical of suspending functions , coroutineScope It's the same behavior , A typical response is to use try-catch Block in flow External catch exception of ：

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

Please note that , Use catch It does not prevent exceptions in terminal operation （ because catch Cannot use after the last operation ）. therefore , If collect There is an exception in , It will not capture , Instead, an error will be thrown .

val flow = flow {
    
    emit("Message1")
    emit("Message2")
}

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

therefore , The usual approach is to transform the logic layer from collect Move to onEach, And put it in catch Before . When we doubt collect When the logic of may be abnormal , This can be particularly useful . If we change the logical operation from collect Move away , You can be sure catch Can catch all exceptions .

val flow = flow {
    
    emit("Message1")
    emit("Message2")
}
suspend fun main(): Unit {
    
    flow.onStart {
     println("Before") }
        .onEach {
     throw MyError() }
        .catch {
     println("Caught $it") }
        .collect()
}
// Before
// Caught MyError: My error

flowOn

Pass to flow operation （ Such as onEach、onStart、onCompletion etc. ） Of lambda Expressions and their builders （ Such as flow {..} or channelFlow{..}） It's all suspended . The pending function needs to have a context , And should be associated with their parent coroutine （ Structured concurrency ）. therefore , You may want to know where the context of these functions comes from . The answer is ： From the collect From the context of .

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

How does this code work ？ The terminal operation will call the elements from the upstream , Thus providing the context of the collaboration . However , It can also pass through flowOn Function to modify （ Modify the context of the collaboration ）：

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

please remember , flowOn Only applicable to flow Functions upstream in .

launchIn

collect Is a suspend function , It will suspend a process until flow complete . We usually use them launch The builder wraps it , In order to flow Processing can be started on another collaboration . To optimize this situation , There is one launchIn function , It will call on the scope passed as a parameter collect.

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

launchIn It is usually used to start a flow.

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

summary

In this chapter , We learned different flow function . Now we know how to flow At the beginning of the 、 Perform some operations at the end or on each element ; We also know how to catch exceptions , And how to start in the new process flow. These are basic tools that are widely used , Especially in Android In development . for example , Here's a paragraph Android Use in flow Code for ：

fun updateNews() {
    
    newsFlow()
        .onStart {
     showProgressBar() }
        .onCompletion {
     hideProgressBar() }
        .onEach {
     view.showNews(it) }
        .catch {
     view.handleError(it) }
        .launchIn(viewModelScope)
}