android - 在继续之前等待几个流程完成

Question

我需要将图像上传到服务器，为此我不能使用其他库，而是将它(base64 编码)分成 block 并全部上传。

我正在为此使用 Kotlin 协程流，我目前所做的是进行第一次调用(返回流)以获取我需要在所有上传请求中附加的图像 ID

这是我用来上传图片的两个函数

fun submitImage(payload: Payload): Flow<String> {
    val request = requestBuilder.buildUploadImageRequest(payload)
    return client.execute(request)
        .serviceFlow({ response ->
            val imageId = response.body.id
            uploadImage(payload.imageBase64, imageId)
            imageId
        }, { response ->
            throw MyServerError("Error ${response.error}")
        })
}

private fun uploadImage(imageBase64: String, imageId: String) {
    val chunks = divideEncodedImageInChunksOfSize(imageBase64)
    var v = 1
    for (chunk in chunks) {
        val payload = generatePayload(imageId, v, chunk, false)
        submitImageChunk(payload)
        v++
    }
    val payload = generatePayload(imageId, v, "", true)
    submitImageChunk(payload)
}

private fun submitImageChunk(payload: JSONObject): Flow<Unit> {
    val request = requestBuilder.buildUploadImageChunkRequest(payload)
    return client.execute(request)
        .serviceFlow({ }, { response ->
            throw MyHttpError(response)
        })
}

我使用以下实用函数

// Extension function to handle Flows and their activation
internal fun MyHttpClient.execute(request: MyHttpRequest): Flow<MyHttpResponse> {
    return flow {
        val deferred = CompletableDeferred<MyHttpResponse>()
        executeHttp(request, object : MyHttpListener {
            override fun onSuccess(response: MyHttpResponse) {
                deferred.complete(response)
            }

            override fun onFailure(response: MyHttpResponse) {
                deferred.completeExceptionally(MyHttpError(response))
            }
        })
        emit(deferred.await())
    }
}

// Extension function to catch exceptions AND to check if the response body is null
internal fun <T> Flow<MyHttpResponse>.serviceFlow(
    onSuccess: (response: MyHttpResponse) -> T,
    onError: (response: MyHttpResponse) -> Unit
) = flatMapConcat { response ->
    flowOf(response)
        .map { res ->
            res.body?.let { it ->
                onSuccess(res)
            } ?: throw MyParseError("MyHttpResponse has a null body")
        }
        .catchException<JSONException, T> { e ->
            throw MyParseError("Parsing exception $e")
        }
}.catchException<MyHttpError, T> { e ->
    onError(e.response)
}

// Function leveraging OkHttpClient to make a HTTPRequest
internal fun executeHttp { ... }

我认为问题是由于函数 submitImage 在启动所有用于上传图像的子流后返回，但它没有等待所有子流完成。我不确定 Kotlin 协程对于这样的用例有什么构造，有人可以帮助我吗？

Answer 1

谢谢musafee让我朝着正确的方向前进。

最后的答案是我在 uploadImage 中创建了这些流功能，但我从来没有真正调用 collect在他们身上，因此他们保持未发射状态。

我选择的解决方案是将在那里创建的流列表返回给调用函数，然后从那里更改 submitImage 的返回类型来自 Flow<String> 的函数到 Flow<List<Flow<Unit>>> ，并从上层触发它们