Compose Multiplatform+kotlin Multiplatfrom第三弹

🕗 发布于 2024-09-20 22:55 kotlin 开发语言 android

前言

上篇文章kmp实战2很多通用业务功能，这篇续着发布队列文件断点下载保存到本地。

package com.your.pkName.utils

import com.benasher44.uuid.uuid4
import io.ktor.client.call.body
import io.ktor.client.request.headers
import io.ktor.client.request.prepareGet
import io.ktor.client.utils.DEFAULT_HTTP_BUFFER_SIZE
import io.ktor.http.ContentType
import io.ktor.http.HttpHeaders
import io.ktor.http.contentLength
import io.ktor.http.contentType
import io.ktor.utils.io.ByteReadChannel
import io.ktor.utils.io.core.isEmpty
import io.ktor.utils.io.core.readBytes
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.IO
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.launch
import kotlinx.datetime.Clock
import okio.FileSystem
import okio.Path
import okio.Path.Companion.toPath
import okio.SYSTEM
import okio.buffer
import okio.use
import kotlin.concurrent.Volatile

/**
 * @author by jason-何伟杰，2024/6/17
 * des:队列下载管理
 */
data class DownloadTask(
    val url: String,
    val savePath: String? = null,
    val progress: ProgressListener,//进度、速度，进度由于1024换算大概率重复
    val callback: (Boolean, String, String?) -> Unit //是否成功，suc-路径，fail-日志
) : Comparable<DownloadTask> {
    var job: Job? = null

    var id = uuid4().toString() //唯一id
    val timestamp: Long = Clock.System.now().toEpochMilliseconds()

    @Volatile
    var state: TaskState = TaskState.Pending

    //比较优先级
    override fun compareTo(other: DownloadTask): Int {
        return (timestamp - other.timestamp).toInt()
    }

    //临时地址转换？ .tmp
    override fun toString(): String {
        return "{$url ,$savePath ,$job}"
    }
}

sealed class TaskState {
    object Pending : TaskState()
    object Downloading : TaskState()
    object Paused : TaskState()
    object Completed : TaskState()
    data class Failed(val msg: String?) : TaskState()
}

object DownloadApi {
    private val downloadQueue: MutableSet<DownloadTask> = HashSet()
    private val client = createHttpClient(null) //用平台引擎的特性客户端才可以下https
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    private var lastProgressTimestamp: Long = 0
    private const val progressInterval: Long = 1600

