gistfile1.txt

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import java.io.File
import java.util.concurrent.CountDownLatch
import kotlin.system.measureTimeMillis

data class UploadResponse(val code: Int,
                          val message: String,
                          val file: File)

fun CoroutineScope.uploadWorker(files: ReceiveChannel<File>,
                                responses: SendChannel<UploadResponse>) {
    launch {
        for (file in files) {
            // this simulates the http client call
            delay(2000)
            responses.send(UploadResponse(200, "done", file))
            println("upload file ${file.name} on thread ${Thread.currentThread().name}")
        }
    }
}

fun CoroutineScope.uploadResponseHandler(responses: ReceiveChannel<UploadResponse>,
                                         latch: CountDownLatch) {
    launch {
        for (response in responses) {
            println("upload had status ${response.code}")
            latch.countDown()
        }
    }
}

fun CoroutineScope.uploader(files: ReceiveChannel<File>,
                            latch: CountDownLatch,
                            nrOfWorkers: Int = 4) {
    val responses = Channel<UploadResponse>()
    repeat(nrOfWorkers) { uploadWorker(files, responses) }
    uploadResponseHandler(responses, latch)
}

object UploadMain {
    @JvmStatic
    fun main(args: Array<String>) {
        val files = (1..10).map { File("$it") }
        val latch = CountDownLatch(files.size)
        val channel = Channel<File>()

        val time = measureTimeMillis {
            GlobalScope.launch {
                uploader(channel, latch)
                files.forEach {
                    channel.send(it)
                }
            }
            latch.await()
        }
        println(time)
    }
}