Exceptions and Completion

Example

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.io.IOException

fun liveScores(): Flow<String> = flow {
    emit("IND 120/2")
    emit("IND 145/3")
    throw IOException("socket closed")
}

fun main() = runBlocking {
    liveScores()
        .onStart { emit("Loading...") }
        .onCompletion { cause -> println("onCompletion, error = ${cause?.message}") }
        .catch { e ->
            println("catch: ${e.message}")
            emit("Score unavailable")
        }
        .collect { println("UI: $it") }
}

Output

UI: Loading...
UI: IND 120/2
UI: IND 145/3
onCompletion, error = socket closed
catch: socket closed
UI: Score unavailable

Example

Retry the whole upstream when a recoverable error happens.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.io.IOException

var attempts = 0

fun prices(): Flow<Int> = flow {
    attempts++
    println("Connecting, attempt $attempts")
    if (attempts < 3) throw IOException("connection reset")
    emit(2941)
    emit(2945)
}

fun main() = runBlocking {
    prices()
        .retryWhen { cause, attempt ->
            val retry = cause is IOException && attempt < 3
            if (retry) delay(100L * (attempt + 1))
            retry
        }
        .catch { println("Giving up: ${it.message}") }
        .collect { println("Price: $it") }
}

Output

Connecting, attempt 1
Connecting, attempt 2
Connecting, attempt 3
Price: 2941
Price: 2945