    private fun tempFile(task: DownloadTask): String {
        //临时文件夹都是FileSystem,不是平台特性，无法直接下载到下载目录下，后续优化
//        return "${GlobalCode.getFileCacheDir()}/${GlobalCode.subFormatUrl(task.url)}.tmp"
        return GlobalCode.getGlobalDCIMPath(task.url, isCopy = true) + "/${
            GlobalCode.setUniqueFileName(
                GlobalCode.subFormatUrl(
                    task.url
                ), GlobalCode.getGlobalDCIMPath(task.url, isCopy = true)
            )
        }.tmp"
    }

    /**添加一个新任务并开始*/
    fun addTask(task: DownloadTask) {
        downloadQueue.add(task)
        processNext()
    }

    /**添加新任务队列并开始*/
    fun startTaskList(taskList: List<DownloadTask>) {
        if (!taskList.isNullOrEmpty()) {
            downloadQueue.addAll(taskList)
            processNext()
        }
    }

    /**打印存在的任务队列*/
    fun printTaskList() {
        if (downloadQueue.size == 0) {
            printLogW("downloadQueue>isEmpty")
        }
        downloadQueue.forEach { task ->
            printLogW("task>$task")
        }
    }

    /**开启任务*/
    fun startTask(task: DownloadTask?) {
//        if (task?.job == null || task.job?.isCompleted == true) {
        if (task?.job == null || task.state != TaskState.Completed) {
            task?.let {
                task.job = createDownloadJob(task)
            }
            task?.job?.start()
        }
    }

    /**重新开始任务队列*/
    fun reStartQueueTask() {
        processNext()
    }

    /**根据id查询任务实体对象*/
    fun queryTask(taskId: String): DownloadTask? {
        for (request in downloadQueue) {
            if (request.id == taskId) {
                return request
            }
        }
        return null
    }

    fun queryTaskByUrl(url: String): DownloadTask? {
        for (request in downloadQueue) {
            if (request.url == url) {
                return request
            }
        }
        return null
    }

    fun pauseTask(task: DownloadTask?) {
        task?.state = TaskState.Paused
        task?.job?.cancel()
    }

    //按照id控制比较好
    fun pauseTask(taskId: String) {
        pauseTask(queryTask(taskId))
    }

    fun pauseTaskByUrl(url: String) {
        pauseTask(queryTaskByUrl(url))
    }

    fun stopAllDownload() {
        downloadQueue.forEach {
            pauseTask(it)
        }
    }

    fun resumeTask(task: DownloadTask?) {
        downloadQueue.forEach { task ->
            if (task.state == TaskState.Downloading) {
                task.state = TaskState.Paused
                task.job?.cancel() //内部嵌套的job,要注意下有无问题
            } else if (task.state == TaskState.Completed) {
                task.state = TaskState.Completed
            } else {
                task.state = TaskState.Pending
            }
        }
        task?.state = TaskState.Downloading
        startTask(task)
    }

    fun resumeTask(taskId: String) {
        resumeTask(queryTask(taskId))
    }

    fun deleteTask(taskId: String) {
        queryTask(taskId)?.let {
            it.state = TaskState.Paused
            it.job?.cancel()
            downloadQueue.remove(it)
        }
    }

    fun clearTask() {
        downloadQueue.forEach {
            it.state = TaskState.Paused
            it.job?.cancel()
        }
        downloadQueue.clear()
    }

    //这里是每次轮询任务的逻辑
    private fun processNext() {
        val task = downloadQueue.firstOrNull() { it.job == null || it.state != TaskState.Completed }
//        printLogW("task>$task ${task?.job} ${task?.job?.isCompleted}")
        task?.let { startTask(it) }
    }

    private fun createDownloadJob(task: DownloadTask): Job {
        return scope.launch(start = CoroutineStart.LAZY, context = Dispatchers.IO) {
            try {
                if (task.url.isEmpty()) {
                    throw Exception("cannot download null url")
                } else {
                    //优先创建缓存目录，不然报错
                    createPlatformRootDir()
                    //构建文件路径
                    var tmpFile: Path
                    if (task.savePath.isNullOrEmpty()) {
                        tmpFile = tempFile(task).toPath()
                    } else {
                        tmpFile = (task.savePath + ".tmp").toPath()
                    }
                    printLogW("file>$tmpFile") //应该用临时格式后缀，最后再转
                    createPlatformFile(tmpFile.toString())
                    task.state = TaskState.Downloading
                    val existLength = FileSystem.SYSTEM.metadataOrNull(tmpFile)?.size
                    var sumBytes: Long = 0
                    var startSize: Long = existLength ?: 0 //计算速度
                    lastProgressTimestamp = Clock.System.now().toEpochMilliseconds()
                    client.prepareGet(task.url) {
                        headers {
                            getCacheStr(KmmConfig.DATA_SINGLE_COOKIE)?.let {
                                append("Cookie", it)
                            }
                            existLength?.let {//断点下载剩下的全部
                                if (existLength != 0L)
                                    append(HttpHeaders.Range, "bytes=$existLength-")
                            }
                        }
                        contentType(ContentType.Any)
//                        timeout {  //不设置会更好不
//                            requestTimeoutMillis = 30 * 1000
//                            connectTimeoutMillis = 300 * 1000  //下载大文件超时会报错 600
//                        }
                    }.execute { httpResponse ->
                        val channel: ByteReadChannel = httpResponse.body()
                        if (httpResponse.contentLength() == null || httpResponse.contentLength() == 0L) {
                            (GlobalCode.getOneParameter(task.url, "sum"))?.let {
                                sumBytes = (it.toDouble() * 1024).toLong()
                            }
                        }
                        var pp = 0
                        FileSystem.SYSTEM.appendingSink(tmpFile, true).buffer().use { sink ->
                            while (!channel.isClosedForRead && task.state != TaskState.Paused) {//!task.isPaused
                                val packet =
                                    channel.readRemaining(DEFAULT_HTTP_BUFFER_SIZE.toLong())
                                //捕抓网络突然中断的问题
                                if (packet.isEmpty) {
                                    updateFailed(task, "channel packet isEmpty")
                                    return@use
                                }
                                while (!packet.isEmpty) {
                                    val bytes = packet.readBytes()
                                    sink.write(bytes)
                                    val now = Clock.System.now().toEpochMilliseconds()
                                    if ((now - lastProgressTimestamp) < progressInterval) {

                                    } else {
                                        //计算速度
                                        var speed = 0
                                        val deltaTime = (now - lastProgressTimestamp + 1).toInt()

                                        FileSystem.SYSTEM.metadataOrNull(tmpFile)?.size?.let { size ->
                                            speed =
                                                ((size - startSize) * 1000 / deltaTime).toInt()
                                            startSize = size
//                                            printLogW("speed>$speed")
                                        }
                                        lastProgressTimestamp = now
                                        //获取进度
                                        if ((httpResponse.contentLength() == null || httpResponse.contentLength() == 0L)
                                            && sumBytes != 0L
                                        ) {
                                            //用流下载是无法提前知道文件总长度
//                                    printLogD("total>${channel.totalBytesRead} ")
                                            var tp = 0L
                                            FileSystem.SYSTEM.metadataOrNull(tmpFile)?.size?.let { l ->
                                                tp = l
                                            }
//                                            printLogW("onprogress1>")
                                            task.progress.onProgress(
                                                (tp * 100 / sumBytes).toInt()
                                                    .apply { pp = this },
                                                speed = speed,
                                                url = task.url
                                            )
                                        } else {
//                                            printLogW("onprogress2>")
                                            task.progress.onProgress(
                                                (channel.totalBytesRead * 100 / httpResponse.contentLength()!!).toInt()
                                                    .apply { pp = this }, speed, url = task.url
                                            )
                                        }
                                    }
                                }
                                if (task.state == TaskState.Paused) { //not try
                                    httpResponse.cancel()
                                }
                            }
                            //下载结束，计算的文件总大小不准，有误差
                            if (channel.isClosedForRead) { //&& pp > 95 ,速度过快进度会中间会少值
                                //如果突然中断网络也是close,status也是200
                                if (FileSystem.SYSTEM.exists(tmpFile)) {
//                                    val cur = Clock.System.now()
                                    val desPath = tmpFile.toString().replace(".tmp", "")
//                                    FileSystem.SYSTEM.atomicMove( //这耗时太大了，是移动文件，不算改名
//                                        tmpFile,
//                                        desPath.toPath()
//                                    )
//                                    saveStr(GlobalCode.subFormatUrl(task.url), desPath)
//                                    task.callback(true, desPath)

                                    //如果文件已存在，重命名也耗时，尴尬
                                    renameFile(tmpFile.toString(), desPath)
                                    saveStr(GlobalCode.subFormatUrl(task.url), desPath)
                                    printLogW("callback???")
                                    task.callback(true, task.url, desPath) //结束
//                                    injectAct()
//                                    printLogW("cost-time>${Clock.System.now() - cur}")
                                    if (GlobalCode.canPreview2DCIM(
                                            GlobalCode.getFileTypeByUrl(task.url).toUpperCase()
                                        )
                                    ) {
                                        if (task.url.contains("water=true")) {
                                            createWaterPic(desPath, task.url)
                                        } else {
                                            val path1 = saveMedia2Gallery(desPath, callback = { s ->
//                                                printLogW("copyPath>$s")
                                                s?.let {
                                                    saveStr(GlobalCode.subFormatUrl(task.url), s)
                                                }
                                             //   DownloadHelper.upGradeDownload(task.url, s)
                                            })
//                                            printLogW("path1>$path1")
                                            //通知更新ui
                                          //  EventHelper.post(Event.RefreshEvent)
                                        }
                                    } else {
                                        DownloadHelper.upGradeDownload(task.url, desPath)
//                                        printLogW("${GlobalCode.fileExist(desPath)}")
                                        //通知更新ui
                                        EventHelper.post(Event.RefreshEvent)
                                    }
                                }
                                task.state = TaskState.Completed
                                downloadQueue.remove(task)
                                processNext()
                            }
                        }
                    }
                }
            } catch (e: Exception) {
                e.printStackTrace()
                if (task.state != TaskState.Paused) {
                    task.state = TaskState.Failed(e.message)
                    task.callback(false, task.url, e.message)
                    //需要提示还是直接删除,但是有错误时不删会导致无限循环启动
                    downloadQueue.remove(task)
                    processNext()
                }
            }
        }
    }

    private fun updateFailed(task: DownloadTask?, msg: String?, e: Throwable? = null) {
        task?.let {
            e?.let {
                task.state = TaskState.Failed(e.message)
            }
            task.callback(false, task.url, msg)
            //需要提示还是直接删除,但是有错误时不删会导致无限循环启动
            downloadQueue.remove(task)
            processNext()
        }
        if (isConnected() || true) { //无网络
          //  saveBoolean(KmmConfig.DATA_DOWNLOAD_STOP, true)
         //   EventHelper.post(Event.MyDownloadRefreshEvent(type = FileInfoBean.TYPE_NET_ERROR))
         //类似EventBus的功能，跟我的业务强绑定了，这里参考可以不要
        }
    }
}

