Why does kotlin flow not trigger the transform function after an error is handled

Question

I have the below code in my view model class.

class MarketViewModel @Inject constructor(repo: MarketRepository) : ViewModel() {
    private val retry = MutableStateFlow(0)

    val marketState: LiveData<State<Market>> =
        retry.flatMapLatest{repo.refreshMarket()}
        .map { State.Success(it) as State<T> }
        .catch { error -> emit(State.Error(error)) }
        .stateIn(vmScope, SharingStarted.WhileSubscribed(5000), State.Loading())
        .asLiveData()

    fun retry() {
        retry.value++
    }
}

MarketRepository.kt:

    fun refreshMarket() =
        flow { emit(api.getMarkets()) }
            .onEach { db.upsert(it) }
            .flowOn(dispatchers.IO)

It works fine until a network error occurs in the repository method refreshMarket then when I call the retry() on the view model, it doesn't trigger the flatMapLatest transformer function anymore on the retry MutableStateFlow, why?

Does the flow get complete when it calls a Catch block? how to handle such situation?

Answer 1

You're right, catch won't continue emitting after an exception is caught. As the documentation says, it is conceptually similar to wrapping all the code above it in try. If there is a loop in a traditional try block, it does not continue iterating once something is thrown, for example:

try {
    for (i in 1..10) {
        if (i == 2) throw RuntimeException()
        println(i)
    }
} catch (e: RuntimeException) {
    println("Error!")
}

In this example, once 2 is encountered, the exception is caught, but code flow does not return to the loop in the try block. You will not see any numbers printed that come after 2.

You can use retryWhen instead of catch to be able to restart the flow. To do it on demand like you want, maybe this strategy could be used (I didn't test it):

class MarketViewModel @Inject constructor(repo: MarketRepository) : ViewModel() {
    private val retry = MutableSharedFlow<Unit>()

    val marketState: LiveData<State<Market>> =
        repo.refreshMarket()
        .map { State.Success(it) as State<T> }
        .retryWhen { error, _ -> 
            emit(State.Error(error))
            retry.first() // await next value from retry flow
            true
        }
        .stateIn(vmScope, SharingStarted.WhileSubscribed(5000), State.Loading())
        .asLiveData()

    fun retry() {
        retry.tryEmit(Unit)
    }
}