How can I download a large file with Ktor and Kotlin with a progress indicator?

Question

I've been spending way too much time trying to solve this problem. So the code that I posted below does work in terms of downloading a file, but the problem is, the flow has a very unexpected behaviour. The response.content.readAvailable() method call seems to block until it's completely done downloading the whole file at which point the emit progress happens, so you end up waiting a long time for the file to download, and then in a split second you get all of the progress updates. So I'm wondering if there is a way to do this where I read in a certain number of bytes at a time and then emit a progress and then repeat until the file is done downloading? Or maybe a way to hook into the readAvailable() method and update the progress that way? Any help with this would be greatly appreciated.

Here's the code I found and modified, but still does not work right:

suspend fun HttpClient.downloadFile(
    output: File,
    downloadUrl: String,
    md5Hash: String,
) = flow {
    try {
        val response = get<HttpResponse> { url(downloadUrl) }
        val data = ByteArray(response.contentLength()?.toInt() ?: 0)
        val contentLn = response.contentLength()?.toInt() ?: 0
        var offset = 0
        var bytesRemaining = contentLn
        do {
            val chunkSize = min(maxChunkSize, bytesRemaining)
            logger?.d { "Read Available:" }
            val result = response.content.readAvailable(data, offset, length = chunkSize)
            val progress = ((offset / contentLn.toDouble()) * 100).toInt()
            emit(DownloadResult.Progress(progress))
            logger?.d { "logged progress: $progress" }
            // delay(6000L) this was to test my assumption that the readAvalible was blocking. 
            offset += chunkSize
            bytesRemaining -= chunkSize
        } while (result != -1)

        if (response.status.isSuccess()) {
            if (data.md5().hex == md5Hash) {
                output.write(data)
                emit(DownloadResult.Success)
            } else {
                emit(DownloadResult.ErrorCorruptFile)
            }
        } else {
            emit(DownloadResult.ErrorBadResponseCode(response.status.value))
        }
    } catch (e: TimeoutCancellationException) {
        emit(DownloadResult.ErrorRequestTimeout("Connection timed out", e))
    }
}

Intervalometer

Answer 1

Finally after a stupid amount of time I solved this. What you need to use is this. That gives you access to the byte channel as it is downloading.

and a very crude implementation (that I'm not yet done with) is this:

    get<HttpStatement>(url = downloadUrl).execute {
        var offset = 0
        val byteBufferSize = 1024 * 100
        val channel = it.receive<ByteReadChannel>()
        val contentLen = it.contentLength()?.toInt() ?: 0
        val data = ByteArray(contentLen)
        do {
            val currentRead = channel.readAvailable(data, offset, byteBufferSize)
            val progress = if(contentLen == 0) 0 else ( offset / contentLen.toDouble() ) * 100
            logger?.d { "progress: $progress" }
            offset += currentRead
        } while (currentRead >= 0)

    }

two things to not with this solution. 1.) I'm in the context of HttpClient, so that's how I have access to get(). 2.) I'm creating a byte buffer size of 1024 * 100 in order to not let the readAvailable method block for too long, though this might not be necessary... the one nice thing about it is that it determines how frequently you will be publishing your progress updates.

Intervalometer

Answer 2

get<HttpStatement>() in Alex's answer is for Ktor 1.x. In Ktor 2.x, as the updated documents mention, we should use the prepareGet() function instead (official example).

Led

Answer 3

For future reference an example using Ktor 2 with prepereGet()

suspend fun downloadFileWithProgress(
    url: String,
    outputStream: OutputStream,
    onProgress: (Float) -> Unit
) {
   httpClient.prepareGet(
            urlString = url,
            block = {
                val timeout = 30.minutes.inWholeMilliseconds
                timeout {
                    requestTimeoutMillis = timeout
                    connectTimeoutMillis = timeout
                    socketTimeoutMillis = timeout
                }

                onDownload { bytesSentTotal, contentLength ->
                    val progress = (bytesSentTotal.toFloat() / contentLength.toFloat())
                    onProgress(progress)
                }
            }
        ).execute { response ->

            if (response.status.value in 200..299) {
                val byteReadChannel = response.bodyAsChannel()

                byteReadChannel.copyTo(outputStream)

            } else {
                Log.e("App", "Failed to download file. HTTP Status: ${response.status.value}")
            }
        }
}

If you want to use Flow, you can easiely emit at onDownload {}

For timeout you need Ktor Timeout Plugin Dependency. For myself I had to enable it, as otherwise the connection breacks immeadiatly.

Alfredalfreda