interface ProgressListener {
    fun onProgress(progress: Int, speed: Int, url: String)
}

启动队列下载任务,整个队列下载框架处理的任务对象是DownloadTask类型，我们平时是在页面列表多选项构造出新的下载队列，用covert2TaskList()转换成框架处理的数据集。

 DownloadApi.startTaskList(covert2TaskList(newList))
/**将数据源转为队列任务数据*/
fun covert2TaskList(list: MutableList<FileInfoBean>?): List<DownloadTask> {
    val taskList = mutableListOf<DownloadTask>()
    list?.forEach { bean ->
        val desPath = GlobalCode.getGlobalDCIMPath(bean.downloadUrl, isCopy = true) + "/${
            DownloadHelper.setFileNameByUrl(bean.downloadUrl!!)
        }"
        taskList.add(
            DownloadTask(
                url = bean.downloadUrl!!,
                savePath = desPath, progress = createProgressListener(), callback = createCallback()
            )
        )
    }
    return taskList
}

/**回调每个下载进度*/
private fun createProgressListener(): ProgressListener {
    val listener = object : ProgressListener {
        override fun onProgress(progress: Int, speed: Int, url: String) {
//            val str = "progress>$progress , speed>$speed"
//            printLogW(str)
        //    updateProgressData(progress, speed, url) //自己的业务数据更新到本地
          //  refreshUi() //更新前台ui刷新
        }
    }
    return listener
}

