🚀 Kotlin+Compose+协程+Channel 多文件异步分片断点续传

class UploadTask(
    val file: File,
    val chunkSize: Int = 1024 * 1024 // 默认1MB分片
) {
    // 文件分片信息存储
    data class Chunk(
        val index: Int,
        val start: Long,
        val end: Long,
        var status: Status = Status.PENDING
    )
    
    enum class Status { PENDING, UPLOADING, COMPLETED, FAILED }
    
    // 分片计算逻辑
    fun calculateChunks(): List<Chunk> {
        val chunks = mutableListOf<Chunk>()
        val fileSize = file.length()
        var offset = 0L
        
        for (i in 0 until ceil(fileSize / chunkSize.toDouble()).toInt()) {
            val end = min(offset + chunkSize, fileSize)
            chunks.add(Chunk(i, offset, end))
            offset = end
        }
        return chunks
    }
    
    // 续传恢复逻辑
    fun restoreFromCache(cache: List<Chunk>) {
        // 实现省略：比对缓存恢复进度状态
    }
}

class UploadManager(
    private val scope: CoroutineScope,
    private val maxConcurrency: Int = 3 // 并发上传数
) {
    // 状态Channel（用于UI更新）
    private val _progressState = Channel<UploadProgress>()
    val progressState = _progressState.receiveAsFlow()
    
    // 分片上传协程
    suspend fun uploadTask(task: UploadTask) {
        val chunks = task.calculateChunks()
        val channel = produce {
            chunks.filter { it.status != UploadTask.Status.COMPLETED }
                .forEach { send(it) }
        }
        
        // 启动并发协程组
        (1..maxConcurrency).map { workerId ->
            scope.launch(Dispatchers.IO) {
                for (chunk in channel) {
                    try {
                        uploadChunk(task.file, chunk).let {
                            _progressState.send(
                                UploadProgress(task.id, chunk.index)
                            )
                        }
                    } catch (e: Exception) {
                        // 错误处理
                    }
                }
            }
        }.joinAll() // 等待所有分片完成
    }
    
    // 单个分片上传逻辑
    private suspend fun uploadChunk(file: File, chunk: UploadTask.Chunk) {
        return withContext(Dispatchers.IO) {
            val inputStream = RandomAccessFile(file, "r").use { raf ->
                raf.seek(chunk.start)
                ByteArray((chunk.end - chunk.start).toInt()).apply {
                    raf.read(this)
                }
            }
            
            // 使用OkHttp上传分片
            OkHttpClient().newCall(
                Request.Builder()
                    .url("https://api.yourserver.com/upload")
                    .addHeader("Content-Range", "bytes ${chunk.start}-${chunk.end}/${file.length()}")
                    .post(inputStream.toRequestBody())
                    .build()
            ).execute().use { response ->
                if (!response.isSuccessful) throw IOException("Upload failed")
            }
        }
    }
}

@Composable
fun FileUploadScreen(viewModel: UploadViewModel = viewModel()) {
    val tasks by viewModel.tasks.collectAsState()
    
    LazyColumn {
        items(tasks) { task ->
            FileUploadItem(
                task = task,
                onPause = { viewModel.pauseUpload(task.id) },
                onResume = { viewModel.resumeUpload(task.id) }
            )
        }
    }
}

@Composable
fun FileUploadItem(task: UploadTask) {
    var progress by remember(task.id) { mutableStateOf(0f) }
    
    // 监听进度更新
    LaunchedEffect(task.id) {
        uploadManager.progressState.filter { it.taskId == task.id }
            .collect { progress = it.calculateProgress() }
    }
    
    Row {
        // 文件名显示
        Text(task.file.name)
        // 进度条
        LinearProgressIndicator(progress = progress)
        // 暂停/继续按钮
        Button(onClick = { if (task.isPaused) onResume() else onPause() }) {
            Text(if (task.isPaused) "▶" else "⏸")
        }
    }
}

// 使用DataStore保存断点信息
object UploadCache {
    private val Context.dataStore: DataStore<Preferences> by preferencesDataStore("upload_cache")
    
    suspend fun saveChunkState(context: Context, taskId: String, chunks: List<UploadTask.Chunk>) {
        context.dataStore.edit { preferences ->
            preferences[stringSetPreferencesKey(taskId)] = chunks.map {
                "${it.index}:${it.status}" // 存储分片状态
            }.toSet()
        }
    }
    
    suspend fun loadChunkState(context: Context, taskId: String): Map<Int, UploadTask.Status> {
        return context.dataStore.data.first()[stringSetPreferencesKey(taskId)]
            ?.associate { record ->
                record.split(":").let {
                    it[0].toInt() to UploadTask.Status.valueOf(it[1])
                }
            } ?: emptyMap()
    }
}

// 根据网络动态调整分片大小
fun calculateDynamicChunkSize(networkType: NetworkType): Int {
    return when(networkType) {
        NetworkType.WIFI -> 5 * 1024 * 1024  // 5MB
        NetworkType.ETHERNET -> 10 * 1024 * 1024 // 10MB
        else -> 512 * 1024 // 512KB
    }
}

// 带指数退避的重试机制
private suspend fun uploadChunkWithRetry(chunk: UploadTask.Chunk, maxRetries: Int = 3) {
    var currentRetry = 0
    var delayFactor = 1L
    
    while (currentRetry < maxRetries) {
        try {
            uploadChunk(chunk)
            return
        } catch (e: Exception) {
            if (++currentRetry == maxRetries) throw e
            delay(1000 * delayFactor) // 指数等待
            delayFactor *= 2
        }
    }
}

// 使用内存映射减少内存占用
private suspend fun uploadChunkWithMappedByteBuffer(file: File, chunk: Chunk) {
    val channel = FileInputStream(file).channel
    val buffer = channel.map(
        FileChannel.MapMode.READ_ONLY, 
        chunk.start, 
        chunk.end - chunk.start
    )
    
    // 上传buffer内容（避免创建临时字节数组）
    uploadBuffer(buffer)
}