在 Android 上使用 gRPC 处理文件下载

Question

我目前有一个正在发送视频文件块的 gRPC 服务器。我用 Kotlin 编写的 android 应用程序使用协程进行 UI 更新（在 Dispatchers.MAIN 上）和处理单向块流（在 Dispatchers.IO 上）。像下面这样：

GlobalScope.launch(Dispatchers.Main) {
   viewModel.downloadUpdated().accept(DOWNLOAD_STATE.DOWNLOADING) // MAKE PROGRESS BAR VISIBLE

      GlobalScope.launch(Dispatchers.IO) {
         stub.downloadVideo(request).forEach {
                file.appendBytes(
                    it.data.toByteArray()
                )
            }
      }.join()
      viewModel.downloadUpdated().accept(DOWNLOAD_STATE.FINISHED) // MAKE PROGRESS BAR DISAPPEAR
   } catch (exception: Exception) {
      viewModel.downloadUpdated().accept(DOWNLOAD_STATE.ERROR) // MAKE PROGRESS BAR DISAPPEAR
      screenNavigator.showError(exception) // SHOW DIALOG
   }
}

这很好用，但我想知道是否没有一种“更干净”的方式来处理下载。我已经知道 DownloadManager，但我觉得它只接受 HTTP 查询，所以我不能使用我的 gRPC 存根（我可能错了，如果是，请告诉我）。我还检查了 WorkManager，这是同样的问题，我不知道这是否是处理该案例的正确方法。

所以，这里有两个问题：

• 有没有办法以干净的方式处理 gRPC 查询，这意味着我现在可以在它开始、完成、失败时进行处理，并且我可以正确取消？
• 如果没有，有没有更好的方法来使用协程？

编辑

对于那些感兴趣的人，我相信我想出了一个虚拟算法，用于在更新进度条的同时进行下载（对改进开放）：

suspend fun downloadVideo(callback: suspend (currentBytesRead: Int) -> Unit) {
   println("download")

   stub.downloadVideo(request).forEach {
      val data = it.data.toByteArray()

      file.appendBytes(data)
      callback(x) // Where x is the percentage of download
   }
        
    println("downloaded")
}

class Fragment : CoroutineScope { //NOTE: The scope is the current Fragment
    private val job = Job()
    
    override val coroutineContext: CoroutineContext
        get() = job
    
    fun onCancel() {
        if (job.isActive) {
            job.cancel()
        }
    }
    
    private suspend fun updateLoadingBar(currentBytesRead: Int) {
        println(currentBytesRead)
    }
    
    fun onDownload() {
     
        launch(Dispatchers.IO) {
            downloadVideo { currentBytes ->
                withContext(Dispatchers.Main) {
                        
                    updateLoadingBar(currentBytes)
                    
                    if (job.isCancelled)
                        println("cancelled !")
                }
            }
        }
    }
}

更多信息，请查看：Introduction to coroutines

编辑 2

正如 cmets 中所建议的，我们实际上可以使用 Flows 来处理这个问题，它会给出如下结果：

suspend fun foo(): Flow<Int> = flow { 
   println("download")
   stub.downloadVideo(request).forEach {
      val data = it.data.toByteArray()

      file.appendBytes(data)
      emit(x) // Where x is the percentage of download
   }
   println("downloaded")
}

class Fragment : CoroutineScope {
    private val job = Job()
    
    override val coroutineContext: CoroutineContext
        get() = job
    
    fun onCancel() {
        if (job.isActive) {
            job.cancel()
        }
    }
    
    private suspend fun updateLoadingBar(currentBytesRead: Int) {
        println(currentBytesRead)
    }
    
    fun onDownload() {
     
        launch(Dispatchers.IO) {
            withContext(Dispatchers.Main) {
                foo()
                    .onCompletion { cause -> println("Flow completed with $cause") }
                    .catch { e -> println("Caught $e") }
                    .collect { current -> 
                        if (job.isCancelled)
                            return@collect

                        updateLoadingBar(current)
                }
            }
        }
    }
}

Answer 1

gRPC 可以是很多东西，所以在这方面你的问题还不清楚。最重要的是，它可以完全异步和基于回调，这意味着它可以变成一个Flow，你可以在主线程上收集它。但是，文件写入是阻塞的。

您的代码似乎会在后台启动下载后立即发送FINISHED 信号。您应该将launch(IO) 替换为withContext(IO)。