How to complete a Kotlin Flow in Android Worker

user_423199 0
Viewed 879

I'm investigating the use of Kotlin Flow within my current Android application

My application retrieves its data from a remote server via Retrofit API calls.

Some of these API's return 50,000 data items in 500 item pages.

Each API response contains an HTTP Link header containing the Next pages complete URL.

These calls can take up to 2 seconds to complete.

In an attempt to reduce the elapsed time I have employed a Kotlin Flow to concurrently process each page of data while also making the next page API call.

My flow is defined as follows:

private val persistenceThreadPool = Executors.newFixedThreadPool(3).asCoroutineDispatcher()
private val internalWorkWorkState = MutableStateFlow<Response<List<MyPage>>?>(null)
private val workWorkState = internalWorkWorkState.asStateFlow()

private val myJob: Job

init {
    myJob = GlobalScope.launch(persistenceThreadPool) {
        workWorkState.collect { page ->
            if (page == null) {
            } else managePage(page!!)
        }
    }
}

My Recursive function is defined as follows that fetches all pages:-

    private suspend fun managePages(accessToken: String, response: Response<List<MyPage>>) {
        when {
            result != null -> return
            response.isSuccessful -> internalWorkWorkState.emit(response)
            else -> {
                manageError(response.errorBody())
                result = Result.failure()
                return
            }
        }

        response.headers().filter { it.first == HTTP_HEADER_LINK && it.second.contains(REL_NEXT) }.forEach {
            val parts = it.second.split(OPEN_ANGLE, CLOSE_ANGLE)
            if (parts.size >= 2) {
                managePages(accessToken, service.myApiCall(accessToken, parts[1]))
            }
        }
    }

   private suspend fun managePage(response: Response<List<MyPage>>) {
        val pages = response.body()

        pages?.let {
            persistResponse(it)
        }
    }

    private suspend fun persistResponse(myPage: List<MyPage>) {
        val myPageDOs = ArrayList<MyPageDO>()

        myPage.forEach { page ->
            myPageDOs.add(page.mapDO())
        }

        database.myPageDAO().insertAsync(myPageDOs)
    }

My numerous issues are

  1. This code does not insert all data items that I retrieve

  2. How do complete the flow when all data items have been retrieved

  3. How do I complete the GlobalScope job once all the data items have been retrieved and persisted

UPDATE

By making the following changes I have managed to insert all the data

 private val persistenceThreadPool = Executors.newFixedThreadPool(3).asCoroutineDispatcher()
    private val completed = CompletableDeferred<Int>()

    private val channel = Channel<Response<List<MyPage>>?>(UNLIMITED)
    private val channelFlow = channel.consumeAsFlow().flowOn(persistenceThreadPool)

    private val frank: Job

    init {
        frank = GlobalScope.launch(persistenceThreadPool) {
            channelFlow.collect { page ->
                if (page == null) {
                    completed.complete(totalItems)
                } else managePage(page!!)
            }
        }
    }


...
...
...

   channel.send(null)
   completed.await()

   return result ?: Result.success(outputData)

I do not like having to rely on a CompletableDeferred, is there a better approach than this to know when the Flow has completed everything?

kotlin-flow android
2 Answers

You are looking for the flow builder and Flow.buffer():

suspend fun getData(): Flow<Data> = flow {
    var pageData: List<Data>
    var pageUrl: String? = "bla"
    while (pageUrl != null) {
        TODO("fetch pageData from pageUrl and change pageUrl to the next page")
        emitAll(pageData)
    }
}
    .flowOn(Dispatchers.IO /* no need for a thread pool executor, IO does it automatically */)
    .buffer(3)

You can use it just like a normal Flow, iterate, etc. If you want to know the total length of the output, you should calculate it on the consumer with a mutable closure variable. Note you shouldn't need to use GlobalScope anywhere (ideally ever).

user_55095750

There are a few ways to achieve the desired behaviour. I would suggest to use coroutineScope which is designed specifically for parallel decomposition. It also provides good cancellation and error handling behaviour out of the box. In conjunction with Channel.close behaviour it makes the implementation pretty simple. Conceptually the implementation may look like this:

 suspend fun fetchAllPages() {
    coroutineScope {
        val channel = Channel<MyPage>(Channel.UNLIMITED)
        launch(Dispatchers.IO){ loadData(channel) }
        launch(Dispatchers.IO){ processData(channel) }
    }
}

suspend fun loadData(sendChannel: SendChannel<MyPage>){
    while(hasMoreData()){
        sendChannel.send(loadPage())
    }
    sendChannel.close()
}

suspend fun processData(channel: ReceiveChannel<MyPage>){
    for(page in channel){
        // process page
    }
}

It works in the following way:

  1. coroutineScope suspends until all children are finished. So you don't need CompletableDeferred anymore.
  2. loadData() loads pages in cycle and posts them into the channel. It closes the channel as soon as all pages have been loaded.
  3. processData fetches items from the channel one by one and process them. The cycle will finish as soon as all the items have been processed (and the channel has been closed).

In this implementation the producer coroutine works independently, with no back-pressure, so it can take a lot of memory if the processing is slow. Limit the buffer capacity to have the producer coroutine suspend when the buffer is full. It might be also a good idea to use channels fan-out behaviour to launch multiple processors to speed up the computation.

user_25489880
Related