/**回调下载结束是否成功 suc,url,path*/
private fun createCallback(): (Boolean, String, String?) -> Unit {
    val callBack = { suc: Boolean, url: String, msg: String? ->
        printLogW("callback>$suc ,$msg")
        if (suc) {
          //  refreshUi(downloadUrl = url, localPath = msg)
        }
    }
    return callBack
}

/**将数据源转为队列任务数据*/
fun covert2TaskList(list: MutableList<FileInfoBean>?): List<DownloadTask> {
    val taskList = mutableListOf<DownloadTask>()
    list?.forEach { bean ->
        val desPath = GlobalCode.getGlobalDCIMPath(bean.downloadUrl, isCopy = true) + "/${
            DownloadHelper.setFileNameByUrl(bean.downloadUrl!!)
        }"
        taskList.add(
            DownloadTask(
                url = bean.downloadUrl!!,
                savePath = desPath, progress = createProgressListener(), callback = createCallback()
            )
        )
    }
    return taskList
}

//在Platform.android.kt的 ,Platform.ios.kt可以重写一次一样的
actual fun createHttpClient(timeout: Long?): HttpClient {
    return HttpClient {
        defaultRequest {
            url.takeFrom(URLBuilder().takeFrom("http://wfserver.gree.com/"))
        }
        install(HttpTimeout) {
            timeout?.let {
                requestTimeoutMillis = timeout
            }
        }
        install(ContentNegotiation) {
            json(Json {
                ignoreUnknownKeys = true
                prettyPrint = true
            })
        }
        install(Logging) {
//                level = LogLevel.BODY
//            level=LogLevel.HEADERS
            level= LogLevel.INFO
//            level = LogLevel.NONE //接口日志屏蔽
            logger = object : io.ktor.client.plugins.logging.Logger {
                override fun log(message: String) {
                    println(message)
                }
            }
        }
    }
}

@Serializable
class FileInfoBean : Comparable<FileInfoBean> {

    companion object {
        const val TYPE_DONE = 0     //已下载
        const val TYPE_DOWNLOADING = 1 //下载中
        const val TYPE_HEADER = 2   //下载日期布局
        const val TYPE_NET_ERROR = 3  //网络错误
        const val TYPE_FILE_ERROR = 4 //无法正常下载的文件，如文件防火墙隔离
        const val TYPE_NET_STOP = 5  //暂停下载
        const val TYPE_FILTER=6 //6= 剔除gdoc文件并开始,7=复制成功,8=权限错误
        const val TYPE_COPY=7
        const val TYPE_PERMISSION=8
    }

    var fileName: String
    var fileSize: Double = 0.0
    var downloadStatus: Int = 0 //是否已下载 0-已完成/未开始 1-进行中
    var createTime: Long = Clock.System.now().toEpochMilliseconds()
    var fileFlag: Int = 0   //是分组名还是文件项 0-已完成 1-下载中 2-组名(日期)
    var present: Int = 0    //进度
    var localPath: String? = null
    var downloadUrl: String? = null
     var id: Int = -1
    var speed: Int = 0
 override fun compareTo(other: FileInfoBean): Int {  //以时间作为排序依据
        return this.createTime.compareTo(other.createTime)
    }


