KotlinConf 2018 - Kotlin Coroutines in Practice by Roman Elizarov code Code https://youtu.be/a3agLJQ6vt8

actor.kt

const N_WORKERS = 5

class Content
class Location
class Reference {
    fun resolveLocation(): Location = TODO()
}

data class LocContent(val location: Location, val content: Content)

fun CoroutineScope.downloader(
    references: ReceiveChannel<Reference>,
    locations: SendChannel<Location>,
    contents: ReceiveChannel<LocContent>
) = launch {
    val requested = mutableMapOf<Location, MutableList<Reference>>()
    
    while (true) {
        select<Unit> {
            references.onReceive { ref ->
                val location = ref.resolveLocation()
                val refs = requested[location]
                if (refs == null) {
                    requested[location] = mutableListOf(ref)
                    locations.send(location)
                } else refs.add(ref)
            }
            contents.onReceive { (loc, content) ->
                val refs = requested.remove(loc)!!
                for (ref in refs) {
                    processContent(ref, content)
                }
            }
        }
    }
}

fun CoroutineScope.worker(
    locations: ReceiveChannel<Location>,
    contents: SendChannel<LocContent>
) = launch {
    for (location in locations) {
        val content = downloadContent(location)
        contents.send(LocContent(location, content))
    }
}

fun downloadContent(loc: Location): Content = TODO()
fun processContent(ref: Reference, content: Content): Unit = TODO()

fun CoroutineScope.processReferences(references: ReceiveChannel<Reference>) {
    val locations = Channel<Location>()
    val contents = Channel<LocContent>()

    repeat(N_WORKERS) { worker(locations, contents) }
    downloader(references, locations, contents)
}