Kotlin Coroutine Flow: Limit the number of collector

Question

Is there a way to limit the number of collector in a function that returns a Flow using flow builder?

I have this public method in a ViewModel

fun fetchAssets(limit: String) {

        viewModelScope.launch {

            withContext(Dispatchers.IO){
                getAssetsUseCase(AppConfigs.ASSET_PARAMS, limit).onEach {

                    when (it) {

                        is RequestStatus.Loading -> {
                            _assetState.tryEmit(AssetState.FetchLoading)
                        }

                        is RequestStatus.Success -> {
                            _assetState.tryEmit(AssetState.FetchSuccess(it.data.assetDataDomain))
                        }

                        is RequestStatus.Failed -> {
                            _assetState.tryEmit(AssetState.FetchFailed(it.message))
                        }

                    }

                }.collect()
            }

        }

    }

This method is called on ViewModel's init block, but can also be called manually on UI.

This flow emits value every 10 seconds.

Repository

override fun fetchAssets(
        query: String,
        limit: String
    ) = flow {
        while (true) {
            try {
                interceptor.baseUrl = AppConfigs.ASSET_BASE_URL
                emit(RequestStatus.Loading())
                val domainModel = mapper.mapToDomainModel(service.getAssetItems(query, limit))
                emit(RequestStatus.Success(domainModel))
            } catch (e: HttpException) {
                emit(RequestStatus.Failed(e))
            } catch (e: IOException) {
                emit(RequestStatus.Failed(e))
            }
            delay(10_000)
        }
    }

Unfortunately every time fetch() was invoke from UI, I noticed that it creates another collectors thus can ended up having tons of collector which is really bad and incorrect.

The idea is having a flow that emits value every 10 seconds but can also be invoke manually via UI for immediate data update without having multiple collectors.

What type does `_assetState` property have? Is it `StateFlow`? — Sergio, Mar 19 '22 at 13:34

broot · Accepted Answer · 2022-03-19T18:44:35.743

You seem to misunderstand what does it mean to collect the flow or you misuse the collect operation. By collecting the flow we mean we observe it for changes. But you try to use collect() to introduce changes to the flow, which can't really work. It just starts another flow in the background.

You should collect the flow only once, so keep it inside init or wherever it is appropriate for your case. Then you need to update the logic of the flow to make it possible to trigger reloading on demand. There are many ways to do it and the solution will differ depending whether you need to reset the timer on manual update or not. For example, we can use the channel to notify the flow about the need to reload:

val reloadChannel = Channel<Unit>(Channel.CONFLATED)

fun fetchAssets(
    query: String,
    limit: String
) = flow {
    while (true) {
        try {
            ...
        }
        
        withTimeoutOrNull(10.seconds) { reloadChannel.receive() } // replace `delay()` with this
    }
}

fun reload() {
    reloadChannel.trySend(Unit)
}

Whenever you need to trigger the manual reload, do not start another flow or invoke another collect() operation, but instead just invoke reload(). Then the flow that is already being collected, will start reloading and will emit state changes.

This solution resets the timer on manual reload, which I believe is better for the user experience.

If this solution resets the timer on manual reload, then upon invoking `fun reload()` we will still need to wait for another 10 seconds before the task was executed? Because that is what I am experiencing right now after following your code, calling `reload()` does not actually execute the code block inside the flow immediately. Can't we ignore the timer for on demand reload, because if the current timer is in 5 seconds already and I call `reload()` it looks like I ended up having 15 second in total? — Bitwise DEVS, Mar 20 '22 at 03:54
It should reload immediately when invoked `reload()`. See: https://pl.kotl.in/XFGrGU0FB . If it doesn't in your case then I think there must be something else going on in your code that delays this. — broot, Mar 20 '22 at 10:28
Marking this as answer, although I found a better solution this answer is using the same approach but for some reason it is better to move this approach on ViewModel and use `shareIn` or `stateIn` which will make it more optimize — Bitwise DEVS, Mar 24 '22 at 16:35

score 0 · Answer 2 · answered Mar 20 '22 at 09:53

I ended up moving the timer on ViewModel as I can request on demand fetch while also not having multiple collectors that runs at the same time.

private var job: Job? = null

    private val _assetState = defaultMutableSharedFlow<AssetState>()

    fun getAssetState() = _assetState.asSharedFlow()

    init {
        job = viewModelScope.launch {
            while(true) {
                if (lifecycleState == LifeCycleState.ON_START || lifecycleState == LifeCycleState.ON_RESUME)
                    fetchAssets()
                delay(10_000)
            }
        }
    }

    fun fetchAssets() {

        viewModelScope.launch {

            withContext(Dispatchers.IO) {
                getAssetsUseCase(
                    AppConfigs.ASSET_BASE_URL,
                    AppConfigs.ASSET_PARAMS,
                    AppConfigs.ASSET_SIZES[AppConfigs.ASSET_LIMIT_INDEX]
                ).onEach {

                    when(it){

                        is RequestStatus.Loading -> {
                            _assetState.tryEmit(AssetState.FetchLoading)
                        }

                        is RequestStatus.Success -> {
                            _assetState.tryEmit(AssetState.FetchSuccess(it.data.assetDataDomain))
                        }

                        is RequestStatus.Failed -> {
                            _assetState.tryEmit(AssetState.FetchFailed(it.message))
                        }

                    }

                }.collect()
            }

        }

    }

    override fun onCleared() {
        job?.cancel()
        super.onCleared()
    }

Please correct me if this one is a code smell.

Kotlin Coroutine Flow: Limit the number of collector

2 Answers2