    fun clone(): FileInfoBean { //要实现实体复制
        return FileInfoBean(
            this.fileName,
            this.fileFlag,
           ....//自己补充属性吧，我业务上的实体属性很多上面已经删除了部分
    }

注意事项

saveStr(GlobalCode.subFormatUrl(task.url), desPath) ，这是代表以下载链接最后一截为key保存下载路径到本地，保存的库我是com.russhwolf:multiplatform-settings，自行实现savaStr的封装就好。
断点下载的关键是请求的头部header的Range字节范围，val existLength = FileSystem.SYSTEM.metadataOrNull(tmpFile)?.size我是依靠这个获取已下载的文件大小，这样就不需要记录每次的下载位置。
iOS的下载文件路径每次获取都被沙盒重构再输出，就是会变的记录也没什么用，他获取文件是靠label的标识，上篇文章有提到，比较难处理的文件管理，android和iOS差异挺大的，而且android内部迭代后权限和存储又不一样。
模拟了几种下载突然中断的场景，网络断开，忽然杀掉app，重新启动app也可正常断点下载。
我开始试过用下载进度来代表下载结束，但是进度的计算有误差容易导致下载到99%没跳进结束逻辑，但是下载其实已经结束，而且我内部下载进度是1600ms的间隔对外通知，后来试channel.isClosedForRead就很好判断。
DownloadHelper类是我自己的业务数据更加队列下载情况处理，可以注释掉，正常下载肯定有进度，速度，下载路径，而且缓存文件的信息等，我也把文件拆成已下载和下载中，当然是我的业务上的逻辑，注释即可。
还有问题，大家可以思考的，队列下载这里是并没有启动线程池并发下再合并文件块，这里是循环下载结束一个接着下一个，那么看createDownloadJob每次创建一个Job,在他的下载结束回调会进入processNext()，这里是获取下一个任务在创建createDownloadJob，相当于Job之间是嵌套了，但是最佳应该是结束一个下载Job也自动被协程释放。这里我也没想好>_<。

原文地址：https://blog.csdn.net/j7a2son/article/details/142200471

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：leetcode 165.比较版本号
下一篇：go多线程

【go从零单排】上下文（context）
在 Go 语言中，上下文（context）是一个重要的概念，主要用于管理请求的生命周期、传递取消信号、超时控制以及传递请求范围内的值。上下文通常与并发编程相关，尤其是在处理 HTTP 请求和其他 I/
阅读更多2024-11-15
Ubuntu 20.04 配置开发环境(持续更新)
搜狗输入法不能显示中文。
阅读更多2024-11-15
在ubuntu上安装ubuntu22.04并ros2 humble版本的docker容器记录
"registry-mirrors" : ["https://你的.mirror.swr.myhuaweicloud.com"],4、已经打包了ubuntu22
阅读更多2024-11-15
如何在 Ubuntu 22.04 上安装 ownCloud
ownCloud 是一个开源的个人云存储平台，它允许用户在本地服务器上存储和同步文件，提供了一个类似于 Dropbox 或 Google Drive 的服务，但是更加注重隐私和数据控制。文件存储：用户
阅读更多2024-11-15
idea 删除本地分支后，弹窗 delete tracked brank
在 IntelliJ IDEA 中，删除本地分支后弹出 “Delete tracked branch” 的提示，这表示删除的是一个跟踪分支（tracked branch）。具体来说，当前删除的本地分
阅读更多2024-11-15
「JVM详解」
JVM：全称 Java Virtual Machine，即 Java 虚拟机，一种规范，本身是一个虚拟计算机，直接和操作系统进行交互，与硬件不直接交互，而操作系统可以帮我们完成和硬件进行交互的工作
阅读更多2024-11-15
WPF中Prism框架的简单使用
自动匹配导入4.写Login.XAML代码和LoginViewModel代码LoginViewModel如下：5.创建其他的UserControl 并学好MainWindow.XAML和MainWin
阅读更多2024-11-15
Linux——GPIO输入输出裸机实验
在程序启动时，BSS段会被清零，并且其大小会被计算到程序的总内存占用中，尽管它在磁盘上的表示可能非常小或甚至没有。在程序启动时，BSS段会被清零，并且其大小会被计算到程序的总内存占用中，尽管它在磁盘上
阅读更多2024-11-15
planRAG运行记录
运行planRAG过程记录
阅读更多2024-11-15
vue面试题8|[2024-11-14]
vue面试题
阅读更多2024-11-15

Compose Multiplatform+kotlin Multiplatfrom第三弹

前言

注意事项

相关文章