Im investigating the use of corountines in my curr...

# coroutines

[Tower Guidev2]

Im investigating the use of corountines in my current Android project.

Specifically kotlinx.coroutines.channels.actor in my Android `androidx.work.CoroutineWorker`(s).

What I am trying to achieve is that my Worker makes multiples calls to a remote API endpoint
and each API response is sent to an actor channel to be processed on another thread so that I can immediatley
start the API call for subsequent pages of data

I have these message types

sealed class Message {
class Responsible(val value: Response<List<MyApiData>>) : Message()
class GetValue(val deferred: CompletableDeferred<Int>) : Message()
}

This Actor

fun CoroutineScope.persistActor() = actor<Message>(context = Dispatchers.Unconfined, capacity = Channel.UNLIMITED) {
for (msg in channel) {
when (msg) {
is Message.Responsible -> managePage(msg.value)
is Message.GetValue -> msg.deferred.complete(1)
}
}
}

private lateinit var persister: SendChannel<Message>

each page returned from my API is processed by this recursively called function:-

private suspend fun managePages(accessToken: String, response: Response<List<MyApiData>>) {
when {
result != null -> return
response.isSuccessful -> persister.send(Message.Responsible(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.documents(accessToken, parts[1]))
}
}

Once all the pages have been retrieved from the remote end point I execute the following code to await the actor to complete persisting the returned data

val completed = CompletableDeferred<Int>()
persister.send(Message.GetValue(completed))
println("Counter = ${completed.await()}")

persister.close()

what I am concerned about

1). is this a "good" approach?
2). Does my worker wait for all data to be persisted before completing?
3). What improvements could I make?

[gildor]

It's not clear for me why you need actor, just launching multiple jobs using launch from coroutineScope block should be enough (or async + awaitAll()), scope will wait all of them to complete

[gildor]

Your current approach looks overcomplicated and quite brittle.

But maybe there are some additional requirements

Would be nice to see some example what you trying to achieve, because usage of CoroutineWorker is not different from any other suspend function

[Tower Guidev2]

My API endpoint returns a maximum 500 data items each Page. These API calls can take up to 1.5 seconds to return. While waiting for the NEXT page to return I can be insertinging the PREVIOUS pages data.

[Rob]

The actor pattern is best used when state needs to be synchronized from events generated asynchronously.
Here is what I'm using for my concurrent API calls.

Untitled.kt

[Tower Guidev2]

I require to extract the NEXT Http Header Link from call1 before I can make Call2, then I cannot make Call3 until I have the Link from Call2 etc...

[Rob]

I see. Then you would get first result for the next call like this.

Untitled.kt

[Rob]

I think this works as well without need to to jump back and fourth between Main and IO threads between calls.

Untitled.kt

[Rob]

If it wasn’t clear from the examples, the apiCall1() functions are suspend functions on Retrofit services.

[gildor]

still not quite sure. If you just need result of previous operation simple while is enough, like any non-suspend code, not sure why you need actor for this

[gildor]

If you need intermediate results (so it’s a stream of data) just create a flow, it a lot easier to implement and use

[gildor]

fun allPages(): = flow<Page> {
    val firstPage = api.loadFirstPage() // suspend call to API
    emit(firstPage)
    var nextPageLink = firstPage.nextPageLink
    while(nextPageLink != null) {
        val currentPage = api.loadPage(nextPageLink) // suspend call to API
        nextPageLink = currentPage.nextPageLink
        emit(firstPage)
    }
}

[gildor]

so now you can use it any way how you want:

allPages().collect { doSomethingWith(it) }

// or if you want get all data too:
allPages().onEach { 
   // Still possible to access every page
   doSomethingWith(it) 
}.toList() // but as result collect all the data as List<Page> and await result

[Tower Guidev2]

The Flow approach does look promising, thanks very much for taking the time to look at this 😄

[gildor]

It's not much different from using callback, just more idiomatic and flexible